okami.mom/albedo

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

nekomimi.petsession: publish text and thinking deltas once a frame, cheaper per-delta bookkeeping, replay window fits one response6b74a53

main
branchesmainall refs →
8.0 KiB191 linesraw
1"""A replay cursor cannot skip durable events from a replacement session actor."""
2
3import json
4import sqlite3
5import unittest
6import urllib.error
7
8from inspect_support import Inspection
9from harness import Albedo, Provider, ROOT, Reply, exclusive, text
10
11
12def restart_actor(app, session):
13    inspection = Inspection(app, ROOT / "test/e2e/albedo_stream_replay_probe.erl")
14    result = inspection.call("restart", f"[<<{json.dumps(session)}>>]")
15    if result != "actor_stopped":
16        raise AssertionError(result)
17
18
19def interleaved(pairs):
20    for _ in range(pairs):
21        for delta in [{"reasoning_content": "thought "}, {"content": "answer "}]:
22            yield {
23                "id": "fixture",
24                "choices": [{"index": 0, "delta": delta, "finish_reason": None}],
25            }
26    yield {
27        "id": "fixture",
28        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
29    }
30
31
32class StreamReplayTest(unittest.TestCase):
33    def setUp(self):
34        self.provider = Provider(lambda _: text("answer"))
35        self.addCleanup(self.provider.close)
36
37    def turn(self, app, session, prompt):
38        app.prompt(session, prompt).close()
39        app.idle(session)
40
41    def assert_reset_history(self, page, prompts):
42        self.assertEqual(page["events"][0]["type"], "reset")
43        self.assertEqual(
44            [
45                part["text"]
46                for entry in page["snapshot"]["history"]["items"]
47                if entry["kind"] == "user"
48                for part in entry["content"]
49                if part["kind"] == "text"
50            ],
51            prompts,
52        )
53
54    # exclusive: enables inspection and loads a deliberately malformed probe
55    @exclusive
56    def test_inspection_reports_rpc_and_invalid_result_failures(self):
57        with Albedo(
58            self.provider,
59            prepare=lambda app: app.daemon.env.update(ALBEDO_INSPECT="1"),
60        ) as app:
61            inspection = Inspection(app)
62            with self.assertRaisesRegex(AssertionError, "rpc_failed"):
63                inspection.evaluate("Call(albedo_missing_probe, missing, [])")
64            source = app.root / "albedo_invalid_probe.erl"
65            source.write_text(
66                "-module(albedo_invalid_probe).\n-export([answer/0]).\n"
67                'answer() -> <<"not JSON">>.\n'
68            )
69            with self.assertRaisesRegex(AssertionError, "returned invalid JSON"):
70                Inspection(app, source).call_json("answer")
71
72    # exclusive: enables daemon inspection at startup
73    @exclusive
74    def test_old_actor_cursor_inside_replacement_window_resets_durable_history(self):
75        def prepare(app):
76            app.daemon.env["ALBEDO_INSPECT"] = "1"
77
78        with Albedo(self.provider, prepare=prepare) as app:
79            session = app.session()
80            self.turn(app, session, "before actor restart")
81            old = app.stream_page(session)
82            restart_actor(app, session)
83            self.turn(app, session, "after actor restart")
84            current = app.stream_page(session)
85            replay = app.stream_page(session, old)
86            self.assert_reset_history(
87                replay, ["before actor restart", "after actor restart"]
88            )
89            self.assertNotEqual(replay["generation"], old["generation"])
90            self.assertEqual(replay["generation"], current["generation"])
91
92    # exclusive: restarts the daemon
93    @exclusive
94    def test_old_daemon_cursor_resets_durable_history(self):
95        with Albedo(self.provider) as app:
96            session = app.session()
97            self.turn(app, session, "before daemon restart")
98            old = app.stream_page(session)
99            app.restart(crash=True)
100            self.turn(app, session, "after daemon restart")
101            current = app.stream_page(session)
102            replay = app.stream_page(session, old)
103            self.assert_reset_history(
104                replay, ["before daemon restart", "after daemon restart"]
105            )
106            self.assertNotEqual(replay["generation"], old["generation"])
107            self.assertEqual(replay["generation"], current["generation"])
108
109    def test_same_generation_replays_only_subsequent_events_in_order(self):
110        with Albedo(self.provider) as app:
111            session = app.session()
112            self.turn(app, session, "already consumed")
113            old = app.stream_page(session)
114            self.turn(app, session, "next prompt")
115            self.turn(app, session, "last prompt")
116            replay = app.stream_page(session, old)
117            self.assertEqual(replay["generation"], old["generation"])
118            self.assertGreater(replay["cursor"], old["cursor"])
119            self.assertNotIn("reset", [event["type"] for event in replay["events"]])
120            self.assertEqual(
121                [
122                    part["text"]
123                    for event in replay["events"]
124                    if event["type"] == "message"
125                    and event["data"]["entry"]["kind"] == "user"
126                    for part in event["data"]["entry"]["content"]
127                    if part["kind"] == "text"
128                ],
129                ["next prompt", "last prompt"],
130            )
131            keepalive = app.stream_page(session, replay)
132            self.assertEqual(keepalive["events"], [])
133            self.assertEqual(keepalive["generation"], replay["generation"])
134            self.assertEqual(keepalive["cursor"], replay["cursor"])
135
136    # exclusive: renames the global transcript table
137    @exclusive
138    def test_failed_transcript_read_emits_failure_without_replacement_cursor(self):
139        with Albedo(self.provider) as app:
140            session = app.session()
141            self.turn(app, session, "committed history")
142            old = app.stream_page(session)
143            other_generation = app.stream_page(app.session())["generation"]
144            with sqlite3.connect(app.home / "albedo.sqlite") as database:
145                database.execute(
146                    "ALTER TABLE transcript RENAME TO unavailable_transcript"
147                )
148            with self.assertRaises(urllib.error.HTTPError) as caught:
149                app.api(
150                    f"/sessions/{session}?after_generation={other_generation}&after_seq={old['cursor']}&tail=2",
151                    headers={"Accept": "text/event-stream"},
152                )
153            self.assertEqual(caught.exception.code, 503)
154            problem = json.load(caught.exception)
155            self.assertEqual(problem["code"], "request_unavailable")
156            self.assertNotIn("snapshot", problem)
157            self.assertNotIn("generation", problem)
158            self.assertNotIn("cursor", problem)
159
160    def test_evicted_cursor_resets_in_same_generation_and_initial_tail_pages_history(
161        self,
162    ):
163        with Albedo(self.provider) as app:
164            session = app.session()
165            self.turn(app, session, "oldest prompt")
166            old = app.stream_page(session)
167            # Alternating thinking and text never merge into one event, so a
168            # long interleaved answer evicts the count-limited replay window.
169            self.provider.script = lambda _: Reply("text", events=interleaved(150))
170            self.turn(app, session, "newest prompt")
171            replay = app.stream_page(session, old)
172            self.assertEqual(replay["generation"], old["generation"])
173            self.assert_reset_history(replay, ["oldest prompt", "newest prompt"])
174            initial = app.stream_page(session, tail=3)
175            self.assert_reset_history(initial, ["newest prompt"])
176            history = initial["snapshot"]["history"]
177            self.assertIsNotNone(history["older"])
178            with app.api(
179                f"/sessions/{session}/history?limit=3&next={history['older']}"
180            ) as response:
181                older = json.load(response)
182            self.assertEqual(
183                [
184                    part["text"]
185                    for entry in older["items"]
186                    if entry["kind"] == "user"
187                    for part in entry["content"]
188                    if part["kind"] == "text"
189                ],
190                ["oldest prompt"],
191            )