"""Deleting a session removes every row any table keeps for it. The delete route clears a list of tables the daemon names itself, so a table an extension adds is forgotten until someone remembers to add it. The check reads the schema, so it covers a table nobody has listed yet. """ from contextlib import closing import json from pathlib import Path import sqlite3 import threading import unittest import urllib.parse import urllib.error from harness import ( Albedo, Provider, alive, exclusive, operation_id, python, text, wait_until, ) # Paperclip provenance and retained input outcomes outlive the session. KEPT = {"paperclips", "input_turns"} def reply(request): return text("older conversation summary") class SessionDeleteTests(unittest.TestCase): def setUp(self): self.provider = Provider(reply) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) def database(self): return closing( sqlite3.connect(f"file:{self.app.home / 'albedo.sqlite'}?mode=ro", uri=True) ) def keyed_tables(self, db): tables = [ row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'") ] return [ table for table in tables if table not in KEPT and any( column[1] == "session" for column in db.execute(f"PRAGMA table_info({table})") ) ] def rows(self, db, table, session): return db.execute( f"SELECT count(*) FROM {table} WHERE session=?", (session,) ).fetchone()[0] def compacted_session(self, strategy): session = self.app.session() # Long enough that a forced compaction has something to evict. for prompt in ("first", "second", "third", "fourth"): self.app.prompt(session, prompt + " " + "x" * 6000).close() self.app.idle(session) self.compact(session, strategy) return session def compact(self, session, strategy): with self.app.api( f"/sessions/{session}/compaction", {"strategy": strategy}, ) as response: self.assertEqual(json.load(response)["state"], "compacted") self.app.idle(session) def delete_session(self, session): resource = f"/sessions/{session}?view=configuration" with self.app.api(resource) as response: validator = response.getheader("ETag") response.read() with self.app.api( resource, method="DELETE", headers={"If-Match": validator} ) as response: self.assertEqual(json.load(response)["state"], "complete") def assert_delete_clears(self, strategy, table, then=None): session = self.compacted_session(strategy) with self.app.api( "/extensions/schedule/jobs", { "session_id": session, "kind": "once", "delay_seconds": 3600, "prompt": "build", }, ) as response: response.read() with self.app.api( "/extensions/work/items?" + urllib.parse.urlencode({"workspace": str(self.app.workspace)}), {"title": "assigned", "session_id": session}, ) as response: response.read() with self.database() as db: for owned_table in ("work", "schedules"): self.assertGreater(self.rows(db, owned_table, session), 0, owned_table) self.assertGreater(self.rows(db, table, session), 0, table) self.assertIn(table, self.keyed_tables(db)) if then: self.compact(session, then) with self.database() as db: self.assertGreater(self.rows(db, table, session), 0, table) self.delete_session(session) with self.database() as db: left = { name: self.rows(db, name, session) for name in self.keyed_tables(db) } self.assertEqual({k: v for k, v in left.items() if v}, {}) orphans = db.execute( "SELECT count(*) FROM lcm_compaction_edge WHERE child NOT IN " "(SELECT id FROM lcm_compaction_node) OR parent NOT IN " "(SELECT id FROM lcm_compaction_node)" ).fetchone()[0] self.assertEqual(orphans, 0) for owned_table in ("work", "schedules"): self.assertEqual(self.rows(db, owned_table, session), 0, owned_table) def test_leaf_deletion_refuses_busy_and_stale_observations(self): entered, release = threading.Event(), threading.Event() self.addCleanup(release.set) def held(request): entered.set() release.wait(30) return text("completed before deletion") self.provider.script = held session = self.app.session() resource = f"/sessions/{session}?view=configuration" with self.app.api(resource) as response: revision = response.headers["ETag"] response.read() identity = operation_id() with self.app.api( f"/sessions/{session}/inputs/{identity}", {"kind": "message", "text": "hold deletion"}, method="PUT", ) as response: response.read() self.assertTrue(entered.wait(30)) with self.app.api(resource) as response: revision = response.headers["ETag"] response.read() with self.assertRaises(urllib.error.HTTPError) as caught: self.app.api(resource, method="DELETE", headers={"If-Match": revision}) self.assertEqual(caught.exception.code, 409) release.set() self.app.idle(session) with self.app.api(resource) as response: revision = response.headers["ETag"] response.read() with self.app.api( resource, {"name": "changed before deletion"}, method="PATCH", headers={"If-Match": revision}, ) as response: response.read() with self.assertRaises(urllib.error.HTTPError) as caught: self.app.api(resource, method="DELETE", headers={"If-Match": revision}) self.assertEqual(caught.exception.code, 412) with self.app.api(f"/sessions/{session}") as response: self.assertEqual(json.load(response)["name"], "changed before deletion") self.delete_session(session) with self.app.api(f"/sessions/{session}/inputs/{identity}") as response: receipt = json.load(response) self.assertEqual(receipt["delivery"], "committed") self.assertEqual(receipt["turn"]["state"], "completed") with self.assertRaises(urllib.error.HTTPError) as caught: self.app.api(f"/sessions/{session}") self.assertEqual(caught.exception.code, 410) self.assertEqual(json.load(caught.exception)["code"], "session_deleted") def test_deletion_clears_compaction_work_and_schedule_rows(self): for strategy, table, then in ( ("rolling", "compaction_notes", None), ("lcm", "lcm_compaction_node", None), ("lcm", "lcm_compaction_node", "rolling"), ("snapcompact", "snapcompact_archive", None), ): with self.subTest(strategy=strategy, then=then): self.assert_delete_clears(strategy, table, then=then) @exclusive def test_snapcompact_selection_keeps_archive_across_switch_and_restart(self): self.app.write_extensions({"snapcompact": {"enabled": False}}) self.app.restart() session = self.compacted_session("snapcompact") def selected(): with self.app.api(f"/sessions/{session}?tail=0") as response: return json.load(response)["selection"]["effective"]["extensions"] def archive(): with self.database() as db: return db.execute( "SELECT * FROM snapcompact_archive WHERE session=?", (session,) ).fetchall() self.assertTrue(selected()["snapcompact"]) self.assertFalse(selected()["rolling"]) saved = archive() self.assertTrue(saved) self.compact(session, "rolling") self.assertTrue(selected()["rolling"]) self.assertFalse(selected()["snapcompact"]) self.assertEqual(archive(), saved) self.app.restart() self.assertTrue(selected()["rolling"]) self.assertFalse(selected()["snapcompact"]) self.assertEqual(archive(), saved) # Holds a cold-booted kernel at interpreter startup while the gate file exists. BOOT_GATE = """import os, sys, time gate = os.environ.get("TEST_KERNEL_BOOT_GATE") argv = " ".join(getattr(sys, "orig_argv", sys.argv)) if gate and "albedo_kernel.py" in argv: with open(gate + ".pid", "w") as out: out.write(str(os.getpid())) while os.path.exists(gate): time.sleep(0.02) """ STUBBORN = ( "from pathlib import Path\nimport time\n" "Path('stubborn-started').write_text('x')\n" "end = time.monotonic() + 4\n" "while time.monotonic() < end:\n" " try:\n time.sleep(0.05)\n" " except KeyboardInterrupt:\n pass" ) # exclusive: cold boots without spares or the zygote, through a user site hook @exclusive class KernelDeleteTests(unittest.TestCase): gate: Path code: str def setUp(self): self.provider = Provider( lambda request: ( text("done") if request["messages"][-1].get("role") == "tool" else python(self.code) ) ) self.addCleanup(self.provider.close) def prepare(app): base = app.root / "userbase" for site in ["lib/python/site-packages"] + [ f"lib/python3.{minor}/site-packages" for minor in range(9, 16) ]: (base / site).mkdir(parents=True) (base / site / "usercustomize.py").write_text(BOOT_GATE) self.gate = app.root / "boot-gate" self.gate.write_text("held") # The nix dev shell turns the user site off. app.env.pop("PYTHONNOUSERSITE", None) app.env.update( PYTHONUSERBASE=str(base), TEST_KERNEL_BOOT_GATE=str(self.gate), ALBEDO_KERNEL_SPARES="0", ALBEDO_KERNEL_ZYGOTE="0", ) self.app = Albedo(self.provider, prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) def test_deleting_a_session_whose_kernel_is_booting_waits_for_the_boot(self): self.code = "print('never needed')" session = self.app.session() self.app.prompt(session, "start the kernel").close() pidfile = self.gate.with_suffix(".pid") wait_until(pidfile.exists, 30, "the kernel never started booting") outcome = {} deleting = threading.Thread( target=lambda: outcome.update(self.app.delete_tree(session)) ) deleting.start() deleting.join(1) self.assertTrue(deleting.is_alive(), f"delete did not wait: {outcome}") self.gate.unlink() deleting.join(30) self.assertEqual(outcome.get("state"), "complete", outcome) kernel = int(pidfile.read_text()) wait_until(lambda: not alive(kernel), 15, "the booted kernel survived") def test_deleting_right_after_a_forced_cell_stop_succeeds(self): self.gate.unlink() self.code = STUBBORN session = self.app.session() self.app.prompt(session, "run a stubborn cell").close() wait_until( (self.app.workspace / "stubborn-started").exists, 30, "cell never ran" ) with self.app.api(f"/sessions/{session}?tail=0") as response: state = json.load(response) self.app.api( f"/sessions/{session}/interrupt", { "run_id": state["status"]["run_id"], "through_input_order": state["input_order"], }, ).close() self.app.idle(session) outcome = self.app.delete_tree(session) self.assertEqual(outcome["state"], "complete", outcome) if __name__ == "__main__": unittest.main()