Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234"""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 anextension adds is forgotten until someone remembers to add it. The check readsthe schema, so it covers a table nobody has listed yet."""
import jsonimport sqlite3import threadingimport unittestimport urllib.parseimport urllib.error
from harness import Albedo, Provider, exclusive, operation_id, text
# 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 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.database() as db: 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)
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_rolling_and_notes_rows_go_with_the_session(self): self.assert_delete_clears("rolling", "compaction_notes")
def test_lcm_rows_go_with_the_session(self): self.assert_delete_clears("lcm", "lcm_compaction_node")
def test_rows_of_a_strategy_switched_off_go_too(self): # Switching to rolling disables lcm; its graph still belongs to the session. self.assert_delete_clears("lcm", "lcm_compaction_node", then="rolling")
def test_work_and_schedule_rows_go_with_the_session(self): session = self.app.session() 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 table in ("work", "schedules"): self.assertGreater(self.rows(db, table, session), 0, table) self.delete_session(session) with self.database() as db: for table in ("work", "schedules"): self.assertEqual(self.rows(db, table, session), 0, table)
def test_snapcompact_archive_goes_with_the_session(self): self.assert_delete_clears("snapcompact", "snapcompact_archive")
@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)
if __name__ == "__main__": unittest.main()