okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(test): share inspection and stream supportcba5ff8

main
branchesmainall refs →
3.9 KiB101 linesraw
1"""Snapshots survive real actor eviction without retaining opaque replay binaries."""
2
3import json
4import os
5from pathlib import Path
6import unittest
7
8from inspect_support import Inspection
9from harness import Albedo, Provider, Reply, ROOT, exclusive
10
11
12# exclusive: enables daemon inspection and global rolling settings
13@exclusive
14class ContextRetentionTest(unittest.TestCase):
15    def test_eviction_releases_replay_while_inspection_survives(self):
16        payload = "REPLAY_SENTINEL:" + "x" * (256 * 1024)
17        reply = Reply(
18            "raw",
19            events=[
20                {
21                    "type": "response.completed",
22                    "response": {
23                        "id": "retention-response",
24                        "status": "completed",
25                        "output": [
26                            {
27                                "type": "reasoning",
28                                "id": "retention-reasoning",
29                                "summary": [],
30                                "encrypted_content": payload,
31                            },
32                            {
33                                "type": "message",
34                                "role": "assistant",
35                                "content": [{"type": "output_text", "text": "done"}],
36                            },
37                        ],
38                        "usage": {"input_tokens": 31, "output_tokens": 2},
39                    },
40                }
41            ],
42        )
43        provider = Provider(lambda _: reply)
44        self.addCleanup(provider.close)
45
46        def prepare(app):
47            app.daemon.env["ALBEDO_INSPECT"] = "1"
48            app.write_extensions({"rolling": {"contextWindowTokens": 2_000_000}})
49
50        with Albedo(provider, protocol="responses", prepare=prepare) as app:
51            session = app.session()
52            for prompt in ("first", "second"):
53                app.prompt(session, prompt).close()
54                app.idle(session)
55            self.assertEqual(len(provider.requests), 2)
56            self.assertTrue(
57                any(
58                    item.get("encrypted_content") == payload
59                    for item in provider.requests[1]["request"]["input"]
60                ),
61                "second request lost opaque replay: "
62                + json.dumps(
63                    [
64                        (item.get("type") or item.get("role"), len(json.dumps(item)))
65                        for item in provider.requests[1]["request"]["input"]
66                    ]
67                )
68                + " events="
69                + json.dumps(
70                    [entry["kind"] for entry in app.history(session)["items"]]
71                ),
72            )
73            route = f"/sessions/{session}/context"
74
75            def read(path):
76                with app.api(path) as response:
77                    return json.load(response)
78
79            self.assertEqual(read(route)["state"], "ready")
80            inspection = Inspection(
81                app, ROOT / "test/daemon/albedo_context_snapshot_probe.erl"
82            )
83            measured = inspection.call_json(
84                "actor_json", f"[<<{json.dumps(session)}>>]"
85            )
86            artifact = os.environ.get("ALBEDO_INSPECTOR_ARTIFACT")
87            if artifact:
88                Path(artifact).write_text(json.dumps(measured, indent=2) + "\n")
89            self.assertTrue(measured["history_unloaded"])
90            self.assertFalse(measured["replay_reachable"])
91            self.assertLess(measured["actor_binary_bytes"], len(payload))
92            for key in (
93                "history_readable",
94                "kernel_released",
95                "context_cleared",
96                "actor_stopped",
97                "workers_terminated",
98            ):
99                self.assertTrue(measured[key], key)
100            self.assertEqual(measured["replay_backing_binary_bytes"], 0)
101            self.assertEqual(len(provider.requests), 2)