From dfc274b1bb3319bd7f8decdc33bd03400fa96705 Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Sun, 27 Sep 2026 15:19:20 -0400 Subject: [PATCH] Reclaim evicted tool images when compaction commits --- robot-docs/compaction.md | 6 +- src/albedo/daemon/albedo_conversation.erl | 14 +- src/albedo/daemon/context_snapshot.gleam | 43 ++++-- src/albedo/daemon/images.gleam | 119 ++++++++++++++ src/albedo/daemon/session.gleam | 39 ++++- src/albedo/daemon/session_run.gleam | 12 +- src/albedo/harness/compaction.gleam | 8 +- .../harness/extensions/lcm/extension.gleam | 31 ++-- .../extensions/python/albedo_native.erl | 11 +- .../extensions/rolling/extension.gleam | 12 +- .../extensions/snapcompact/extension.gleam | 2 + src/albedo/harness/loop.gleam | 6 +- src/albedo/harness/runtime.gleam | 2 +- test/daemon/session_run_test.gleam | 2 +- test/e2e/compaction_test.py | 145 +++++++++++++++++- 15 files changed, 404 insertions(+), 48 deletions(-) diff --git a/robot-docs/compaction.md b/robot-docs/compaction.md index f89f2268..41a8cbce 100644 --- a/robot-docs/compaction.md +++ b/robot-docs/compaction.md @@ -18,6 +18,10 @@ Provider projection can combine several durable rows into one model input, split Live source references need the sequence IDs from the transaction that commits inputs. The session coordinator can then project those identified entries before preparing a request. Its current commit callback returns only a timestamp, and its in-memory entries have no sequence field. Callers that do not need references can keep the plain-input path. Text or list positions cannot identify rows reliably because messages can repeat and provider projection can change item counts. An excursion marker belongs to its parent session. The branch has different row references, so returning its outcome requires an explicit link to the parent marker. +## image payload lifecycle + +When a compaction plugin commits a new projection — through `/compact` or the automatic trigger — it signals cleanup before the next provider request. Tool-output images absent from that projection are elided: the transcript row and its Python cell keep their text and gain `[image elided and is no longer available]`, and each payload is deleted once no transcript row or pinned prompt names its hash. Tool outputs in the verbatim tail keep their images. User uploads are never elided, because their hashes are part of the saved compaction cut. Reusing a saved projection does not trigger cleanup; neither do ordinary turns or failed compactions. + ## stored folds A strategy reads folds another strategy stored through `context.prior(history)`, never by importing that strategy. It returns `Prior(folds, rest)`: the stored summaries, oldest first, and the history they do not cover. Without stored folds, `folds` is empty and `rest` is the history unchanged. The extension that owns the storage registers a `FoldPlugin`; the runtime applies every enabled provider in registry order, except a provider owned by the active strategy, which reads its own state. `lcm-memory` provides LCM's folds and `snapcompact-memory` the snapcompact archive, so each stays visible after a switch away. `lcm-memory` comes first, so the archive covers history past LCM's folds. `lcm` ignores `prior`. @@ -28,7 +32,7 @@ A strategy reads folds another strategy stored through `context.prior(history)`, Before the configured trigger, `lcm` sends ordinary history without a summarizer call. At the trigger, it summarizes complete older conversation units in chunks. The request then contains ordered summary nodes with source ranges, followed by a verbatim tail. The latest user/tool unit remains in the tail. Condensation joins older nodes into parents and keeps the children in SQLite. A failed summary does not advance the covered cursor. If a summary exceeds the size of its source, `lcm` retries with a smaller output limit and then stores a short source pointer. The original rows remain available. -The default-enabled `lcm-memory` extension provides four read-only tools. `lcm_list` pages through every stored fold in the current session, including nodes absent from the request view. `lcm_grep` searches literal text in raw history and summaries with paged results and optional node scope. `lcm_describe` returns a node's source range, children, and summary. `lcm_expand` reads bounded pages of the original text rows covered by a node. These tools remain enabled when the session switches from `lcm` to `rolling`. Image bytes remain in the transcript; the text tools show image metadata. The extension implements hierarchical memory. It does not implement the paper's large-file dispatcher, map operators, or subagent-only expansion. +The default-enabled `lcm-memory` extension provides four read-only tools. `lcm_list` pages through every stored fold in the current session, including nodes absent from the request view. `lcm_grep` searches literal text in raw history and summaries with paged results and optional node scope. `lcm_describe` returns a node's source range, children, and summary. `lcm_expand` reads bounded pages of the original text rows covered by a node. These tools remain enabled when the session switches from `lcm` to `rolling`. Image hashes and metadata remain in transcript rows until compaction elides a tool-output image, which leaves a text marker; the text tools show whichever remains. The extension implements hierarchical memory. It does not implement the paper's large-file dispatcher, map operators, or subagent-only expansion. The `lcm` section in `$ALBEDO_HOME/extensions.json` accepts the same setting names as `rolling`: diff --git a/src/albedo/daemon/albedo_conversation.erl b/src/albedo/daemon/albedo_conversation.erl index 07cd5e79..d09954cf 100644 --- a/src/albedo/daemon/albedo_conversation.erl +++ b/src/albedo/daemon/albedo_conversation.erl @@ -1,5 +1,5 @@ -module(albedo_conversation). --export([pack/1,unpack/2,unpack_trace/1,pack_list/1,unpack_list/2,row_atoms/0]). +-export([pack/1,unpack/2,unpack_trace/1,pack_list/1,unpack_list/2,row_atoms/0,elide_tool_images/1,elision_marker/1]). 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 @@ -42,3 +42,15 @@ unpack_list(Bytes,Read) -> try binary_to_term(Bytes,[safe]) of end; _ -> {error,nil} catch _:_ -> {error,nil} end. + +%% A packed tool output without its images: {Id, Row, Hashes}, the marker +%% appended to its text. {error, nil} when the row holds no image. +elide_tool_images(Bytes) -> + try binary_to_term(Bytes, [safe]) of + {1, {tool_output, Id, Text, [_ | _]}} when is_binary(Id), is_binary(Text) -> + Row = term_to_binary({1, {tool_output, Id, elision_marker(Text), []}}), + {ok, {Id, Row, albedo_images:hashes(Bytes)}}; + _ -> {error, nil} + catch _:_ -> {error, nil} end. + +elision_marker(Text) -> <>. diff --git a/src/albedo/daemon/context_snapshot.gleam b/src/albedo/daemon/context_snapshot.gleam index c8d70724..da8da6ba 100644 --- a/src/albedo/daemon/context_snapshot.gleam +++ b/src/albedo/daemon/context_snapshot.gleam @@ -72,6 +72,7 @@ pub opaque type Snapshot { context_window_tokens: Option(Int), compaction: Compaction, sections: List(Section), + retained_tool_calls: List(String), ) } @@ -176,7 +177,7 @@ pub fn with_usage(snapshot: Snapshot, metadata: usage.Metadata) -> Snapshot { pub fn estimate(snapshot: Snapshot) -> Option(Int) { case snapshot { Pending(_) -> None - Ready(_, _, _, _, _, compaction, _) -> compaction.estimated_input_tokens + Ready(compaction: compaction, ..) -> compaction.estimated_input_tokens } } @@ -201,6 +202,7 @@ pub fn ready( non_negative(context_window_tokens), compaction, sections, + [], ) } @@ -269,15 +271,35 @@ pub fn from_request( ..sections ] } - ready( - captured_at, - Some(provider), - request.model, - Some(protocol_name), - observation_capacity(observation), - observation_compaction(observation), - sections, - ) + let calls = + list.filter_map(history, fn(input) { + case input { + types.ToolOutput(id, _, _) -> Ok(id) + _ -> Error(Nil) + } + }) + case + ready( + captured_at, + Some(provider), + request.model, + Some(protocol_name), + observation_capacity(observation), + observation_compaction(observation), + sections, + ) + { + Ready(..) as snapshot -> Ready(..snapshot, retained_tool_calls: calls) + pending -> pending + } +} + +/// Tool calls in the exact prepared request, not in the durable history. +pub fn retained_tool_calls(snapshot: Snapshot) -> Option(List(String)) { + case snapshot { + Pending(_) -> None + Ready(retained_tool_calls: calls, ..) -> Some(calls) + } } fn render_inputs(inputs: List(types.Input)) -> String { @@ -408,6 +430,7 @@ pub fn summary(snapshot: Snapshot) -> json.Json { context_window_tokens, compacted, sections, + _, ) -> json.object( [ diff --git a/src/albedo/daemon/images.gleam b/src/albedo/daemon/images.gleam index e9518dbe..d679b72d 100644 --- a/src/albedo/daemon/images.gleam +++ b/src/albedo/daemon/images.gleam @@ -164,6 +164,125 @@ pub fn migrate(ledger: store.Store, backup: String) -> Result(Int, String) { } } +/// Drops the images of tool outputs a committed compaction left out of the +/// prepared request: the row and its Python cell keep their text plus a +/// marker, and payloads nothing else names are released. User uploads stay; +/// their hashes are part of the saved compaction cut. +pub fn elide_evicted( + ledger: store.Store, + session: String, + retained_calls: List(String), +) -> Result(Int, String) { + store.query(ledger, fn(db) { + use _ <- result.try( + sqlight.exec("BEGIN IMMEDIATE", db) + |> result.map_error(fn(e) { e.message }), + ) + case elide_rows(db, session, retained_calls) { + Ok(count) -> + sqlight.exec("COMMIT", db) + |> result.replace(count) + |> result.map_error(fn(e) { e.message }) + Error(e) -> { + let _ = sqlight.exec("ROLLBACK", db) + Error(e) + } + } + }) +} + +fn elide_rows( + db: sqlight.Connection, + session: String, + retained_calls: List(String), +) -> Result(Int, String) { + use rows <- result.try( + sqlight.query( + "SELECT seq,payload FROM transcript WHERE session=? AND instr(payload,CAST('tool_output' AS BLOB))>0", + db, + [sqlight.text(session)], + { + 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 }), + ) + let evicted = + list.filter_map(rows, fn(row) { + case elide_tool_images(row.1) { + Ok(#(call, payload, hashes)) -> + case list.contains(retained_calls, call) { + True -> Error(Nil) + False -> Ok(#(row.0, session <> "/" <> call, payload, hashes)) + } + Error(_) -> Error(Nil) + } + }) + use _ <- result.try( + list.try_each(evicted, fn(row) { + use _ <- result.try(write( + db, + "transcript", + "seq", + sqlight.int(row.0), + row.2, + )) + elide_cell(db, row.1) + }), + ) + use _ <- result.try(release( + db, + list.flat_map(evicted, fn(row) { row.3 }) |> list.unique, + )) + Ok(list.length(evicted)) +} + +/// A journaled cell holds its own inline copy of each image. +fn elide_cell(db: sqlight.Connection, id: String) -> Result(Nil, String) { + case + sqlight.query( + "SELECT payload FROM cells WHERE id=? AND payload IS NOT NULL", + db, + [sqlight.text(id)], + decode.field(0, decode.bit_array, decode.success), + ) + { + Ok([payload]) -> + case elide_cell_images(payload) { + Ok(elided) -> write(db, "cells", "id", sqlight.text(id), elided) + Error(_) -> Ok(Nil) + } + _ -> Ok(Nil) + } +} + +fn write( + db: sqlight.Connection, + table: String, + key: String, + value: sqlight.Value, + payload: BitArray, +) -> Result(Nil, String) { + sqlight.query( + "UPDATE " <> table <> " SET payload=? WHERE " <> key <> "=?", + db, + [sqlight.blob(payload), value], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) +} + +@external(erlang, "albedo_conversation", "elide_tool_images") +fn elide_tool_images( + payload: BitArray, +) -> Result(#(String, BitArray, List(String)), Nil) + +@external(erlang, "albedo_native", "elide_cell_images") +fn elide_cell_images(payload: BitArray) -> Result(BitArray, Nil) + /// Rows per migration transaction; a page of screenshot rows is tens of MB. const migrate_page_rows = 16 diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index 8882284f..efae5320 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -6,6 +6,7 @@ import albedo/daemon/conversation import albedo/daemon/event_buffer import albedo/daemon/events as view import albedo/daemon/family +import albedo/daemon/images import albedo/daemon/mail import albedo/daemon/session_extensions import albedo/daemon/session_history @@ -126,7 +127,7 @@ pub type Message { Option(Int), Subject(Result(Int, String)), ) - RecordContext(String, context_snapshot.Snapshot, Subject(Nil)) + RecordContext(String, context_snapshot.Snapshot, Bool, Subject(Nil)) RecordUsage(String, usage.Metadata, Subject(Result(Nil, String))) ReportPin(String, Option(Int), Subject(Nil)) Compact(Option(String), Subject(Result(json.Json, String))) @@ -779,12 +780,19 @@ fn handle(state: State, message: Message) { _, _, _ -> actor.continue(state) } } - RecordContext(id, snapshot, reply) -> { - process.send(reply, Nil) - case turn.live(state.activity, id) { - True -> actor.continue(session_state.State(..state, context: snapshot)) - False -> actor.continue(state) + RecordContext(id, snapshot, compacted, reply) -> { + let state = case turn.live(state.activity, id) { + True -> { + let state = session_state.State(..state, context: snapshot) + case compacted { + True -> elide_compacted_images(state) + False -> state + } + } + False -> state } + process.send(reply, Nil) + actor.continue(state) } ReadContext(reply) -> { process.send(reply, context_snapshot.summary(state.context)) @@ -1464,6 +1472,25 @@ fn resume(state: State) -> State { } } +/// The plugin has committed its new projection. Clean up before the next +/// provider request, while the session actor serializes transcript writes. +fn elide_compacted_images(state: State) -> State { + let elided = case context_snapshot.retained_tool_calls(state.context) { + Some(calls) -> + images.elide_evicted(runtime.ledger(state.host), state.info.id, calls) + None -> Ok(0) + } + case elided { + Ok(0) -> state + Ok(_) -> session_state.State(..state, history: None) + Error(error) -> + session_state.emit( + state, + view.text("error", "image cleanup skipped: " <> error), + ) + } +} + fn compact( state: State, strategy: Option(String), diff --git a/src/albedo/daemon/session_run.gleam b/src/albedo/daemon/session_run.gleam index b711b227..ad4b7142 100644 --- a/src/albedo/daemon/session_run.gleam +++ b/src/albedo/daemon/session_run.gleam @@ -22,7 +22,8 @@ pub type Messages(message) { Option(Int), Subject(Result(Int, String)), ) -> message, - context: fn(String, context_snapshot.Snapshot, Subject(Nil)) -> message, + context: fn(String, context_snapshot.Snapshot, Bool, Subject(Nil)) -> + message, usage: fn(String, usage.Metadata, Subject(Result(Nil, String))) -> message, drain: fn(String, Subject(Result(List(types.Input), String))) -> message, pin: fn(String, Option(Int), Subject(Nil)) -> message, @@ -181,7 +182,7 @@ pub fn start( client, publish_fn(owner, run_id, messages, stop, 30_000), commit_fn(owner, run_id, messages, 10_000), - fn(request, observation) { + fn(request, observation, compacted) { let snapshot = context_snapshot.from_request( Some(usage.now()), @@ -191,9 +192,10 @@ pub fn start( request, observation, ) - // The snapshot only feeds the live view, so a stalled session actor - // must not kill the turn over it; fire and forget after the bound. - let _ = try_call(owner, 5000, messages.context(run_id, snapshot, _)) + // Inspection and post-compaction cleanup must not kill the turn if + // the session actor stalls; leave the message queued after the bound. + let _ = + try_call(owner, 5000, messages.context(run_id, snapshot, compacted, _)) Nil }, usage_fn(owner, run_id, messages, 10_000), diff --git a/src/albedo/harness/compaction.gleam b/src/albedo/harness/compaction.gleam index cafa9eef..ece40625 100644 --- a/src/albedo/harness/compaction.gleam +++ b/src/albedo/harness/compaction.gleam @@ -101,7 +101,13 @@ pub fn compose_prior( /// What a strategy prepared for one request. The observation describes this /// exact preparation, so inspectors do not need to query strategy-owned state. pub type Prepared { - Prepared(inputs: List(types.Input), observation: Option(Observation)) + Prepared( + inputs: List(types.Input), + observation: Option(Observation), + /// True only when this preparation committed a new compaction, not when + /// it reused a saved projection. Triggers post-compaction cleanup. + compacted: Bool, + ) } /// Strategy-neutral facts for the request inspector. A strategy may add more diff --git a/src/albedo/harness/extensions/lcm/extension.gleam b/src/albedo/harness/extensions/lcm/extension.gleam index 18795d53..ec7311b6 100644 --- a/src/albedo/harness/extensions/lcm/extension.gleam +++ b/src/albedo/harness/extensions/lcm/extension.gleam @@ -198,6 +198,7 @@ fn prepare( history, )) False -> { + let previous_frontier = frontier use latest_user <- result.try(latest_user_seq(sources)) let eligible = list.filter(sources, fn(item) { @@ -255,17 +256,24 @@ fn prepare( "LCM summaries and the newest conversation/tool unit do not fit the context window", ) }) - Ok(prepared( - next, - frontier, - "compacted", - "estimated from source-backed summary nodes; window from " - <> capacity_source, - Some(capacity), - next_estimated, - config, - history, - )) + let prepared = + prepared( + next, + frontier, + "compacted", + "estimated from source-backed summary nodes; window from " + <> capacity_source, + Some(capacity), + next_estimated, + config, + history, + ) + Ok( + compaction.Prepared( + ..prepared, + compacted: frontier != previous_frontier, + ), + ) } } } @@ -300,6 +308,7 @@ fn prepared( Some(list.length(original)), Some(list.length(inputs)), )), + False, ) } diff --git a/src/albedo/harness/extensions/python/albedo_native.erl b/src/albedo/harness/extensions/python/albedo_native.erl index 52d35ce4..385e2da3 100644 --- a/src/albedo/harness/extensions/python/albedo_native.erl +++ b/src/albedo/harness/extensions/python/albedo_native.erl @@ -1,11 +1,20 @@ -module(albedo_native). --export([new_id/0, pack_cell/1, unpack_cell/1]). +-export([new_id/0, pack_cell/1, unpack_cell/1, elide_cell_images/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}). +%% A finished cell without its images, the elision marker after its output. +elide_cell_images(Binary) -> + case unpack_cell(Binary) of + {ok, {ok, {outcome, Id, Status, Output, Value, Truncated, [_ | _], Errors}}} -> + Marked = albedo_conversation:elision_marker(Output), + {ok, pack_cell({ok, {outcome, Id, Status, Marked, Value, Truncated, [], Errors}})}; + _ -> {error, nil} + end. + unpack_cell(Binary) -> try binary_to_term(Binary, [safe]) of %% Cells journaled before images carry neither image field. diff --git a/src/albedo/harness/extensions/rolling/extension.gleam b/src/albedo/harness/extensions/rolling/extension.gleam index 1b26f8da..27001cca 100644 --- a/src/albedo/harness/extensions/rolling/extension.gleam +++ b/src/albedo/harness/extensions/rolling/extension.gleam @@ -148,7 +148,7 @@ fn prepare_view( use compaction.Prior(folds, rest) <- result.try(context.prior(history)) let folded = list.append(folds, rest) let carries_folds = folds != [] - use inputs <- result.try(prepare(config, context, folded)) + use #(inputs, compacted) <- result.try(prepare(config, context, folded)) // Diagnostics must not fail a request after its projection was committed. let recorded = observation(context.store, context.session) |> result.unwrap(None) @@ -177,7 +177,7 @@ fn prepare_view( )) None -> None } - Ok(compaction.Prepared(inputs, observed)) + Ok(compaction.Prepared(inputs, observed, compacted)) } fn validate_config(config: Config) -> Result(Config, String) { @@ -297,7 +297,7 @@ fn prepare( config: Config, context: compaction.Context, history: List(types.Input), -) -> Result(List(types.Input), String) { +) -> Result(#(List(types.Input), Bool), String) { let compaction.Context( ledger, session, @@ -364,7 +364,7 @@ fn prepare( current.2, ), )) - Ok(current.3) + Ok(#(current.3, False)) } Some(compaction.Capacity(capacity, capacity_source)) -> { let estimated = pinned_tokens + compaction.estimate_inputs(current.3) @@ -416,7 +416,7 @@ fn prepare( current.2, ), )) - Ok(current.3) + Ok(#(current.3, False)) } False -> { let #(previous_summary, evicted, rest) = case resumed { @@ -485,7 +485,7 @@ fn prepare( next_state, observation, )) - Ok(next.3) + Ok(#(next.3, True)) } } } diff --git a/src/albedo/harness/extensions/snapcompact/extension.gleam b/src/albedo/harness/extensions/snapcompact/extension.gleam index 8c39c33b..5e256447 100644 --- a/src/albedo/harness/extensions/snapcompact/extension.gleam +++ b/src/albedo/harness/extensions/snapcompact/extension.gleam @@ -428,6 +428,7 @@ fn visual_view( Ok(compaction.Prepared( list.append(folds, current), observe(status, current), + False, )) True, #(newly_evicted, tail) -> { let archive = extend(shape, limit, previous, evicted, newly_evicted) @@ -436,6 +437,7 @@ fn visual_view( Ok(compaction.Prepared( list.append(folds, prepared), observe(status, prepared), + True, )) } } diff --git a/src/albedo/harness/loop.gleam b/src/albedo/harness/loop.gleam index 79314b27..f2969f24 100644 --- a/src/albedo/harness/loop.gleam +++ b/src/albedo/harness/loop.gleam @@ -27,7 +27,8 @@ pub type Loop { /// model thought before it. commit: fn(List(types.Input), conversation.Stage, Option(Int)) -> Result(Int, String), - record_context: fn(types.Request, Option(compaction.Observation)) -> Nil, + record_context: fn(types.Request, Option(compaction.Observation), Bool) -> + Nil, record_usage: fn(usage.Metadata) -> Result(Nil, String), drain_steering: fn() -> Result(List(types.Input), String), /// Reports the pin's compaction baseline (`Some`) or that compaction @@ -87,7 +88,7 @@ pub fn run( None, types.Options(..types.defaults, effort: state.effort), ) - state.record_context(request, prepared.observation) + state.record_context(request, prepared.observation, prepared.compacted) use turn <- result.try( stream_with_retries(state.upstream, request, state.publish, fn(event) { case view.stream_event(id, step, event) { @@ -261,6 +262,7 @@ pub fn compact(state: Loop, inputs: List(types.Input)) -> Result(Nil, String) { types.defaults, ), prepared.observation, + prepared.compacted, ) Ok(Nil) } diff --git a/src/albedo/harness/runtime.gleam b/src/albedo/harness/runtime.gleam index 9ff17a7d..b83b78b4 100644 --- a/src/albedo/harness/runtime.gleam +++ b/src/albedo/harness/runtime.gleam @@ -1217,7 +1217,7 @@ pub fn prepare_view_scoped( let enabled = extension.extensions(session.composition) case extension.compaction(enabled) { None if force -> Error("no compaction strategy is enabled") - None -> Ok(compaction.Prepared(history, None)) + None -> Ok(compaction.Prepared(history, None, False)) Some(strategy) -> strategy.prepare( compaction.Context( diff --git a/test/daemon/session_run_test.gleam b/test/daemon/session_run_test.gleam index dbef79ad..52514cd7 100644 --- a/test/daemon/session_run_test.gleam +++ b/test/daemon/session_run_test.gleam @@ -89,7 +89,7 @@ fn messages() -> session_run.Messages(Owner) { session_run.Messages( publish: fn(_, _, reply) { Publish(reply) }, commit: fn(_, _, _, _, reply) { Commit(reply) }, - context: fn(_, _, _) { Unused }, + context: fn(_, _, _, _) { Unused }, usage: fn(_, _, _) { Unused }, drain: fn(_, _) { Unused }, pin: fn(_, _, _) { Unused }, diff --git a/test/e2e/compaction_test.py b/test/e2e/compaction_test.py index 79e00221..6cde523a 100644 --- a/test/e2e/compaction_test.py +++ b/test/e2e/compaction_test.py @@ -1,11 +1,16 @@ -"""Image tool outputs reach the model without transporting invalid images.""" +"""Image tool outputs reach the model, and compaction reclaims the ones it evicts.""" +import hashlib import json +import sqlite3 import unittest -from harness import Albedo, Provider, python, text +from harness import Albedo, Provider, error, exclusive, python, text PNG = "iVBORw0KGgoAAAANSUhEUgAAAAIAAAAD" +SHOWN = "iVBORw0KGgoAAAANSUhEUgAAABEAAAAD" +UPLOADED = "iVBORw0KGgoAAAANSUhEUgAAABIAAAAD" +MARKER = "image elided and is no longer available" class ImageToolOutputTest(unittest.TestCase): @@ -44,3 +49,139 @@ class ImageToolOutputTest(unittest.TestCase): self.assertNotIn(PNG, json.dumps(output)) finally: provider.close() + + def test_compaction_elides_evicted_tool_images_only(self): + def reply(request): + inputs = request["input"] + user = next((str(item.get("content", "")) for item in reversed(inputs) + if item.get("role") == "user"), "") + if "" in user: + return text("older conversation summary") + if inputs[-1].get("type") != "function_call_output" and user == "show": + return python("show_image(__import__('base64').b64decode('" + SHOWN + "'))") + return text("done") + + provider = Provider(reply) + try: + with Albedo(provider, protocol="responses") as app: + shown = hashlib.sha256(SHOWN.encode()).hexdigest() + uploaded = hashlib.sha256(UPLOADED.encode()).hexdigest() + database = f"file:{app.home / 'albedo.sqlite'}?mode=ro" + + def converse(sid, prompts): + for prompt in prompts: + app.prompt(sid, prompt).close() + app.idle(sid) + with app.api(f"/sessions/{sid}/commands", {"name": "/compact"}) as response: + self.assertTrue(json.load(response)["result"]["started"]) + app.idle(sid) + + def rows(sid, table, needle): + with sqlite3.connect(database, uri=True) as db: + return db.execute(f"SELECT count(*) FROM {table} WHERE session=? " + "AND instr(payload, CAST(? AS BLOB))>0", + (sid, needle)).fetchone()[0] + + old = app.session() + app.api(f"/sessions/{old}/events", {"content": "keep this upload", "image": { + "mimeType": "image/png", "data": UPLOADED, "width": 18, "height": 3, + "bytes": 24}}).close() + app.idle(old) + converse(old, ["show", "second", "third", "fourth"]) + self.assertEqual(rows(old, "transcript", shown), 0) + self.assertEqual(rows(old, "cells", SHOWN), 0) + self.assertEqual(rows(old, "transcript", MARKER), 1) + self.assertEqual(rows(old, "cells", MARKER), 1) + self.assertEqual(rows(old, "transcript", uploaded), 1) + with sqlite3.connect(database, uri=True) as db: + stored = {row[0] for row in db.execute("SELECT hash FROM images")} + self.assertNotIn(shown, stored) + self.assertIn(uploaded, stored) + app.prompt(old, "after compaction").close() + app.idle(old) + self.assertIn("older conversation summary", + json.dumps(provider.requests[-1]["request"]["input"])) + + recent = app.session() + converse(recent, ["first", "second", "third", "show"]) + self.assertEqual(rows(recent, "transcript", shown), 1) + self.assertEqual(rows(recent, "cells", SHOWN), 1) + finally: + provider.close() + + @exclusive + def test_automatic_compaction_elides_evicted_tool_images(self): + fail_summary = True + attempts = [] + cleaned_before_request = [] + + def reply(request): + inputs = request["input"] + user = next((str(item.get("content", "")) for item in reversed(inputs) + if item.get("role") == "user"), "") + if "" in user: + attempts.append(fail_summary) + if fail_summary: + return error(400, "summary refused") + return text("older conversation summary") + if "[older conversation summary;" in json.dumps(inputs): + cleaned_before_request.append( + (rows("transcript", MARKER), rows("cells", MARKER))) + return error(400, "provider refused after committed compaction") + if inputs[-1].get("type") != "function_call_output" and user == "show": + return python("show_image(__import__('base64').b64decode('" + SHOWN + "'))") + return text("done") + + provider = Provider(reply) + try: + with Albedo(provider, protocol="responses") as app: + settings_path = app.home / "extensions.json" + original = settings_path.read_text() + + def restore(): + settings_path.write_text(original) + app.restart() + + self.addCleanup(restore) + settings = json.loads(original) + settings["rolling"] = {"contextWindowTokens": 50000} + settings_path.write_text(json.dumps(settings)) + app.restart() + session = app.session() + shown = hashlib.sha256(SHOWN.encode()).hexdigest() + database = f"file:{app.home / 'albedo.sqlite'}?mode=ro" + + def rows(table, needle): + with sqlite3.connect(database, uri=True) as db: + return db.execute(f"SELECT count(*) FROM {table} WHERE session=? " + "AND instr(payload, CAST(? AS BLOB))>0", + (session, needle)).fetchone()[0] + + app.prompt(session, "show").close() + app.idle(session) + self.assertEqual(rows("transcript", shown), 1) + self.assertEqual(rows("cells", SHOWN), 1) + pad = "history " * 8000 + for prompt in (pad, pad, pad): + app.prompt(session, prompt).close() + app.idle(session) + self.assertTrue(attempts, "the automatic trigger must attempt compaction") + self.assertEqual(rows("transcript", shown), 1) + self.assertEqual(rows("cells", SHOWN), 1) + fail_summary = False + app.prompt(session, "retry compaction").close() + app.idle(session) + self.assertTrue(cleaned_before_request, "expected a post-compaction request") + self.assertTrue(all(pair == (1, 1) for pair in cleaned_before_request)) + self.assertEqual(rows("transcript", shown), 0) + self.assertEqual(rows("cells", SHOWN), 0) + self.assertEqual(rows("transcript", MARKER), 1) + self.assertEqual(rows("cells", MARKER), 1) + summaries = len(attempts) + app.prompt(session, "after compaction").close() + app.idle(session) + self.assertEqual(len(attempts), summaries, "reuse the committed summary") + self.assertIn("older conversation summary", + json.dumps(provider.requests[-1]["request"]["input"])) + finally: + provider.close() -- knotmirror 0.0.1