Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104"""Snapshots survive real actor eviction without retaining opaque replay binaries."""
import jsonimport osfrom pathlib import Pathimport unittest
from inspect_support import Inspectionfrom harness import Albedo, Provider, Reply, ROOT, exclusive
# exclusive: enables daemon inspection and global rolling settings@exclusiveclass ContextRetentionTest(unittest.TestCase): @exclusive def test_eviction_releases_replay_while_inspection_survives(self): payload = "REPLAY_SENTINEL:" + "x" * (256 * 1024) reply = Reply( "raw", events=[ { "type": "response.completed", "response": { "id": "retention-response", "status": "completed", "output": [ { "type": "reasoning", "id": "retention-reasoning", "summary": [], "encrypted_content": payload, }, { "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "done"}], }, ], "usage": {"input_tokens": 31, "output_tokens": 2}, }, } ], ) provider = Provider(lambda _: reply) self.addCleanup(provider.close)
def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1" app.write_extensions({"rolling": {"contextWindowTokens": 2_000_000}})
with Albedo(provider, protocol="responses", prepare=prepare) as app: session = app.session() for prompt in ("first", "second"): app.prompt(session, prompt).close() app.idle(session) self.assertEqual(len(provider.requests), 2) self.assertTrue( any( item.get("encrypted_content") == payload for item in provider.requests[1]["request"]["input"] ), "second request lost opaque replay: " + json.dumps( [ (item.get("type") or item.get("role"), len(json.dumps(item))) for item in provider.requests[1]["request"]["input"] ] ) + " events=" + json.dumps( [entry["kind"] for entry in app.history(session)["items"]] ), ) route = f"/sessions/{session}/context"
def read(path): with app.api(path) as response: return json.load(response)
self.assertEqual(read(route)["state"], "ready") inspection = Inspection( app, ROOT / "test/daemon/albedo_context_snapshot_probe.erl" ) measured = inspection.call_json( "actor_json", f"[<<{json.dumps(session)}>>]" ) artifact = os.environ.get("ALBEDO_INSPECTOR_ARTIFACT") if artifact: Path(artifact).write_text(json.dumps(measured, indent=2) + "\n") self.assertTrue(measured["history_unloaded"]) self.assertFalse(measured["replay_reachable"]) self.assertLess(measured["actor_binary_bytes"], len(payload)) # Text turns never open a kernel, so there is none to release. self.assertFalse(measured["kernel_released"]) for key in ( "history_readable", "context_cleared", "actor_stopped", "workers_terminated", ): self.assertTrue(measured[key], key) self.assertEqual(measured["replay_backing_binary_bytes"], 0) self.assertEqual(len(provider.requests), 2)