hey! this is a static git repo viewer. commit diffs, file history, older commits and other revisions won't load without javascript.

you are still able to view source files on branch tips, but the experience is degraded.

okami.mom/albedo

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

Aly Raffaufrefactor(test): share inspection and stream supportcba5ff82026-10-05

1"""An attachment during generation must recover text and thinking before replay.
2
3The provider pauses only after a responsive subscriber has observed its prefix;
4this exercises real actor capture, HTTP leases, and spill files without sleeps.
5"""
6
7import base64
8import json
9import threading
10import time
11import unittest
12import urllib.error
13import urllib.parse
14
15from harness import Albedo, Provider, Reply, ROOT, exclusive, operation_id, text
16from inspect_support import Inspection
17from tool_progress_test import current_progress, tool_events
18
19
20def delta(text, thinking=False):
21    return {
22        "id": "fixture",
23        "choices": [
24            {
25                "index": 0,
26                "delta": {"reasoning_content" if thinking else "content": text},
27                "finish_reason": None,
28            }
29        ],
30    }
31
32
33def reference_text(app, reference):
34    pieces = []
35    token = None
36    while True:
37        url = urllib.parse.urlsplit(reference["url"])
38        query = urllib.parse.parse_qs(url.query)
39        if token is not None:
40            query["next"] = [token]
41        path = urllib.parse.urlunsplit(
42            ("", "", url.path, urllib.parse.urlencode(query, doseq=True), "")
43        )
44        with app.api(path) as response:
45            page = json.load(response)
46        for part in page["parts"]:
47            if part["field"] == reference["field"]:
48                pieces.append(
49                    base64.b64decode(part["text"])
50                    if part["encoding"] == "base64"
51                    else part["text"].encode()
52                )
53        token = page["next"]
54        if token is None:
55            return b"".join(pieces).decode()
56
57
58def active_text(app, descriptor):
59    return (
60        descriptor["text"]
61        if descriptor["text"] is not None
62        else reference_text(app, descriptor["reference"])
63    )
64
65
66class ActiveOutputProbe:
67    """Inspect actor memory or age a retired lease without advancing wall time."""
68
69    def __init__(self, app):
70        self.inspection = Inspection(
71            app, ROOT / "test/e2e/albedo_active_output_probe.erl"
72        )
73
74    def call(self, function, *arguments):
75        values = ",".join(f"<<{json.dumps(str(argument))}>>" for argument in arguments)
76        return self.inspection.call_json(function, f"[{values}]")
77
78
79class PausedOutput:
80    """One scripted response, with an actor-observed barrier and cleanup release."""
81
82    def __init__(self, test, app, provider, chunks, thinking="", *, session=None):
83        self.app, self.prefix, self.thinking = app, "".join(chunks), thinking
84        self.suffix = "FINAL_SUFFIX"
85        self.release, self.ready, self.observed = (threading.Event() for _ in range(3))
86        self.events, self.errors, self.initial = [], [], None
87        self.first_text_at = self.prefix_at = None
88        test.addCleanup(self.release.set)
89
90        def response():
91            if thinking:
92                yield delta(thinking, True)
93            for chunk in chunks:
94                yield delta(chunk)
95            if not self.release.wait(60):
96                raise AssertionError("paused provider was not released")
97            yield delta(self.suffix)
98            yield {
99                "id": "fixture",
100                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
101            }
102
103        provider.script = lambda _: Reply("text", events=response())
104        self.session = session or app.session()
105        self.input_id = operation_id()
106
107        def observe():
108            text_chars = 0
109            try:
110                with app.api(
111                    f"/sessions/{self.session}", headers={"Accept": "text/event-stream"}
112                ) as stream:
113                    for line in stream:
114                        if not line.startswith(b"data: "):
115                            continue
116                        batch = json.loads(line[6:])
117                        if self.initial is None:
118                            self.initial = batch
119                            self.ready.set()
120                        for event in batch["events"]:
121                            self.events.append(event)
122                            if event["type"] == "text":
123                                now = time.perf_counter()
124                                self.first_text_at = self.first_text_at or now
125                                text_chars += len(event["data"]["text"])
126                                if text_chars == len(self.prefix):
127                                    self.prefix_at = now
128                                    self.observed.set()
129                            if event["type"] == "turn_completed":
130                                return
131            except Exception as error:
132                self.errors.append(error)
133                self.ready.set()
134                self.observed.set()
135
136        self.thread = threading.Thread(target=observe, daemon=True)
137        self.thread.start()
138        test.assertTrue(self.ready.wait(10), "subscriber did not attach")
139        app.api(
140            f"/sessions/{self.session}/inputs/{self.input_id}",
141            {"kind": "message", "text": "generate the fixture"},
142            method="PUT",
143        ).close()
144        test.assertTrue(self.observed.wait(20), "actor did not emit complete prefix")
145        test.assertFalse(self.errors)
146
147    def snapshot(self):
148        with self.app.api(f"/sessions/{self.session}?tail=20") as response:
149            return json.load(response)
150
151    def finish(self, test):
152        self.release.set()
153        self.app.idle(self.session)
154        self.thread.join(10)
155        test.assertFalse(self.thread.is_alive(), "subscriber did not see completion")
156        test.assertFalse(self.errors)
157
158
159class ActiveOutputTest(unittest.TestCase):
160    def setUp(self):
161        self.provider = Provider(lambda _: Reply("text", value="unused"))
162        self.addCleanup(self.provider.close)
163        self.app = self.enterContext(Albedo(self.provider))
164
165    def assert_prefix(self, paused, snapshot):
166        descriptors = snapshot["active_output"]
167        by_kind = {item["kind"]: item for item in descriptors}
168        self.assertEqual(active_text(self.app, by_kind["text"]), paused.prefix)
169        self.assertEqual(by_kind["text"]["bytes"], len(paused.prefix.encode()))
170        if paused.thinking:
171            self.assertEqual(
172                active_text(self.app, by_kind["thinking"]), paused.thinking
173            )
174        self.assertTrue(
175            all(item["message_id"] and item["run_id"] for item in descriptors)
176        )
177        return descriptors
178
179    def test_fresh_attachment_restores_text_and_thinking_before_future_deltas(self):
180        paused = PausedOutput(
181            self, self.app, self.provider, ["prefix α🙂"], "reasoning β"
182        )
183        batch = self.app.stream_page(paused.session)
184        descriptors = self.assert_prefix(paused, batch["snapshot"])
185        paused.finish(self)
186        commits = [
187            event["data"] for event in paused.events if event["type"] == "committed"
188        ]
189        replaced = {
190            identifier
191            for commit in commits
192            for identifier in commit["replaces_live_ids"]
193        }
194        self.assertTrue({item["message_id"] for item in descriptors} <= replaced)
195        self.assertEqual(paused.snapshot()["active_output"], [])
196
197    def test_evicted_cursor_restores_large_prefix_and_unflushed_tail(self):
198        chunks = [str(index).zfill(6) + "x" * 250 for index in range(4096)] + [
199            "small tail α🙂"
200        ]
201        paused = PausedOutput(self, self.app, self.provider, chunks, "thinking" * 10000)
202        batch = self.app.stream_page(paused.session, paused.initial)
203        self.assertEqual(batch["events"][0]["type"], "reset")
204        descriptors = self.assert_prefix(paused, batch["snapshot"])
205        self.assertTrue(any(item["reference"] is not None for item in descriptors))
206        paused.finish(self)
207
208    def test_encoded_budget_spills_escaped_unicode_without_truncation(self):
209        paused = PausedOutput(
210            self, self.app, self.provider, ['"\\\n🙂' * 12000], "β" * 18000
211        )
212        snapshot = paused.snapshot()
213        descriptors = self.assert_prefix(paused, snapshot)
214        inline = [item["text"] for item in descriptors if item["text"] is not None]
215        self.assertLessEqual(
216            sum(len(json.dumps(text, ensure_ascii=False).encode()) for text in inline),
217            65536,
218        )
219        self.assertTrue(any(item["reference"] is not None for item in descriptors))
220        paused.finish(self)
221
222    def test_json_string_budget_below_at_and_above_boundary(self):
223        for size in [65533, 65534, 65535]:
224            with self.subTest(bytes=size):
225                paused = PausedOutput(self, self.app, self.provider, ["x" * size])
226                descriptors = self.assert_prefix(paused, paused.snapshot())
227                self.assertEqual(descriptors[0]["reference"] is not None, size > 65534)
228                paused.finish(self)
229
230    def test_snapshot_lease_remains_readable_after_commit(self):
231        paused = PausedOutput(self, self.app, self.provider, ["large α🙂" * 20000])
232        descriptors = paused.snapshot()["active_output"]
233        reference = next(
234            item["reference"] for item in descriptors if item["reference"] is not None
235        )
236        path = urllib.parse.urlsplit(reference["url"]).path
237        for query in ["", "?snapshot="]:
238            with self.assertRaises(urllib.error.HTTPError) as caught:
239                self.app.api(path + query)
240            self.assertEqual(caught.exception.code, 400)
241        paused.finish(self)
242        self.assertEqual(reference_text(self.app, reference), paused.prefix)
243
244    def test_cancellation_clears_prefix_before_next_run(self):
245        paused = PausedOutput(self, self.app, self.provider, ["cancelled prefix α🙂"])
246        old_ids = {item["message_id"] for item in paused.snapshot()["active_output"]}
247        self.app.api(
248            f"/sessions/{paused.session}/inputs/{paused.input_id}/cancel", {}
249        ).close()
250        paused.release.set()
251        self.app.idle(paused.session)
252        self.assertEqual(paused.snapshot()["active_output"], [])
253        next_run = PausedOutput(
254            self, self.app, self.provider, ["new run prefix"], session=paused.session
255        )
256        descriptors = self.assert_prefix(next_run, next_run.snapshot())
257        self.assertTrue(old_ids.isdisjoint(item["message_id"] for item in descriptors))
258        next_run.finish(self)
259
260    def gated_model_steps(self, *, first_completes):
261        gates = [threading.Event(), threading.Event()]
262        ready = [threading.Event(), threading.Event()]
263        for gate in gates:
264            self.addCleanup(gate.set)
265
266        def model_events(index):
267            for event in [
268                delta(f"step {index} prefix α🙂"),
269                delta(f"step {index} thinking β", True),
270            ]:
271                event["id"] = "progress-response"
272                yield event
273            yield from tool_events(
274                "chat_completions",
275                [(0, f"step-{index}", f"pass # step {index}")],
276                ready[index],
277                gates[index],
278                complete=index == 1 or first_completes,
279            )
280
281        self.provider.script = lambda _: [
282            Reply("text", events=model_events(0)),
283            Reply("text", events=model_events(1)),
284            text("done"),
285        ]
286        session = self.app.session()
287        self.app.prompt(session, "exercise model steps").close()
288        self.assertTrue(ready[0].wait(20))
289        old = current_progress(self.app, session, lambda values: len(values) == 1)[
290            "snapshot"
291        ]
292        old_ids = {item["message_id"] for item in old["active_output"]}
293        self.assertEqual(
294            {active_text(self.app, item) for item in old["active_output"]},
295            {"step 0 prefix α🙂", "step 0 thinking β"},
296        )
297        gates[0].set()
298        self.assertTrue(ready[1].wait(20))
299        new = current_progress(
300            self.app,
301            session,
302            lambda values: (
303                len(values) == 1
304                and values[0].get("preview", {}).get("text") == "pass # step 1"
305            ),
306        )["snapshot"]
307        self.assertEqual(
308            {active_text(self.app, item) for item in new["active_output"]},
309            {"step 1 prefix α🙂", "step 1 thinking β"},
310        )
311        self.assertTrue(
312            old_ids.isdisjoint(item["message_id"] for item in new["active_output"])
313        )
314        gates[1].set()
315        self.app.idle(session)
316        with self.app.api(f"/sessions/{session}?tail=20") as response:
317            self.assertEqual(json.load(response)["active_output"], [])
318
319    def test_retry_replaces_failed_attempt_output_before_reusing_indices(self):
320        self.gated_model_steps(first_completes=False)
321
322    def test_later_model_step_contains_only_its_own_output_segments(self):
323        self.gated_model_steps(first_completes=True)
324
325
326@exclusive
327class ActiveOutputFailureTest(unittest.TestCase):
328    def test_daemon_restart_removes_spill_owned_by_crashed_incarnation(self):
329        provider = Provider(lambda _: Reply("text", value="unused"))
330        self.addCleanup(provider.close)
331        with Albedo(
332            provider, prepare=lambda app: app.daemon.env.update(ALBEDO_INSPECT="1")
333        ) as app:
334            paused = PausedOutput(self, app, provider, ["crashed prefix " * 10000])
335            reference = next(
336                item["reference"]
337                for item in paused.snapshot()["active_output"]
338                if item["reference"] is not None
339            )
340            app.restart(crash=True)
341            paused.release.set()
342            with self.assertRaises(urllib.error.HTTPError) as caught:
343                reference_text(app, reference)
344            # A daemon restart rotates the signing secret, invalidating old tokens.
345            self.assertEqual(caught.exception.code, 400)
346            content_id = urllib.parse.urlsplit(reference["url"]).path.rsplit("/", 1)[1]
347            probe = ActiveOutputProbe(app)
348            probe.call("expire", app.home, content_id)
349            probe.call("sweep", app.home)
350            self.assertEqual(list((app.home / "active-output").glob("*.data")), [])
351
352    def test_expired_retired_lease_is_unreadable_and_swept(self):
353        provider = Provider(lambda _: Reply("text", value="unused"))
354        self.addCleanup(provider.close)
355        with Albedo(
356            provider, prepare=lambda app: app.daemon.env.update(ALBEDO_INSPECT="1")
357        ) as app:
358            paused = PausedOutput(self, app, provider, ["expiry prefix " * 10000])
359            reference = next(
360                item["reference"]
361                for item in paused.snapshot()["active_output"]
362                if item["reference"] is not None
363            )
364            content_id = urllib.parse.urlsplit(reference["url"]).path.rsplit("/", 1)[1]
365            paused.finish(self)
366            probe = ActiveOutputProbe(app)
367            probe.call("expire", app.home, content_id)
368            with self.assertRaises(urllib.error.HTTPError) as caught:
369                reference_text(app, reference)
370            self.assertEqual(caught.exception.code, 410)
371            probe.call("sweep", app.home)
372            self.assertFalse(
373                (app.home / "active-output" / f"{content_id}.data").exists()
374            )
375            self.assertFalse(
376                (app.home / "active-output" / f"{content_id}.meta").exists()
377            )
378
379    def test_deleted_session_revokes_snapshot_and_removes_spill(self):
380        provider = Provider(lambda _: Reply("text", value="unused"))
381        self.addCleanup(provider.close)
382        with Albedo(provider) as app:
383            paused = PausedOutput(self, app, provider, ["delete prefix " * 20000])
384            reference = next(
385                item["reference"]
386                for item in paused.snapshot()["active_output"]
387                if item["reference"] is not None
388            )
389            app.api(
390                f"/sessions/{paused.session}/inputs/{paused.input_id}/cancel", {}
391            ).close()
392            paused.finish(self)
393            resource = f"/sessions/{paused.session}?view=configuration"
394            with app.api(resource) as response:
395                revision = response.getheader("ETag")
396                response.read()
397            app.api(resource, method="DELETE", headers={"If-Match": revision}).close()
398            paused.release.set()
399            with self.assertRaises(urllib.error.HTTPError) as caught:
400                reference_text(app, reference)
401            self.assertEqual(caught.exception.code, 410)
402            self.assertEqual(list((app.home / "active-output").glob("*.data")), [])
403
404    def test_spill_read_failure_is_visible_but_durable_commit_survives(self):
405        provider = Provider(lambda _: Reply("text", value="unused"))
406        self.addCleanup(provider.close)
407        with Albedo(provider) as app:
408            paused = PausedOutput(self, app, provider, ["read failure prefix " * 10000])
409            reference = next(
410                item["reference"]
411                for item in paused.snapshot()["active_output"]
412                if item["reference"] is not None
413            )
414            content_id = urllib.parse.urlsplit(reference["url"]).path.rsplit("/", 1)[1]
415            spill = app.home / "active-output" / f"{content_id}.data"
416            spill.unlink()
417            spill.mkdir()
418            with self.assertRaises(urllib.error.HTTPError) as caught:
419                reference_text(app, reference)
420            self.assertEqual(caught.exception.code, 503)
421            paused.finish(self)
422            self.assertEqual(paused.snapshot()["active_output"], [])
423            self.assertTrue(
424                any(event["type"] == "committed" for event in paused.events)
425            )
426
427    def test_spill_creation_failure_does_not_interrupt_generation(self):
428        provider = Provider(lambda _: Reply("text", value="unused"))
429        self.addCleanup(provider.close)
430
431        def prepare(app):
432            (app.home / "active-output").write_text("injected creation failure")
433
434        with Albedo(provider, prepare=prepare) as app:
435            paused = PausedOutput(self, app, provider, ["unavailable spill " * 12000])
436            with self.assertRaises(urllib.error.HTTPError) as caught:
437                paused.snapshot()
438            self.assertEqual(caught.exception.code, 503)
439            paused.finish(self)
440            snapshot = paused.snapshot()
441            self.assertEqual(snapshot["active_output"], [])
442            assistant = [
443                entry
444                for entry in snapshot["history"]["items"]
445                if entry["kind"] == "assistant"
446            ]
447            self.assertTrue(assistant, snapshot)