okami.mom/albedo

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

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca

main
branchesmainall refs →
11.6 KiB273 linesraw
1"""Retention preserves pending and unfinished turns, and deleted identities cannot revive work."""
2
3import json
4import sqlite3
5import threading
6import time
7import urllib.error
8
9from harness import Albedo, Provider, exclusive, operation_id, text
10from operations_test import InputScenario
11
12RETENTION_MS = 7 * 24 * 60 * 60 * 1000
13
14
15@exclusive
16class OperationRetentionTests(InputScenario):
17    def setUp(self):
18        self.provider = Provider(lambda _: text("done"))
19        self.addCleanup(self.provider.close)
20        self.app = Albedo(
21            self.provider,
22            protocol="responses",
23            prepare=lambda app: app.env.update(ALBEDO_SCHEDULE_TICK_MS="50"),
24        )
25        self.app.__enter__()
26        self.addCleanup(self.app.__exit__, None, None, None)
27
28    def database(self):
29        return sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10)
30
31    def wait_expiry(self, session, input_id, status):
32        deadline = time.monotonic() + 10
33        while time.monotonic() < deadline:
34            try:
35                self.receipt(session, input_id)
36            except urllib.error.HTTPError as failure:
37                self.assertEqual(failure.code, status)
38                return
39            time.sleep(0.05)
40        self.fail("terminal receipt did not expire")
41
42    def delete(self, session):
43        with self.app.api(f"/sessions/{session}?view=configuration") as response:
44            validator = response.getheader("ETag")
45            response.read()
46        with self.app.api(
47            f"/sessions/{session}?view=configuration",
48            method="DELETE",
49            headers={"If-Match": validator},
50        ) as response:
51            outcome = json.load(response)
52            self.assertEqual(outcome["state"], "complete", outcome)
53
54    def test_terminal_receipt_expiry_prevents_reexecution(self):
55        session = self.app.session()
56        input_id = operation_id()
57        intent = {"kind": "message", "text": "terminal input"}
58        self.put_input(session, input_id, intent)
59        self.committed(session, input_id)
60        self.app.idle(session)
61        old_time = int(time.time() * 1000) - RETENTION_MS - 10000
62        aged_id = operation_id(old_time)
63        with self.database() as database:
64            database.execute(
65                "UPDATE operations SET id=?,created_at=?,terminal_at=? WHERE id=?",
66                (aged_id, old_time, old_time, input_id),
67            )
68        self.wait_expiry(session, aged_id, 410)
69        with self.assertRaises(urllib.error.HTTPError) as failure:
70            self.put_input(session, aged_id, intent)
71        self.assertEqual(failure.exception.code, 410)
72        self.assertEqual(json.load(failure.exception)["code"], "identity_expired")
73        self.assertEqual(len(self.users(session)), 1)
74        self.assertEqual(len(self.provider.requests), 1)
75        with self.app.api(f"/sessions/{session}") as response:
76            self.assertIsNotNone(json.load(response)["creation"])
77
78    def test_aged_pending_receipt_remains_recoverable_after_restart(self):
79        session = self.app.session()
80        input_id = operation_id()
81        intent = {"kind": "message", "text": "old pending input"}
82        with self.database() as database:
83            database.execute(
84                "CREATE TRIGGER hold_pending BEFORE INSERT ON transcript WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END"
85            )
86        accepted = self.put_input(session, input_id, intent)
87        self.assertEqual(self.receipt(session, input_id)["delivery"], "pending")
88        old_time = int(time.time() * 1000) - RETENTION_MS - 10000
89        aged_id = operation_id(old_time)
90
91        def age_pending(_):
92            with self.database() as database:
93                payload = database.execute(
94                    "SELECT payload FROM pending_inputs WHERE operation_id=?",
95                    (input_id,),
96                ).fetchone()[0]
97                payload = payload.replace(input_id.encode(), aged_id.encode())
98                response = database.execute(
99                    "SELECT response FROM operations WHERE id=?", (input_id,)
100                ).fetchone()[0]
101                response = response.replace(input_id, aged_id)
102                database.execute(
103                    "UPDATE operations SET id=?,created_at=?,response=? WHERE id=?",
104                    (aged_id, old_time, response, input_id),
105                )
106                database.execute(
107                    "UPDATE pending_inputs SET operation_id=?,payload=?,accepted_at=? WHERE operation_id=?",
108                    (aged_id, payload, old_time, input_id),
109                )
110
111        self.app.restart(crash=True, prepare=age_pending)
112        self.assertEqual(self.receipt(session, aged_id)["delivery"], "pending")
113        self.assertEqual(
114            self.put_input(session, aged_id, intent)["acceptance_order"],
115            accepted["acceptance_order"],
116        )
117        with self.database() as database:
118            database.execute("DROP TRIGGER hold_pending")
119        self.committed(session, aged_id)
120        self.app.idle(session)
121        self.app.restart(crash=True)
122        self.assertEqual(self.receipt(session, aged_id)["delivery"], "committed")
123        self.assertEqual(
124            [entry["input_id"] for entry in self.users(session)], [aged_id]
125        )
126
127    def test_unfinished_consumed_turn_does_not_start_receipt_retention(self):
128        session = self.app.session()
129        completed_id = operation_id()
130        self.put_input(
131            session, completed_id, {"kind": "message", "text": "completed control"}
132        )
133        self.committed(session, completed_id)
134        self.app.idle(session)
135        started, release = threading.Event(), threading.Event()
136        self.addCleanup(release.set)
137
138        def held_reply(_):
139            started.set()
140            if not release.wait(30):
141                raise AssertionError("unfinished turn was not released")
142            return text("finished held turn")
143
144        self.provider.script = held_reply
145        held_id = operation_id()
146        self.put_input(
147            session, held_id, {"kind": "message", "text": "unfinished control"}
148        )
149        self.assertTrue(started.wait(10))
150        self.assertEqual(self.receipt(session, held_id)["delivery"], "committed")
151        old_time = int(time.time() * 1000) - RETENTION_MS - 10000
152        with self.database() as database:
153            database.execute(
154                "UPDATE operations SET terminal_at=? WHERE id=?",
155                (old_time, completed_id),
156            )
157            database.execute(
158                "UPDATE operations SET created_at=? WHERE id=?", (old_time, held_id)
159            )
160            self.assertIsNone(
161                database.execute(
162                    "SELECT terminal_at FROM operations WHERE id=?", (held_id,)
163                ).fetchone()[0]
164            )
165        # Expiration of the completed control proves the real prune pass ran.
166        self.wait_expiry(session, completed_id, 404)
167        self.assertEqual(self.receipt(session, held_id)["turn"]["state"], "running")
168        with self.database() as database:
169            self.assertIsNone(
170                database.execute(
171                    "SELECT terminal_at FROM operations WHERE id=?", (held_id,)
172                ).fetchone()[0]
173            )
174        release.set()
175        self.app.idle(session)
176        self.assertEqual(self.receipt(session, held_id)["turn"]["state"], "completed")
177        with self.database() as database:
178            self.assertGreater(
179                database.execute(
180                    "SELECT terminal_at FROM operations WHERE id=?", (held_id,)
181                ).fetchone()[0],
182                old_time,
183            )
184
185    def test_deleted_session_keeps_creation_and_cancelled_input_receipts(self):
186        session = self.app.session()
187        with self.app.api(f"/sessions/{session}") as response:
188            creation = json.load(response)["creation"]
189        with self.database() as database:
190            database.execute(
191                "CREATE TRIGGER hold_pending BEFORE INSERT ON transcript WHEN NEW.row_class='user' BEGIN SELECT RAISE(ABORT,'fixture'); END"
192            )
193        input_id = operation_id()
194        intent = {"kind": "message", "text": "cancel when deleted"}
195        accepted = self.put_input(session, input_id, intent)
196        deadline = time.monotonic() + 10
197        while time.monotonic() < deadline:
198            if self.receipt(session, input_id)["blocking_reason"]:
199                break
200            time.sleep(0.03)
201        else:
202            self.fail("input did not reach blocked consumption")
203        self.delete(session)
204        self.assertEqual(self.receipt(session, input_id)["delivery"], "cancelled")
205        self.app.restart(crash=True)
206        with self.assertRaises(urllib.error.HTTPError) as lookup:
207            self.app.api(f"/sessions/{session}")
208        self.assertEqual(lookup.exception.code, 410)
209        self.assertEqual(json.load(lookup.exception)["decision"]["creation"], creation)
210        with self.assertRaises(urllib.error.HTTPError) as retry_creation:
211            self.app.api(
212                f"/sessions/{session}",
213                creation["submitted"],
214                method="PUT",
215                headers={"If-None-Match": "*"},
216            )
217        self.assertEqual(retry_creation.exception.code, 410)
218        replay = self.put_input(session, input_id, intent)
219        self.assertEqual(replay["acceptance_order"], accepted["acceptance_order"])
220        self.assertEqual(replay["delivery"], "cancelled")
221        with self.database() as database:
222            self.assertEqual(
223                database.execute("SELECT count(*) FROM sessions").fetchone()[0], 0
224            )
225            self.assertEqual(
226                database.execute("SELECT count(*) FROM pending_inputs").fetchone()[0], 0
227            )
228
229    def test_continuation_marker_survives_receipt_pruning_and_restart(self):
230        session = self.app.session()
231        self.app.prompt(session, "start").close()
232        self.app.idle(session)
233        input_id = operation_id()
234        self.put_input(session, input_id, {"kind": "continue"})
235        self.committed(session, input_id)
236        self.app.idle(session)
237        old_time = int(time.time() * 1000) - RETENTION_MS - 10000
238        with self.database() as database:
239            database.execute(
240                "UPDATE operations SET terminal_at=? WHERE id=?", (old_time, input_id)
241            )
242        self.wait_expiry(session, input_id, 404)
243        self.app.restart(crash=True)
244        with self.database() as database:
245            marker = database.execute(
246                "SELECT session,seq FROM continuation_markers WHERE operation_id=?",
247                (input_id,),
248            ).fetchall()
249            self.assertEqual(len(marker), 1)
250            self.assertEqual(marker[0][0], session)
251            self.assertGreater(marker[0][1], 0)
252        self.assertEqual(len(self.users(session)), 1)
253        self.assertEqual(len(self.provider.requests), 2)
254        deadline = time.monotonic() + 15
255        while time.monotonic() < deadline:
256            with self.app.api(f"/sessions/{session}?tail=0") as response:
257                kernel = json.load(response)["kernel"]
258            if kernel["state"] == "attached":
259                break
260            time.sleep(0.05)
261        else:
262            self.fail(
263                f"the retained kernel did not finish reattachment before deletion: {kernel}"
264            )
265        self.delete(session)
266        with self.database() as database:
267            self.assertEqual(
268                database.execute(
269                    "SELECT count(*) FROM continuation_markers WHERE operation_id=?",
270                    (input_id,),
271                ).fetchone()[0],
272                0,
273            )