From 3d0260d2b4df42fbb28f7ef42ca77516417f2d0b Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Thu, 24 Sep 2026 21:34:41 -0400 Subject: [PATCH] feat(daemon): store image payloads out of transcript rows Every turn decoded and validated every transcript row, so a session with screenshots held all of them for the whole turn (20 MB for one session). - images live once in a content-addressed `images` table; rows keep the hash and metadata, and loads attach a reader instead of the bytes - request encoders emit placeholders; the transport streams the body and reads each image just before sending it, one at a time, with an exact content-length. Request bytes are unchanged, so prompt caching holds - rolling fingerprints hash images by content; a hash saved before this is checked once in its old form and rewritten, so summaries survive - a one-time startup migration moves existing inline images, after a VACUUM INTO backup under $ALBEDO_HOME/backups - deleting a session releases images nothing else references - the cell journal keeps images in the row shape --- src/albedo/daemon/albedo_conversation.erl | 25 +- src/albedo/daemon/albedo_images.erl | 144 ++++++++++ src/albedo/daemon/context_snapshot.gleam | 2 +- src/albedo/daemon/conversation.gleam | 52 ++-- src/albedo/daemon/events.gleam | 2 +- src/albedo/daemon/history.gleam | 16 +- src/albedo/daemon/images.gleam | 254 ++++++++++++++++++ src/albedo/daemon/server.gleam | 18 ++ src/albedo/harness/compaction.gleam | 8 +- .../harness/extensions/antigravity/wire.gleam | 7 +- src/albedo/harness/extensions/lcm/tools.gleam | 2 +- .../extensions/python/albedo_native.erl | 13 +- .../extensions/rolling/albedo_rolling.erl | 16 +- .../extensions/rolling/extension.gleam | 81 +++++- src/albedo/harness/loop.gleam | 2 +- src/albedo/openai_api/albedo_openai_json.erl | 17 +- .../openai_api/albedo_openai_transport.erl | 82 +++++- src/albedo/openai_api/request.gleam | 6 +- src/albedo/openai_api/types.gleam | 34 ++- .../albedo_image_store_test_support.erl | 14 + test/daemon/image_store_test.gleam | 239 ++++++++++++++++ test/daemon/image_test.gleam | 28 +- test/harness/tool_image_test.gleam | 16 +- 23 files changed, 1001 insertions(+), 77 deletions(-) create mode 100644 src/albedo/daemon/albedo_images.erl create mode 100644 src/albedo/daemon/images.gleam create mode 100644 test/daemon/albedo_image_store_test_support.erl create mode 100644 test/daemon/image_store_test.gleam diff --git a/src/albedo/daemon/albedo_conversation.erl b/src/albedo/daemon/albedo_conversation.erl index 1713304..07cd5e7 100644 --- a/src/albedo/daemon/albedo_conversation.erl +++ b/src/albedo/daemon/albedo_conversation.erl @@ -1,15 +1,19 @@ -module(albedo_conversation). --export([pack/1,unpack/1,unpack_trace/1,pack_list/1,unpack_list/1]). -pack(Input) -> term_to_binary({1,Input}). -unpack(Bytes) -> +-export([pack/1,unpack/2,unpack_trace/1,pack_list/1,unpack_list/2,row_atoms/0]). +pack(Input) -> term_to_binary({1,albedo_images:pack(Input)}). + +%% binary_to_term/2 with `safe` rejects atoms that do not exist yet; a stored +%% image reference is decoded here, so this module keeps its atom alive. +row_atoms() -> [stored_data]. + +%% Read fetches a stored image payload by hash (see albedo_images). +unpack(Bytes,Read) -> try binary_to_term(Bytes,[safe]) of {1,{Tag,Text}=Input} when (Tag=:=user orelse Tag=:=assistant), is_binary(Text) -> {ok,Input}; - {1,{user_image,Text,Image}=Input} when is_binary(Text) -> - case albedo_image:valid(Image) of true -> {ok,Input}; false -> {error,nil} end; + {1,{user_image,Text,_}=Input} when is_binary(Text) -> attached(Input,Read); %% Tool outputs saved before tool images carry no image list. {1,{tool_output,Id,Text}} when is_binary(Id),is_binary(Text) -> {ok,{tool_output,Id,Text,[]}}; - {1,{tool_output,Id,Text,Images}=Input} when is_binary(Id),is_binary(Text),is_list(Images) -> - case lists:all(fun albedo_image:valid/1,Images) of true -> {ok,Input}; false -> {error,nil} end; + {1,{tool_output,Id,Text,Images}=Input} when is_binary(Id),is_binary(Text),is_list(Images) -> attached(Input,Read); {1,{replay,{replay_item,Protocol,Value}}=Input} when is_map(Value), (Protocol=:=responses orelse Protocol=:=chat_completions) -> case {Protocol,Value} of {responses,#{<<"type">> := Type}} when is_binary(Type) -> {ok,Input}; @@ -19,6 +23,9 @@ unpack(Bytes) -> _ -> {error,nil} catch _:_ -> {error,nil} end. +attached(Input,Read) -> + case albedo_images:attach(Input,Read) of {ok,_}=Ok -> Ok; error -> {error,nil} end. + unpack_trace(Bytes) -> try binary_to_term(Bytes,[safe]) of {1,#{<<"activities">> := A, <<"changes">> := C}=Trace} when is_list(A),is_list(C) -> {ok,Trace}; _ -> {error,nil} @@ -26,9 +33,9 @@ catch _:_ -> {error,nil} end. %% A pinned prompt's context inputs, each packed like a transcript entry. pack_list(Inputs) -> term_to_binary({1,[pack(I) || I <- Inputs]}). -unpack_list(Bytes) -> try binary_to_term(Bytes,[safe]) of +unpack_list(Bytes,Read) -> try binary_to_term(Bytes,[safe]) of {1,Packed} when is_list(Packed) -> - Inputs = [unpack(P) || P <- Packed], + Inputs = [unpack(P,Read) || P <- Packed], case lists:all(fun({ok,_}) -> true; (_) -> false end, Inputs) of true -> {ok,[I || {ok,I} <- Inputs]}; false -> {error,nil} diff --git a/src/albedo/daemon/albedo_images.erl b/src/albedo/daemon/albedo_images.erl new file mode 100644 index 0000000..0809b37 --- /dev/null +++ b/src/albedo/daemon/albedo_images.erl @@ -0,0 +1,144 @@ +%% Image payloads live in the `images` table, keyed by the SHA-256 of their +%% base64 text; transcript rows keep only that hash and the image metadata. +%% +%% In memory an image is {image, Mime, Data, Width, Height, Bytes} where Data is +%% {inline_data, Base64} or {stored_data, Hash, Size, Read}, Read being a fun/0 +%% that fetches the payload (see types.ImageData). A packed row stores +%% {stored_data, Hash, Size} (no fun), or the legacy bare Base64 binary. +-module(albedo_images). +-export([externalize/2, attach/2, pack/1, pack_image/1, load/2, canonical/1, legacy/1, hashes/1, migrate/2, clean/1, ensure_dir/1]). + +-define(MAX_DATA_BYTES, 6990508). + +%% Moves the inline images of one input into Blobs [{Hash, Base64}], returning +%% the input with stored references. Read is fun(Hash) -> {ok, Base64} | {error, nil}. +%% Payloads that would need JSON escaping stay inline (a request writes a +%% stored payload verbatim between quotes). +externalize({user_image, Text, Image}, Read) -> + {Stored, Blobs} = store(Image, Read), + {{user_image, Text, Stored}, Blobs}; +externalize({tool_output, Id, Text, Images}, Read) -> + Pairs = [store(Image, Read) || Image <- Images], + {{tool_output, Id, Text, [S || {S, _} <- Pairs]}, lists:append([B || {_, B} <- Pairs])}; +externalize(Input, _) -> {Input, []}. + +store({image, Mime, {inline_data, Data}, W, H, Bytes} = Image, Read) -> + case byte_size(Data) =:= 4 * ((Bytes + 2) div 3) andalso clean(Data) of + true -> + Hash = hash(Data), + {{image, Mime, stored(Hash, byte_size(Data), Read), W, H, Bytes}, [{Hash, Data}]}; + false -> {Image, []} + end; +store(Image, _) -> {Image, []}. + +stored(Hash, Size, Read) -> {stored_data, Hash, Size, fun() -> Read(Hash) end}. + +hash(Data) -> binary:encode_hex(crypto:hash(sha256, Data), lowercase). + +%% A decoded row's images with their readers attached; packed refs that fail +%% the shape check make the row invalid. +attach({user_image, Text, Image}, Read) -> + case load(Image, Read) of + {ok, Loaded} -> {ok, {user_image, Text, Loaded}}; + error -> error + end; +attach({tool_output, Id, Text, Images}, Read) -> + Loaded = [load(Image, Read) || Image <- Images], + case lists:all(fun({ok, _}) -> true; (_) -> false end, Loaded) of + true -> {ok, {tool_output, Id, Text, [I || {ok, I} <- Loaded]}}; + false -> error + end; +attach(Input, _) -> {ok, Input}. + +load({image, Mime, {stored_data, Hash, Size}, W, H, Bytes}, Read) + when is_binary(Mime), is_binary(Hash), byte_size(Hash) =:= 64, + is_integer(Size), Size > 0, Size =< ?MAX_DATA_BYTES, + is_integer(W), is_integer(H), is_integer(Bytes), Bytes > 0 -> + %% A stored payload was validated as canonical base64 on insert. + case Size =:= 4 * ((Bytes + 2) div 3) of + true -> {ok, {image, Mime, stored(Hash, Size, Read), W, H, Bytes}}; + false -> error + end; +load({image, Mime, Data, W, H, Bytes} = Legacy, _) when is_binary(Data) -> + case albedo_image:valid(Legacy) of + true -> {ok, {image, Mime, {inline_data, Data}, W, H, Bytes}}; + false -> error + end; +load(_, _) -> error. + +%% The row form of an input: readers dropped, inline payloads in the legacy +%% bare-binary shape so rows that were never externalized read as before. +pack({user_image, Text, Image}) -> {user_image, Text, pack_image(Image)}; +pack({tool_output, Id, Text, Images}) -> {tool_output, Id, Text, [pack_image(I) || I <- Images]}; +pack(Input) -> Input. + +pack_image({image, Mime, {inline_data, Data}, W, H, Bytes}) -> {image, Mime, Data, W, H, Bytes}; +pack_image({image, Mime, {stored_data, Hash, Size, _}, W, H, Bytes}) -> + {image, Mime, {stored_data, Hash, Size}, W, H, Bytes}; +pack_image(Image) -> Image. + +%% Any term with every image's payload replaced by its content hash, so equal +%% content fingerprints equally whether it is inline or stored. Terms without +%% images are returned unchanged. +canonical({image, Mime, {inline_data, Data}, W, H, Bytes}) when is_binary(Mime) -> + {image, Mime, {sha256, hash(Data)}, W, H, Bytes}; +canonical({image, Mime, {stored_data, Hash, _, _}, W, H, Bytes}) when is_binary(Mime) -> + {image, Mime, {sha256, Hash}, W, H, Bytes}; +canonical(T) when is_tuple(T) -> list_to_tuple(canonical(tuple_to_list(T))); +canonical([H | T]) -> [canonical(H) | canonical(T)]; +canonical(M) when is_map(M) -> maps:from_list(canonical(maps:to_list(M))); +canonical(Other) -> Other. + +%% The term as it was before images were stored: payloads inline as bare +%% binaries. Reads every stored payload; {error, nil} if one is missing. +legacy(Term) -> + try {ok, legacy_term(Term)} + catch throw:missing -> {error, nil} + end. + +legacy_term({image, Mime, {inline_data, Data}, W, H, Bytes}) when is_binary(Mime) -> + {image, Mime, Data, W, H, Bytes}; +legacy_term({image, Mime, {stored_data, _, _, Read}, W, H, Bytes}) when is_binary(Mime) -> + case Read() of + {ok, Data} -> {image, Mime, Data, W, H, Bytes}; + _ -> throw(missing) + end; +legacy_term(T) when is_tuple(T) -> list_to_tuple(legacy_term(tuple_to_list(T))); +legacy_term([H | T]) -> [legacy_term(H) | legacy_term(T)]; +legacy_term(M) when is_map(M) -> maps:from_list(legacy_term(maps:to_list(M))); +legacy_term(Other) -> Other. + +%% Hashes a packed row references, without attaching or validating. +hashes(Payload) -> + try binary_to_term(Payload, [safe]) of + {1, {user_image, _, Image}} -> ref(Image); + {1, {tool_output, _, _, Images}} when is_list(Images) -> lists:append([ref(I) || I <- Images]); + _ -> [] + catch _:_ -> [] + end. + +ref({image, _, {stored_data, Hash, _}, _, _, _}) -> [Hash]; +ref(_) -> []. + +%% One legacy row, rewritten: {rewrite, Payload, Blobs} when it held inline images +%% that moved to the image table, keep otherwise (including rows that do not +%% decode, which are left for the loader to report). +migrate(Payload, Read) -> + case albedo_conversation:unpack(Payload, Read) of + {ok, Input} -> + case externalize(Input, Read) of + {_, []} -> keep; + {Stored, Blobs} -> {rewrite, albedo_conversation:pack(Stored), Blobs} + end; + _ -> keep + end. + +%% Canonical base64 needs no JSON escaping; anything else stays inline. +clean(<>) + when (C >= $A andalso C =< $Z); (C >= $a andalso C =< $z); + (C >= $0 andalso C =< $9); C =:= $+; C =:= $/; C =:= $= -> + clean(Rest); +clean(<<>>) -> true; +clean(_) -> false. + +ensure_dir(Path) -> _ = filelib:ensure_dir(Path), nil. diff --git a/src/albedo/daemon/context_snapshot.gleam b/src/albedo/daemon/context_snapshot.gleam index de5ac8c..5081668 100644 --- a/src/albedo/daemon/context_snapshot.gleam +++ b/src/albedo/daemon/context_snapshot.gleam @@ -357,7 +357,7 @@ fn render_input(input: types.Input) -> String { } fn image_label(image: types.Image) -> String { - let #(mime, _, width, height, bytes) = types.image_parts(image) + let #(mime, width, height, bytes) = types.image_meta(image) "[image · " <> mime <> " · " diff --git a/src/albedo/daemon/conversation.gleam b/src/albedo/daemon/conversation.gleam index 3ae899a..fb6d064 100644 --- a/src/albedo/daemon/conversation.gleam +++ b/src/albedo/daemon/conversation.gleam @@ -1,4 +1,5 @@ import albedo/daemon/events +import albedo/daemon/images import albedo/daemon/note import albedo/daemon/store import albedo/daemon/transcript @@ -65,7 +66,8 @@ pub fn initialise(store: store.Store) -> Result(Nil, String) { store.query(store, fn(db) { use _ <- result.try( sqlight.exec( - "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER,effort TEXT); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);", + "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER,effort TEXT); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);" + <> images.schema, db, ) |> result.map_error(fn(e) { e.message }), @@ -232,7 +234,7 @@ fn recover_sessions(db) -> Result(Nil, String) { |> result.map_error(fn(e) { e.message }) |> result.map(fn(rows) { rows - |> list.filter_map(unpack) + |> list.filter_map(unpack(_, unread)) |> latest_user |> title_or_default }) @@ -333,6 +335,7 @@ pub fn delete(store: store.Store, id: String) -> Result(Nil, String) { |> result.map_error(fn(e) { e.message }), ) let deleted = { + use hashes <- result.try(images.session_hashes(db, id)) use tables <- result.try( sqlight.query( "SELECT name FROM sqlite_master WHERE type='table'", @@ -382,14 +385,16 @@ pub fn delete(store: store.Store, id: String) -> Result(Nil, String) { }, ), ) - sqlight.query( - "DELETE FROM sessions WHERE id=?", - db, - [sqlight.text(id)], - decode.dynamic, + use _ <- result.try( + sqlight.query( + "DELETE FROM sessions WHERE id=?", + db, + [sqlight.text(id)], + decode.dynamic, + ) + |> result.map_error(fn(e) { e.message }), ) - |> result.replace(Nil) - |> result.map_error(fn(e) { e.message }) + images.release(db, hashes) } case deleted { Ok(_) -> @@ -547,6 +552,7 @@ fn load_source_pages( upper: Int, pages: List(List(transcript.SourcedEntry)), ) -> Result(List(transcript.SourcedEntry), String) { + let read = images.reader(store) let page = store.query(store, fn(db) { use rows <- result.try( @@ -572,7 +578,7 @@ fn load_source_pages( use entries <- result.try( list.try_map(rows, fn(row) { use input <- result.try( - unpack(row.1) + unpack(row.1, read) |> result.replace_error("invalid saved transcript item"), ) Ok(transcript.SourcedEntry( @@ -686,6 +692,7 @@ fn read_before( upper: Int, limit: Int, ) -> Result(#(List(transcript.SourcedEntry), Bool), String) { + let read = images.reader(store) store.query(store, fn(db) { use rows <- result.try( sqlight.query( @@ -705,7 +712,7 @@ fn read_before( use entries <- result.try( list.try_map(list.reverse(rows), fn(row) { use input <- result.try( - unpack(row.1) + unpack(row.1, read) |> result.replace_error("invalid saved transcript item"), ) Ok(transcript.SourcedEntry( @@ -741,6 +748,7 @@ pub fn source( reference: transcript.SourceRef, ) -> Result(Option(transcript.Entry), String) { let transcript.SourceRef(session, seq) = reference + let read = images.reader(store) store.query(store, fn(db) { use rows <- result.try( sqlight.query( @@ -759,7 +767,7 @@ pub fn source( case rows { [] -> Ok(None) [row] -> - unpack(row.0) + unpack(row.0, read) |> result.replace_error("invalid saved transcript item") |> result.map(fn(input) { Some(transcript.Entry(input, row.1, row.2)) }) _ -> Error("duplicate transcript source reference") @@ -864,7 +872,7 @@ pub fn prompt_pin( case rows { [#(Some(instructions), Some(context))] -> Ok( - unpack_list(context) + unpack_list(context, images.reader(store)) |> result.map(PinnedPrompt(instructions, _)) |> option.from_result, ) @@ -908,6 +916,7 @@ pub fn commit_from( provider: Option(String), ) -> Result(Int, String) { let timestamp = usage.now() + let read = images.reader(store) store.query(store, fn(db) { use _ <- result.try( sqlight.exec("BEGIN IMMEDIATE", db) @@ -920,6 +929,7 @@ pub fn commit_from( } use _ <- result.try( list.try_each(inputs, fn(input) { + use input <- result.try(images.externalize(db, input, read)) sqlight.query( "INSERT INTO transcript(session,payload,timestamp,provider) VALUES(?,?,?,?)", db, @@ -1066,13 +1076,25 @@ pub fn record_usage( fn pack(input: types.Input) -> BitArray @external(erlang, "albedo_conversation", "unpack") -fn unpack(bytes: BitArray) -> Result(types.Input, Nil) +fn unpack( + bytes: BitArray, + read: fn(String) -> Result(String, Nil), +) -> Result(types.Input, Nil) + +/// For rows read only for their text, inside the store (which a real reader +/// would call back into). +fn unread(_hash: String) -> Result(String, Nil) { + Error(Nil) +} @external(erlang, "albedo_conversation", "pack_list") fn pack_list(inputs: List(types.Input)) -> BitArray @external(erlang, "albedo_conversation", "unpack_list") -fn unpack_list(bytes: BitArray) -> Result(List(types.Input), Nil) +fn unpack_list( + bytes: BitArray, + read: fn(String) -> Result(String, Nil), +) -> Result(List(types.Input), Nil) pub fn set_effort( store: store.Store, diff --git a/src/albedo/daemon/events.gleam b/src/albedo/daemon/events.gleam index fb1bb3f..f7e93ea 100644 --- a/src/albedo/daemon/events.gleam +++ b/src/albedo/daemon/events.gleam @@ -131,7 +131,7 @@ pub fn user_image( /// What clients show of an image: never its payload. fn image_metadata(image: types.Image) -> json.Json { - let #(mime_type, _, width, height, bytes) = types.image_parts(image) + let #(mime_type, width, height, bytes) = types.image_meta(image) json.object([ #("mimeType", json.string(mime_type)), #("width", json.int(width)), diff --git a/src/albedo/daemon/history.gleam b/src/albedo/daemon/history.gleam index 149fca9..12763bf 100644 --- a/src/albedo/daemon/history.gleam +++ b/src/albedo/daemon/history.gleam @@ -281,7 +281,8 @@ fn row_decoder() { fn decode_row(row) -> Result(Row, String) { let #(seq, payload, timestamp, provider) = row use input <- result.try( - unpack(payload) |> result.replace_error("invalid saved transcript item"), + unpack(payload, unread) + |> result.replace_error("invalid saved transcript item"), ) Ok(Row(seq, payload, input, timestamp, provider)) } @@ -301,7 +302,7 @@ fn row_item(row: Row) -> Item { } fn image_label(image: types.Image) -> String { - let #(mime, _, width, height, _) = types.image_parts(image) + let #(mime, width, height, _) = types.image_meta(image) " [" <> mime <> " " @@ -486,4 +487,13 @@ fn append_incomplete_results( fn pack(input: types.Input) -> BitArray @external(erlang, "albedo_conversation", "unpack") -fn unpack(bytes: BitArray) -> Result(types.Input, Nil) +fn unpack( + bytes: BitArray, + read: fn(String) -> Result(String, Nil), +) -> Result(types.Input, Nil) + +/// History rows are previews and fork bookkeeping (a fork copies payload bytes, +/// references included); none is sent to a model, so none reads an image. +fn unread(_hash: String) -> Result(String, Nil) { + Error(Nil) +} diff --git a/src/albedo/daemon/images.gleam b/src/albedo/daemon/images.gleam new file mode 100644 index 0000000..e9518db --- /dev/null +++ b/src/albedo/daemon/images.gleam @@ -0,0 +1,254 @@ +//// Image payloads stored once by content hash, out of transcript rows. +//// +//// A transcript row keeps an image's hash and metadata; its base64 text lives +//// in `images`. Loading a session therefore reads references, and a payload is +//// read only while a request body is written (albedo_openai_transport.erl). +//// Rows are content-addressed, so forks and repeated screenshots share one copy. + +import albedo/daemon/store +import albedo/openai_api/types +import gleam/dynamic/decode +import gleam/list +import gleam/result +import sqlight + +pub const schema = "CREATE TABLE IF NOT EXISTS images(hash TEXT PRIMARY KEY,data TEXT NOT NULL); CREATE TABLE IF NOT EXISTS migrations(name TEXT PRIMARY KEY,applied_at INTEGER NOT NULL);" + +/// Fetches a stored payload. Never call the result from inside `store.query`: +/// the read is itself a store query, and the store runs one at a time. +pub fn reader(ledger: store.Store) -> fn(String) -> Result(String, Nil) { + fn(hash) { store.query(ledger, fn(db) { read(db, hash) }) } +} + +fn read(db: sqlight.Connection, hash: String) -> Result(String, Nil) { + case + sqlight.query( + "SELECT data FROM images WHERE hash=?", + db, + [sqlight.text(hash)], + decode.field(0, decode.string, decode.success), + ) + { + Ok([data]) -> Ok(data) + _ -> Error(Nil) + } +} + +/// The input with its inline images replaced by stored references, after +/// writing their payloads. Runs inside the caller's transaction. +pub fn externalize( + db: sqlight.Connection, + input: types.Input, + read: fn(String) -> Result(String, Nil), +) -> Result(types.Input, String) { + let #(stored, blobs) = split(input, read) + use _ <- result.try(insert(db, blobs)) + Ok(stored) +} + +fn insert( + db: sqlight.Connection, + blobs: List(#(String, String)), +) -> Result(Nil, String) { + list.try_each(blobs, fn(blob) { + sqlight.query( + "INSERT OR IGNORE INTO images(hash,data) VALUES(?,?)", + db, + [sqlight.text(blob.0), sqlight.text(blob.1)], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) + }) +} + +/// Hashes the given session's rows reference, read before those rows are +/// deleted so `release` can drop payloads nothing else needs. +pub fn session_hashes( + db: sqlight.Connection, + session: String, +) -> Result(List(String), String) { + sqlight.query( + "SELECT payload FROM transcript WHERE session=? AND instr(payload,CAST('stored_data' AS BLOB))>0", + db, + [sqlight.text(session)], + decode.field(0, decode.bit_array, decode.success), + ) + |> result.map(fn(rows) { list.flat_map(rows, hashes) |> list.unique }) + |> result.map_error(fn(e) { e.message }) +} + +/// Deletes payloads no transcript row or pinned prompt still names. A packed +/// reference holds the hash text verbatim, so a byte search finds every user. +pub fn release( + db: sqlight.Connection, + candidates: List(String), +) -> Result(Nil, String) { + list.try_each(candidates, fn(hash) { + sqlight.query( + "DELETE FROM images WHERE hash=?1 AND NOT EXISTS(SELECT 1 FROM transcript WHERE instr(payload,CAST(?1 AS BLOB))>0) AND NOT EXISTS(SELECT 1 FROM sessions WHERE pinned_context IS NOT NULL AND instr(pinned_context,CAST(?1 AS BLOB))>0)", + db, + [sqlight.text(hash)], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) + }) +} + +/// Moves inline images in existing transcript rows into `images`, once per +/// database. The database is first copied to `backup` with VACUUM INTO. Rows +/// are rewritten a page at a time, each page in its own transaction, so an +/// interrupted run resumes where it stopped (a rewritten row has nothing inline). +pub fn migrate(ledger: store.Store, backup: String) -> Result(Int, String) { + let read = reader(ledger) + use applied <- result.try( + store.query(ledger, fn(db) { + sqlight.query( + "SELECT 1 FROM migrations WHERE name='image_store'", + db, + [], + decode.dynamic, + ) + |> result.map(fn(rows) { rows != [] }) + |> result.map_error(fn(e) { e.message }) + }), + ) + case applied { + True -> Ok(0) + False -> { + use pending <- result.try( + store.query(ledger, fn(db) { + sqlight.query( + "SELECT count(*) FROM transcript WHERE instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0", + db, + [], + decode.field(0, decode.int, decode.success), + ) + |> result.map_error(fn(e) { e.message }) + }), + ) + use _ <- result.try(case pending { + [0] -> Ok(Nil) + _ -> { + ensure_dir(backup) + store.query(ledger, fn(db) { + sqlight.query( + "VACUUM INTO ?", + db, + [sqlight.text(backup)], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { + "image store backup failed: " <> e.message + }) + }) + } + }) + use moved <- result.try(migrate_pages(ledger, read, -1, 0)) + use _ <- result.try( + store.query(ledger, fn(db) { + sqlight.query( + "INSERT OR IGNORE INTO migrations(name,applied_at) VALUES('image_store',unixepoch())", + db, + [], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) + }), + ) + Ok(moved) + } + } +} + +/// Rows per migration transaction; a page of screenshot rows is tens of MB. +const migrate_page_rows = 16 + +fn migrate_pages( + ledger: store.Store, + read: fn(String) -> Result(String, Nil), + after: Int, + moved: Int, +) -> Result(Int, String) { + let page = + store.query(ledger, fn(db) { + use rows <- result.try( + sqlight.query( + "SELECT seq,payload FROM transcript WHERE seq>? AND (instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0) ORDER BY seq LIMIT ?", + db, + [sqlight.int(after), sqlight.int(migrate_page_rows)], + { + use seq <- decode.field(0, decode.int) + use payload <- decode.field(1, decode.bit_array) + decode.success(#(seq, payload)) + }, + ) + |> result.map_error(fn(e) { e.message }), + ) + use _ <- result.try( + sqlight.exec("BEGIN IMMEDIATE", db) + |> result.map_error(fn(e) { e.message }), + ) + let written = + list.try_fold(rows, 0, fn(count, row) { + case migrate_row(row.1, read) { + Keep -> Ok(count) + Rewrite(payload, blobs) -> { + use _ <- result.try(insert(db, blobs)) + sqlight.query( + "UPDATE transcript SET payload=? WHERE seq=?", + db, + [sqlight.blob(payload), sqlight.int(row.0)], + decode.dynamic, + ) + |> result.replace(count + 1) + |> result.map_error(fn(e) { e.message }) + } + } + }) + case written { + Ok(count) -> { + use _ <- result.try( + sqlight.exec("COMMIT", db) + |> result.map_error(fn(e) { e.message }), + ) + Ok(#(count, list.last(rows) |> result.map(fn(row) { row.0 }))) + } + Error(e) -> { + let _ = sqlight.exec("ROLLBACK", db) + Error(e) + } + } + }) + case page { + Error(e) -> Error("image store migration failed: " <> e) + Ok(#(count, Ok(last))) -> migrate_pages(ledger, read, last, moved + count) + Ok(#(count, Error(_))) -> Ok(moved + count) + } +} + +type Migration { + Keep + Rewrite(payload: BitArray, blobs: List(#(String, String))) +} + +@external(erlang, "albedo_images", "externalize") +fn split( + input: types.Input, + read: fn(String) -> Result(String, Nil), +) -> #(types.Input, List(#(String, String))) + +@external(erlang, "albedo_images", "hashes") +fn hashes(payload: BitArray) -> List(String) + +@external(erlang, "albedo_images", "migrate") +fn migrate_row( + payload: BitArray, + read: fn(String) -> Result(String, Nil), +) -> Migration + +@external(erlang, "albedo_images", "ensure_dir") +fn ensure_dir(path: String) -> Nil diff --git a/src/albedo/daemon/server.gleam b/src/albedo/daemon/server.gleam index c7fa25e..72d1678 100644 --- a/src/albedo/daemon/server.gleam +++ b/src/albedo/daemon/server.gleam @@ -2,9 +2,11 @@ import albedo/daemon/configuration import albedo/daemon/conversation import albedo/daemon/history import albedo/daemon/image +import albedo/daemon/images import albedo/daemon/reaper import albedo/daemon/session import albedo/daemon/store +import albedo/daemon/usage import albedo/harness/command import albedo/harness/extension import albedo/harness/extensions/schedule/ledger as schedule @@ -86,6 +88,22 @@ pub fn start(config: Config, port: Int) -> Result(Int, String) { |> result.replace_error("could not start runtime"), ) use _ <- result.try(conversation.initialise(runtime.ledger(host))) + use moved <- result.try(images.migrate( + runtime.ledger(host), + config.home + <> "/backups/albedo-before-image-store-" + <> int.to_string(usage.now()) + <> ".sqlite", + )) + case moved { + 0 -> Nil + rows -> + io.println( + "image store: moved images out of " + <> int.to_string(rows) + <> " transcript rows", + ) + } use _ <- result.try(case configuration.legacy(config.home) { Ok(provider) -> conversation.assign_provider(runtime.ledger(host), provider.name) diff --git a/src/albedo/harness/compaction.gleam b/src/albedo/harness/compaction.gleam index 5ba3b49..1d454e2 100644 --- a/src/albedo/harness/compaction.gleam +++ b/src/albedo/harness/compaction.gleam @@ -80,15 +80,13 @@ pub fn input_bytes(input: types.Input) -> Int { case input { types.User(text) | types.Assistant(text) -> string.byte_size(text) types.UserImage(text, image) -> { - let #(_, data, _, _, _) = types.image_parts(image) - string.byte_size(text) + string.byte_size(data) + string.byte_size(text) + types.image_size(image) } types.ToolOutput(id, output, images) -> string.byte_size(id) + string.byte_size(output) + list.fold(images, 0, fn(total, image) { - let #(_, data, _, _, _) = types.image_parts(image) - total + string.byte_size(data) + total + types.image_size(image) }) types.Replay(item) -> types.replay_json(item) |> json.to_string |> string.byte_size @@ -117,7 +115,7 @@ pub fn estimate_input(input: types.Input) -> Int { } fn estimate_image(image: types.Image) -> Int { - let #(_, _, width, height, _) = types.image_parts(image) + let #(_, width, height, _) = types.image_meta(image) // Provider-neutral approximation based on 512px vision tiles. Providers // may tokenize images differently; this is only the rolling trigger signal. 85 + 170 * ceiling_div(width, 512) * ceiling_div(height, 512) diff --git a/src/albedo/harness/extensions/antigravity/wire.gleam b/src/albedo/harness/extensions/antigravity/wire.gleam index 1cc7380..db7f57a 100644 --- a/src/albedo/harness/extensions/antigravity/wire.gleam +++ b/src/albedo/harness/extensions/antigravity/wire.gleam @@ -304,13 +304,13 @@ fn text_part(text: String) -> Json { } fn inline(image: types.Image) -> Json { - let #(mime_type, data, _, _, _) = types.image_parts(image) + let #(mime_type, _, _, _) = types.image_meta(image) json.object([ #( "inlineData", json.object([ #("mimeType", json.string(mime_type)), - #("data", json.string(data)), + #("data", base64_string(types.image_data(image))), ]), ), ]) @@ -487,6 +487,9 @@ fn optional( } } +@external(erlang, "albedo_openai_json", "base64_string") +fn base64_string(data: types.ImageData) -> Json + @external(erlang, "albedo_antigravity", "encode") pub fn encode_value(value: Dynamic) -> Json diff --git a/src/albedo/harness/extensions/lcm/tools.gleam b/src/albedo/harness/extensions/lcm/tools.gleam index a460fa9..ee86a4e 100644 --- a/src/albedo/harness/extensions/lcm/tools.gleam +++ b/src/albedo/harness/extensions/lcm/tools.gleam @@ -413,7 +413,7 @@ fn source_text(input: types.Input) -> String { } fn image_description(image: types.Image) -> String { - let #(mime, _, width, height, _) = types.image_parts(image) + let #(mime, width, height, _) = types.image_meta(image) mime <> " " <> int.to_string(width) <> "x" <> int.to_string(height) } diff --git a/src/albedo/harness/extensions/python/albedo_native.erl b/src/albedo/harness/extensions/python/albedo_native.erl index 8bdcd6b..52d35ce 100644 --- a/src/albedo/harness/extensions/python/albedo_native.erl +++ b/src/albedo/harness/extensions/python/albedo_native.erl @@ -1,6 +1,10 @@ -module(albedo_native). -export([new_id/0, pack_cell/1, unpack_cell/1]). new_id() -> binary:encode_hex(crypto:strong_rand_bytes(16), lowercase). +%% Images are journaled in the transcript row shape (albedo_images:pack_image/1). +pack_cell({ok, {outcome, Id, Status, Output, Value, Truncated, Images, Errors}}) -> + Packed = [albedo_images:pack_image(I) || I <- Images], + term_to_binary({1, {ok, {outcome, Id, Status, Output, Value, Truncated, Packed, Errors}}}); pack_cell(Value) -> term_to_binary({1, Value}). unpack_cell(Binary) -> try binary_to_term(Binary, [safe]) of @@ -15,12 +19,15 @@ unpack_cell(Binary) -> catch _:_ -> {error, nil} end. -unpack_outcome({outcome, Id, Status, Output, Value, Truncated, Images, Errors} = Outcome) +unpack_outcome({outcome, Id, Status, Output, Value, Truncated, Images, Errors}) when is_binary(Id), is_binary(Output), is_binary(Value), is_boolean(Truncated), is_list(Images), is_list(Errors), (Status =:= succeeded orelse Status =:= failed orelse Status =:= interrupted) -> - case lists:all(fun albedo_image:valid/1, Images) andalso lists:all(fun erlang:is_binary/1, Errors) of - true -> {ok, {ok, Outcome}}; + %% A journaled image is always inline, so nothing here reads a payload. + Loaded = [albedo_images:load(I, fun(_) -> {error, nil} end) || I <- Images], + case lists:all(fun({ok, _}) -> true; (_) -> false end, Loaded) + andalso lists:all(fun erlang:is_binary/1, Errors) of + true -> {ok, {ok, {outcome, Id, Status, Output, Value, Truncated, [I || {ok, I} <- Loaded], Errors}}}; false -> {error, nil} end; unpack_outcome(_) -> {error, nil}. diff --git a/src/albedo/harness/extensions/rolling/albedo_rolling.erl b/src/albedo/harness/extensions/rolling/albedo_rolling.erl index c8c65ef..0976f79 100644 --- a/src/albedo/harness/extensions/rolling/albedo_rolling.erl +++ b/src/albedo/harness/extensions/rolling/albedo_rolling.erl @@ -1,3 +1,15 @@ -module(albedo_rolling). --export([fingerprint/1]). -fingerprint(Inputs) -> binary:encode_hex(crypto:hash(sha256, term_to_binary(Inputs, [deterministic])), lowercase). +-export([fingerprint/1, legacy_fingerprint/1]). + +%% Images count by content hash, so a prefix fingerprints the same whether its +%% images are inline or stored; a term without images hashes as it always has. +fingerprint(Inputs) -> hash(albedo_images:canonical(Inputs)). + +%% The fingerprint saved before images were stored, over their payload bytes. +legacy_fingerprint(Inputs) -> + case albedo_images:legacy(Inputs) of + {ok, Term} -> {ok, hash(Term)}; + Error -> Error + end. + +hash(Term) -> binary:encode_hex(crypto:hash(sha256, term_to_binary(Term, [deterministic])), lowercase). diff --git a/src/albedo/harness/extensions/rolling/extension.gleam b/src/albedo/harness/extensions/rolling/extension.gleam index bb5f683..0b7ee03 100644 --- a/src/albedo/harness/extensions/rolling/extension.gleam +++ b/src/albedo/harness/extensions/rolling/extension.gleam @@ -317,7 +317,13 @@ fn prepare( // Without a window there is no threshold to compare, but a saved // projection is still the user's standing instruction: honor it. use saved <- result.try(load_state(ledger, session)) - let #(state, invalidated) = valid_state(saved, source, history) + use #(state, invalidated) <- result.try(valid_state( + ledger, + session, + saved, + source, + history, + )) use _ <- result.try(case invalidated { True -> delete_state(ledger, session) False -> Ok(Nil) @@ -347,7 +353,13 @@ fn prepare( } Some(compaction.Capacity(capacity, capacity_source)) -> { use saved <- result.try(load_state(ledger, session)) - let #(state, invalidated) = valid_state(saved, source, history) + use #(state, invalidated) <- result.try(valid_state( + ledger, + session, + saved, + source, + history, + )) use _ <- result.try(case invalidated { True -> delete_state(ledger, session) False -> Ok(Nil) @@ -553,25 +565,69 @@ fn bounded_excerpt(text: String) -> String { } fn valid_state( + ledger: store.Store, + session: String, saved: Option(State), source: String, history: List(types.Input), -) -> #(Option(State), Bool) { +) -> Result(#(Option(State), Bool), String) { + let length = list.length(history) case saved { - None -> #(None, False) + None -> Ok(#(None, False)) + Some(state) if state.cutoff_count > length -> Ok(#(None, True)) Some(state) -> { - let valid = - state.cutoff_count <= list.length(history) - && source_fingerprint(source, list.take(history, state.cutoff_count)) - == state.source_hash - case valid { - True -> #(Some(state), False) - False -> #(None, True) + let prefix = list.take(history, state.cutoff_count) + let current = source_fingerprint(source, prefix) + case current == state.source_hash { + True -> Ok(#(Some(state), False)) + // A hash saved before images were stored covered their payload + // bytes. Checking it reads them once; a match is rewritten in the + // current form and a mismatch is deleted by the caller, so this runs + // at most once per saved state. + False -> + case + has_images(prefix) + && legacy_fingerprint(#(source, prefix)) == Ok(state.source_hash) + { + True -> { + let upgraded = State(..state, source_hash: current) + use _ <- result.try(save_hash(ledger, session, current)) + Ok(#(Some(upgraded), False)) + } + False -> Ok(#(None, True)) + } } } } } +fn has_images(inputs: List(types.Input)) -> Bool { + list.any(inputs, fn(input) { + case input { + types.UserImage(..) -> True + types.ToolOutput(images: [_, ..], ..) -> True + _ -> False + } + }) +} + +fn save_hash( + ledger: store.Store, + session: String, + source_hash: String, +) -> Result(Nil, String) { + store.query(ledger, fn(db) { + sqlight.query( + "UPDATE rolling_compaction_state SET source_hash=? WHERE session=?", + db, + [sqlight.text(source_hash), sqlight.text(session)], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(error) { error.message }) + }) +} + fn legal_cutoff( history: List(types.Input), tail_budget: Int, @@ -789,3 +845,6 @@ fn source_fingerprint(source: String, inputs: List(types.Input)) -> String { @external(erlang, "albedo_rolling", "fingerprint") fn fingerprint(value: a) -> String + +@external(erlang, "albedo_rolling", "legacy_fingerprint") +fn legacy_fingerprint(value: a) -> Result(String, Nil) diff --git a/src/albedo/harness/loop.gleam b/src/albedo/harness/loop.gleam index 62505b2..dbd1c05 100644 --- a/src/albedo/harness/loop.gleam +++ b/src/albedo/harness/loop.gleam @@ -471,7 +471,7 @@ pub fn render_summary_input(input: types.Input) -> String { } fn describe_image(image: types.Image) -> String { - let #(mime, _, width, height, bytes) = types.image_parts(image) + let #(mime, width, height, bytes) = types.image_meta(image) "image " <> mime <> " " diff --git a/src/albedo/openai_api/albedo_openai_json.erl b/src/albedo/openai_api/albedo_openai_json.erl index e035a6c..7028a76 100644 --- a/src/albedo/openai_api/albedo_openai_json.erl +++ b/src/albedo/openai_api/albedo_openai_json.erl @@ -1,5 +1,5 @@ -module(albedo_openai_json). --export([encode/1, null/0, flatten/1, data_url/2]). +-export([encode/1, null/0, flatten/1, data_url/2, base64_string/1]). encode(Value) -> json:encode(Value). @@ -19,19 +19,30 @@ coalesce([], [Next | Stack], Small, Out) -> coalesce(Next, Stack, Small, Out); coalesce([H | T], Stack, Small, Out) -> coalesce(H, [T | Stack], Small, Out); coalesce(B, Stack, Small, Out) when is_binary(B), byte_size(B) >= ?SHARED -> coalesce([], Stack, [], [B | emit(Small, Out)]); +coalesce({albedo_image, _, _} = Image, Stack, Small, Out) -> + coalesce([], Stack, [], [Image | emit(Small, Out)]); coalesce(Piece, Stack, Small, Out) -> coalesce([], Stack, [Piece | Small], Out). emit([], Out) -> Out; emit(Small, Out) -> [iolist_to_binary(lists:reverse(Small)) | Out]. %% A `data:` URL as a JSON string without concatenating the payload first. The -%% base64 alphabet needs no JSON escaping, so clean data is emitted as is. -data_url(Mime, Data) -> +%% base64 alphabet needs no JSON escaping, so clean data is emitted as is. A +%% stored payload (always clean, see albedo_images.erl) is left as an +%% {albedo_image, Size, Read} placeholder that the transport reads while it +%% writes the body; nothing else may serialize such a body. +data_url(Mime, {stored_data, _, Size, Read}) -> + [<<"\"data:">>, Mime, <<";base64,">>, {albedo_image, Size, Read}, $"]; +data_url(Mime, {inline_data, Data}) -> case base64_clean(Data) of true -> [<<"\"data:">>, Mime, <<";base64,">>, Data, $"]; false -> json:encode_binary(<<"data:", Mime/binary, ";base64,", Data/binary>>) end. +%% The payload alone as a JSON string, placeholder rules as for data_url/2. +base64_string({stored_data, _, Size, Read}) -> [$", {albedo_image, Size, Read}, $"]; +base64_string({inline_data, Data}) -> json:encode_binary(Data). + base64_clean(<>) when (C >= $A andalso C =< $Z); (C >= $a andalso C =< $z); (C >= $0 andalso C =< $9); C =:= $+; C =:= $/; C =:= $= -> diff --git a/src/albedo/openai_api/albedo_openai_transport.erl b/src/albedo/openai_api/albedo_openai_transport.erl index 7f13819..7d05e6f 100644 --- a/src/albedo/openai_api/albedo_openai_transport.erl +++ b/src/albedo/openai_api/albedo_openai_transport.erl @@ -52,15 +52,91 @@ await_connection(Pid, Monitor, Target, Headers, Body, Timeout) -> end. start_request(Pid, Monitor, Target, Headers, Body, Timeout) -> - try gun:post(Pid, Target, Headers, Body, #{flow => ?FLOW}) of - Stream -> - {ok, {connection, self(), Pid, Stream, Monitor, Timeout}} + try send(Pid, Target, Headers, Body, Timeout) of + {ok, Stream} -> + {ok, {connection, self(), Pid, Stream, Monitor, Timeout}}; + {error, Reason} -> + cleanup(Pid, Monitor), + {error, {transport_error, Reason}} catch _Class:_Reason -> cleanup(Pid, Monitor), {error, {transport_error, <<"request failed">>}} end. +%% A body with stored images ({albedo_image, Size, Read} placeholders, see +%% albedo_openai_json:data_url/2) is written in segments: each payload is read +%% just before it is sent, and the next is not read until the connection +%% process has taken it, so at most one image is held at a time. +send(Pid, Target, Headers, Body, Timeout) -> + case segments(Body) of + plain -> + {ok, gun:post(Pid, Target, Headers, Body, #{flow => ?FLOW})}; + {Segments, Length} -> + Sized = [{<<"content-length">>, integer_to_binary(Length)} | Headers], + Stream = gun:post(Pid, Target, Sized, #{flow => ?FLOW}), + Deadline = erlang:monotonic_time(millisecond) + Timeout, + case write(Pid, Stream, Segments, Deadline) of + ok -> {ok, Stream}; + {error, Reason} -> + gun:cancel(Pid, Stream), + {error, Reason} + end + end. + +%% plain, or {[{data, iodata()} | {image, Size, Read}], ContentLength}. +segments(Body) -> + case walk(Body, {[], [], 0, false}) of + {_, _, _, false} -> plain; + {Current, Segments, Length, true} -> + {lists:reverse(flush(Current, Segments)), Length} + end. + +walk({albedo_image, Size, Read}, {Current, Segments, Length, _}) -> + {[], [{image, Size, Read} | flush(Current, Segments)], Length + Size, true}; +walk([H | T], Acc) -> walk(T, walk(H, Acc)); +walk([], Acc) -> Acc; +walk(Piece, {Current, Segments, Length, Images}) when is_binary(Piece) -> + {[Piece | Current], Segments, Length + byte_size(Piece), Images}; +walk(Byte, {Current, Segments, Length, Images}) when is_integer(Byte) -> + {[Byte | Current], Segments, Length + 1, Images}. + +flush([], Segments) -> Segments; +flush(Current, Segments) -> [{data, lists:reverse(Current)} | Segments]. + +write(_, _, [], _) -> ok; +write(Pid, Stream, [{data, Data} | Rest], Deadline) -> + gun:data(Pid, Stream, fin(Rest), Data), + write(Pid, Stream, Rest, Deadline); +write(Pid, Stream, [{image, Size, Read} | Rest], Deadline) -> + case Read() of + {ok, Payload} when byte_size(Payload) =:= Size -> + gun:data(Pid, Stream, fin(Rest), Payload), + case drained(Pid, Deadline) of + ok -> write(Pid, Stream, Rest, Deadline); + Error -> Error + end; + _ -> + {error, <<"a stored image payload is missing or damaged">>} + end. + +fin([]) -> fin; +fin(_) -> nofin. + +%% Waits until the connection process has consumed what it was sent. +drained(Pid, Deadline) -> + case erlang:process_info(Pid, message_queue_len) of + {message_queue_len, 0} -> ok; + undefined -> ok; + _ -> + case erlang:monotonic_time(millisecond) >= Deadline of + true -> {error, <<"timed out writing the request body">>}; + false -> + receive after 1 -> ok end, + drained(Pid, Deadline) + end + end. + receive_message({connection, Owner, Pid, Stream, Monitor, Timeout}) when Owner =:= self() -> case gun:await(Pid, Stream, Timeout, Monitor) of diff --git a/src/albedo/openai_api/request.gleam b/src/albedo/openai_api/request.gleam index d62fb30..6cdd110 100644 --- a/src/albedo/openai_api/request.gleam +++ b/src/albedo/openai_api/request.gleam @@ -299,8 +299,8 @@ fn text_part(protocol: Protocol, text: String) -> Json { } fn image_part(protocol: Protocol, image: types.Image) -> Json { - let #(mime_type, data, _, _, _) = types.image_parts(image) - let url = data_url(mime_type, data) + let #(mime_type, _, _, _) = types.image_meta(image) + let url = data_url(mime_type, types.image_data(image)) case protocol { Responses -> json.object([ @@ -422,4 +422,4 @@ fn optional( fn flatten(json: Json) -> Json @external(erlang, "albedo_openai_json", "data_url") -fn data_url(mime_type: String, data: String) -> Json +fn data_url(mime_type: String, data: types.ImageData) -> Json diff --git a/src/albedo/openai_api/types.gleam b/src/albedo/openai_api/types.gleam index 25a9175..c151655 100644 --- a/src/albedo/openai_api/types.gleam +++ b/src/albedo/openai_api/types.gleam @@ -34,7 +34,17 @@ pub const max_image_pixels = 40_000_000 /// A bounded, header-validated image. The daemon revalidates decoded bytes before /// constructing this value; dimensions are display metadata, not decode proof. pub opaque type Image { - Image(mime_type: String, data: String, width: Int, height: Int, bytes: Int) + Image(mime_type: String, data: ImageData, width: Int, height: Int, bytes: Int) +} + +/// Where an image's base64 payload lives. A saved image is a content hash into +/// the daemon's image table (see albedo_images.erl): its bytes are read only +/// while a request body is being written, so a loaded transcript holds +/// references rather than every screenshot it ever showed the model. +pub type ImageData { + InlineData(data: String) + /// `size` is the payload's byte length; `read` returns exactly those bytes. + StoredData(hash: String, size: Int, read: fn() -> Result(String, Nil)) } pub fn image( @@ -60,13 +70,27 @@ pub fn image( _, False, _, _ -> Error(InvalidRequest("invalid image payload size")) _, _, False, _ -> Error(InvalidRequest("invalid image dimensions")) _, _, _, False -> Error(InvalidRequest("invalid decoded image size")) - True, True, True, True -> Ok(Image(mime_type, data, width, height, bytes)) + True, True, True, True -> + Ok(Image(mime_type, InlineData(data), width, height, bytes)) } } -pub fn image_parts(image: Image) -> #(String, String, Int, Int, Int) { - let Image(mime_type, data, width, height, bytes) = image - #(mime_type, data, width, height, bytes) +/// MIME type, width, height, and decoded byte count. +pub fn image_meta(image: Image) -> #(String, Int, Int, Int) { + let Image(mime_type, _, width, height, bytes) = image + #(mime_type, width, height, bytes) +} + +pub fn image_data(image: Image) -> ImageData { + image.data +} + +/// The base64 payload's byte length, known without reading a stored payload. +pub fn image_size(image: Image) -> Int { + case image.data { + InlineData(data) -> string.byte_size(data) + StoredData(size: size, ..) -> size + } } pub type Input { diff --git a/test/daemon/albedo_image_store_test_support.erl b/test/daemon/albedo_image_store_test_support.erl new file mode 100644 index 0000000..3e51581 --- /dev/null +++ b/test/daemon/albedo_image_store_test_support.erl @@ -0,0 +1,14 @@ +-module(albedo_image_store_test_support). +-export([legacy_payload/2, exists/1, legacy_fingerprint/3]). + +%% A transcript row as written before images were stored: the payload inline. +legacy_payload(Text, Data) -> + term_to_binary({1, {user_image, Text, {image, <<"image/png">>, Data, 2, 3, 24}}}). + +exists(Path) -> filelib:is_regular(Path). + +%% What albedo_rolling:fingerprint/1 returned for #(Source, [UserImage(Text, _)]) +%% before images were stored. +legacy_fingerprint(Source, Text, Data) -> + Term = {Source, [{user_image, Text, {image, <<"image/png">>, Data, 2, 3, 24}}]}, + binary:encode_hex(crypto:hash(sha256, term_to_binary(Term, [deterministic])), lowercase). diff --git a/test/daemon/image_store_test.gleam b/test/daemon/image_store_test.gleam new file mode 100644 index 0000000..8311cc1 --- /dev/null +++ b/test/daemon/image_store_test.gleam @@ -0,0 +1,239 @@ +import albedo/daemon/conversation +import albedo/daemon/image +import albedo/daemon/images +import albedo/daemon/store +import albedo/harness/runtime +import albedo/openai_api +import albedo/openai_api/request +import albedo/openai_api/transport +import albedo/openai_api/types +import gleam/bit_array +import gleam/dynamic/decode +import gleam/list +import gleam/option.{None} +import gleam/string_tree +import gleeunit/should +import sqlight + +const png = "iVBORw0KGgoAAAANSUhEUgAAAAIAAAAD" + +@external(erlang, "albedo_runtime_test_support", "temporary_database") +fn temporary_database() -> String + +@external(erlang, "albedo_runtime_test_support", "cleanup") +fn cleanup(path: String) -> Nil + +@external(erlang, "albedo_image_store_test_support", "legacy_payload") +fn legacy_payload(text: String, data: String) -> BitArray + +@external(erlang, "albedo_image_store_test_support", "exists") +fn exists(path: String) -> Bool + +@external(erlang, "albedo_image_store_test_support", "legacy_fingerprint") +fn old_fingerprint(source: String, text: String, data: String) -> String + +@external(erlang, "albedo_rolling", "fingerprint") +fn fingerprint(value: a) -> String + +@external(erlang, "albedo_rolling", "legacy_fingerprint") +fn legacy_fingerprint(value: a) -> Result(String, Nil) + +type Fixture + +type Mode { + Stream +} + +@external(erlang, "albedo_openai_transport_test_server", "start") +fn start_server(mode: Mode) -> Fixture + +@external(erlang, "albedo_openai_transport_test_server", "url") +fn url(fixture: Fixture) -> String + +@external(erlang, "albedo_openai_transport_test_server", "await_body") +fn await_body(fixture: Fixture) -> BitArray + +@external(erlang, "albedo_openai_transport_test_server", "stop") +fn stop(fixture: Fixture) -> Nil + +fn test_image() -> types.Image { + let assert Ok(value) = image.validate("image/png", png, 2, 3, 24) + value +} + +fn session(id: String) -> conversation.Info { + conversation.Info( + id, + "new session", + "/tmp", + "provider", + "model", + types.Responses, + conversation.Idle, + None, + None, + ) +} + +fn ledger() -> #(String, store.Store) { + let path = temporary_database() + let assert Ok(host) = runtime.start(path) + let ledger = runtime.ledger(host) + let assert Ok(_) = conversation.initialise(ledger) + #(path, ledger) +} + +fn count(ledger: store.Store, sql: String) -> Int { + let assert Ok([n]) = + store.query(ledger, fn(db) { + sqlight.query(sql, db, [], decode.field(0, decode.int, decode.success)) + }) + n +} + +fn inputs() -> List(types.Input) { + [ + types.UserImage("look", test_image()), + types.ToolOutput("call", "rendered", [test_image(), test_image()]), + ] +} + +fn loaded(ledger: store.Store, id: String) -> List(types.Input) { + let assert Ok(entries) = conversation.load_entries(ledger, id) + list.map(entries, fn(entry) { entry.input }) +} + +pub fn commit_keeps_one_payload_and_rows_load_references_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("s")) + let assert Ok(_) = + conversation.commit(ledger, "s", inputs(), conversation.Idle) + + count(ledger, "SELECT count(*) FROM images") |> should.equal(1) + count( + ledger, + "SELECT count(*) FROM transcript WHERE instr(payload,CAST('" + <> png + <> "' AS BLOB))>0", + ) + |> should.equal(0) + + let assert [types.UserImage("look", first), types.ToolOutput(_, _, [_, _])] = + loaded(ledger, "s") + let assert types.StoredData(_, size, read) = types.image_data(first) + size |> should.equal(24 + 8) + read() |> should.equal(Ok(png)) + types.image_meta(first) |> should.equal(#("image/png", 2, 3, 24)) + cleanup(path) +} + +pub fn stored_images_stream_the_same_request_bytes_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("s")) + let assert Ok(_) = + conversation.commit(ledger, "s", inputs(), conversation.Idle) + let assert Ok(inline) = + request.encode(types.Responses, openai_api.request("m", inputs())) + let assert Ok(stored) = + request.encode( + types.Responses, + openai_api.request("m", loaded(ledger, "s")), + ) + + let fixture = start_server(Stream) + let assert Ok(connection) = transport.open(url(fixture), [], stored, 1000) + let body = await_body(fixture) + transport.close(connection) + stop(fixture) + + body |> should.equal(bit_array.from_string(string_tree.to_string(inline))) + cleanup(path) +} + +pub fn a_missing_payload_fails_the_request_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("s")) + let assert Ok(_) = + conversation.commit(ledger, "s", inputs(), conversation.Idle) + let assert Ok(body) = + request.encode( + types.Responses, + openai_api.request("m", loaded(ledger, "s")), + ) + let assert Ok(_) = + store.query(ledger, fn(db) { sqlight.exec("DELETE FROM images", db) }) + + let fixture = start_server(Stream) + let assert Error(transport.TransportError(_)) = + transport.open(url(fixture), [], body, 1000) + stop(fixture) + cleanup(path) +} + +pub fn deleting_a_session_releases_only_unshared_payloads_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("a")) + let assert Ok(_) = conversation.create(ledger, session("b")) + let assert Ok(_) = + conversation.commit(ledger, "a", inputs(), conversation.Idle) + let assert Ok(_) = + conversation.commit(ledger, "b", inputs(), conversation.Idle) + + let assert Ok(_) = conversation.delete(ledger, "a") + count(ledger, "SELECT count(*) FROM images") |> should.equal(1) + let assert [types.UserImage(_, kept), ..] = loaded(ledger, "b") + let assert types.StoredData(read: read, ..) = types.image_data(kept) + read() |> should.equal(Ok(png)) + + let assert Ok(_) = conversation.delete(ledger, "b") + count(ledger, "SELECT count(*) FROM images") |> should.equal(0) + cleanup(path) +} + +pub fn migration_moves_legacy_payloads_once_after_a_backup_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("s")) + let assert Ok(_) = + store.query(ledger, fn(db) { + sqlight.query( + "INSERT INTO transcript(session,payload) VALUES('s',?)", + db, + [sqlight.blob(legacy_payload("old", png))], + decode.dynamic, + ) + }) + let assert [types.UserImage("old", before)] = loaded(ledger, "s") + types.image_data(before) |> should.equal(types.InlineData(png)) + + let backup = path <> ".backup/before.sqlite" + images.migrate(ledger, backup) |> should.equal(Ok(1)) + exists(backup) |> should.be_true + let assert [types.UserImage("old", after)] = loaded(ledger, "s") + let assert types.StoredData(read: read, ..) = types.image_data(after) + read() |> should.equal(Ok(png)) + images.migrate(ledger, backup) |> should.equal(Ok(0)) + cleanup(path) +} + +pub fn fingerprints_ignore_where_a_payload_lives_test() { + let #(path, ledger) = ledger() + let assert Ok(_) = conversation.create(ledger, session("s")) + let assert Ok(_) = + conversation.commit( + ledger, + "s", + [types.UserImage("look", test_image())], + conversation.Idle, + ) + let stored = loaded(ledger, "s") + let inline = [types.UserImage("look", test_image())] + + fingerprint(#("source", stored)) + |> should.equal(fingerprint(#("source", inline))) + legacy_fingerprint(#("source", stored)) + |> should.equal(Ok(old_fingerprint("source", "look", png))) + // Without images the current fingerprint is the one saved before. + fingerprint(#("source", [types.User("hi")])) + |> should.equal(fingerprint(#("source", [types.User("hi")]))) + cleanup(path) +} diff --git a/test/daemon/image_test.gleam b/test/daemon/image_test.gleam index 7cf2f23..219e6ae 100644 --- a/test/daemon/image_test.gleam +++ b/test/daemon/image_test.gleam @@ -7,6 +7,7 @@ import albedo/harness/runtime import albedo/openai_api/types import gleam/dynamic/decode import gleam/json +import gleam/list import gleam/option.{None, Some} import gleeunit/should import sqlight @@ -40,7 +41,8 @@ fn session(id: String) -> conversation.Info { pub fn validation_uses_canonical_bytes_header_and_derived_metadata_test() { let value = test_image() - types.image_parts(value) |> should.equal(#("image/png", png, 2, 3, 24)) + types.image_meta(value) |> should.equal(#("image/png", 2, 3, 24)) + types.image_data(value) |> should.equal(types.InlineData(png)) image.validate("image/jpeg", png, 2, 3, 24) |> should.equal(Error("image metadata does not match its payload")) image.validate("image/png", png, 2, 4, 24) @@ -77,6 +79,22 @@ pub fn user_events_expose_metadata_without_the_base64_payload_test() { |> should.equal(Ok(#("image/png", 2, 3, 24, None))) } +/// Each loaded user image as its text and the payload its reference reads. +fn resolved( + ledger: store.Store, + id: String, +) -> List(#(String, Result(String, Nil))) { + let assert Ok(inputs) = conversation.load(ledger, id) + list.map(inputs, fn(input) { + let assert types.UserImage(text, image) = input + types.image_meta(image) |> should.equal(#("image/png", 2, 3, 24)) + case types.image_data(image) { + types.StoredData(read: read, ..) -> #(text, read()) + types.InlineData(data) -> #(text, Ok(data)) + } + }) +} + pub fn image_inputs_survive_restart_and_fork_without_losing_payload_test() { let path = temporary_database() let assert Ok(host) = runtime.start(path) @@ -96,15 +114,15 @@ pub fn image_inputs_survive_restart_and_fork_without_losing_payload_test() { ) }) let assert Ok(_) = history.fork(ledger, "source", "branch", checkpoint) - conversation.load(ledger, "source") |> should.equal(Ok([input])) - conversation.load(ledger, "branch") |> should.equal(Ok([input])) + resolved(ledger, "source") |> should.equal([#("describe", Ok(png))]) + resolved(ledger, "branch") |> should.equal([#("describe", Ok(png))]) runtime.stop(host) let assert Ok(restarted) = runtime.start(path) let ledger = runtime.ledger(restarted) let assert Ok(_) = conversation.initialise(ledger) - conversation.load(ledger, "source") |> should.equal(Ok([input])) - conversation.load(ledger, "branch") |> should.equal(Ok([input])) + resolved(ledger, "source") |> should.equal([#("describe", Ok(png))]) + resolved(ledger, "branch") |> should.equal([#("describe", Ok(png))]) runtime.stop(restarted) cleanup(path) } diff --git a/test/harness/tool_image_test.gleam b/test/harness/tool_image_test.gleam index 135e30e..712af6c 100644 --- a/test/harness/tool_image_test.gleam +++ b/test/harness/tool_image_test.gleam @@ -43,7 +43,8 @@ pub fn a_shown_image_returns_with_the_result_and_survives_recovery_test() { let assert Ok(types.ToolOutput("shown", text, [image])) = runtime.invoke(host, session, shown) field(text, "value") |> should.equal("'attached image/png, 24 bytes'") - types.image_parts(image) |> should.equal(#("image/png", png, 2, 3, 24)) + types.image_meta(image) |> should.equal(#("image/png", 2, 3, 24)) + types.image_data(image) |> should.equal(types.InlineData(png)) // A result lost in flight is rebuilt from the journal with its image. runtime.recover(host, session, shown) |> should.equal(types.ToolOutput("shown", text, [image])) @@ -71,7 +72,7 @@ pub fn view_code_returns_rendered_pages_as_images_test() { <> "shown" let assert Ok(types.ToolOutput(_, text, [page])) = runtime.invoke(host, session, call("view", code)) - let #(mime, _, width, height, _) = types.image_parts(page) + let #(mime, width, height, _) = types.image_meta(page) #(mime, width > height) |> should.equal(#("image/png", True)) field(text, "value") |> string.contains("(python)") |> should.be_true field(text, "value") @@ -162,7 +163,7 @@ pub fn the_kernel_decoder_reads_images_at_the_boundary_test() { // Both lists keep the order the cell showed its images in. mixed.images |> list.map(fn(image) { - let #(_, _, width, height, _) = types.image_parts(image) + let #(_, width, height, _) = types.image_meta(image) #(width, height) }) |> should.equal([#(2, 3), #(5, 3)]) @@ -247,8 +248,15 @@ fn journaled_png(width: Int) -> BitArray { @external(erlang, "erlang", "term_to_binary") fn term_to_binary(value: a) -> BitArray +fn unpack_input(bytes: BitArray) -> Result(types.Input, Nil) { + unpack_with(bytes, fn(_) { Error(Nil) }) +} + @external(erlang, "albedo_conversation", "unpack") -fn unpack_input(bytes: BitArray) -> Result(types.Input, Nil) +fn unpack_with( + bytes: BitArray, + read: fn(String) -> Result(String, Nil), +) -> Result(types.Input, Nil) @external(erlang, "albedo_native", "unpack_cell") fn unpack_cell( -- 2.51.2