From a5b3cd8a7fc2350c20a396b306efb9e628d4a42a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 19:00:51 -0400 Subject: [PATCH 1/3] Durable stores derive their declared schema on the writer's open A durable store refused any stored schema that differed from its declaration, so every schema change needed a hand-written carry step, and the step and the class drifted: c99becf13 edited carry_problem_numbers after the bus had run it, and install 46 left todos.sqlite3 refusing every write (the bridge's stale state hit the same refusal). DurableTables.reconcile_schema now carries the file onto its declaration inside one write transaction wherever the declaration says how: a row's declared earlier table name (TypedTable.renamed_from) is renamed to it, indexes, triggers and views are rebuilt from their declared SQL, and missing declared objects are created. A table whose columns differ is still refused with its rows untouched. The catalog kinds moved from carry_declared_schema into typed_table, which now uses them, and carry_problem_numbers is deleted: ProblemAllocation declares it was problem_number. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/cli_bridges/shared.py | 2 +- src/agent_comms/goal_history.py | 7 +- src/agent_comms/todos.py | 1 + src/agent_comms/typed_table.py | 119 ++++++++++++++++++++++++- tests/test_commander_forks.py | 22 ++--- tests/test_typed_store_snapshot.py | 22 +++++ tools/cutover/carry_declared_schema.py | 51 ++--------- tools/cutover/carry_problem_numbers.py | 60 ------------- 8 files changed, 154 insertions(+), 130 deletions(-) delete mode 100644 tools/cutover/carry_problem_numbers.py diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index d151d2299e..7877412183 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -167,7 +167,7 @@ def open(cls, state: Path): with SQLiteFile.write_transaction(db): if not BridgeStateTable.stored_schema(db): BridgeStateTable.create_schema(db) - BridgeStateTable.reconcile_schema(db) + BridgeStateTable.reconcile_schema(db) yield db def bind(self, db: CoordinationStore) -> Binding: diff --git a/src/agent_comms/goal_history.py b/src/agent_comms/goal_history.py index b393153eff..904aaf6862 100644 --- a/src/agent_comms/goal_history.py +++ b/src/agent_comms/goal_history.py @@ -149,9 +149,10 @@ def _initialize(self) -> None: if os.name == "posix" and stat.S_IMODE(self.path.stat().st_mode) != 0o600: raise GoalHistoryError("Goal history database must be owner-only.") try: - with closing(self._connect()) as connection, self.write_transaction(connection): - if not GoalHistoryEntry.stored_schema(connection): - GoalHistoryEntry.create_schema(connection) + with closing(self._connect()) as connection: + with self.write_transaction(connection): + if not GoalHistoryEntry.stored_schema(connection): + GoalHistoryEntry.create_schema(connection) GoalHistoryEntry.reconcile_schema(connection) self._sync() except (OSError, sqlite3.Error) as error: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 8bc637a03a..64a8df8811 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -3325,6 +3325,7 @@ class ProblemAllocation(TypedTable): """ number: Annotated[int, Column(primary_key=True, auto_increment=True)] + renamed_from = ("problem_number",) @classmethod def reserve(cls, db: sqlite3.Connection, text: str) -> ProblemNumber: diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index a99efa4182..a7f8ba0b8c 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -588,6 +588,80 @@ def drop(self, db: sqlite3.Connection) -> None: db.execute(f"DROP TABLE {_identifier(self.name)}") +class CatalogKind(DeclaredFamily, affix="Kind"): + """What SQLite's catalog calls an entry; each kind owns how a schema carry treats it.""" + + holds_rows: ClassVar[bool] = False + named_by_coverage: ClassVar[bool] = False + + @classmethod + def drop(cls, db: sqlite3.Connection, name: str) -> None: + db.execute(f"DROP {cls.declared_name.upper()} IF EXISTS {_identifier(name)}") + + @classmethod + def derive(cls, db: sqlite3.Connection, item: SQLiteCatalogObject, + declared: Mapping[str, str]) -> None: + """An object that holds no rows is rebuilt from its declaration: one that differs from + it, or that nothing declares, is dropped, and the carry creates what is declared.""" + if declared.get(item.name) != item.sql: + cls.drop(db, item.name) + + +class TableKind(CatalogKind): + holds_rows = True + + @classmethod + def derive(cls, db: sqlite3.Connection, item: SQLiteCatalogObject, + declared: Mapping[str, str]) -> None: + """Rows are never rebuilt from a declaration: a table that differs stays, and the + caller's schema check refuses it.""" + + +class IndexKind(CatalogKind): + named_by_coverage = True + + +class TriggerKind(CatalogKind): + pass + + +class ViewKind(CatalogKind): + pass + + +@dataclass(frozen=True) +class SQLiteCatalogObject(TypedRow): + """One catalog entry: SQLite's own type, name and table.""" + + type: type[CatalogKind] + name: str + tbl_name: str + sql: str + + def drop(self, db: sqlite3.Connection) -> None: + self.type.drop(db, self.name) + + @classmethod + def stored(cls, db: sqlite3.Connection) -> tuple[SQLiteCatalogObject, ...]: + return tuple(cls.read(db.execute( + "SELECT type,name,tbl_name,sql FROM sqlite_master " + "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' ORDER BY type,name"))) + + @classmethod + def carry(cls, db: sqlite3.Connection, declared: Mapping[str, str], + compared: frozenset[str]) -> None: + """Inside the caller's write transaction, the compared objects onto the declaration: + each kind derives its stored objects (rebuilt, or left for the caller's check), then + every declared object the file lacks is created from its declared SQL.""" + for item in cls.stored(db): + if item.name in compared: + item.type.derive(db, item, declared) + present = {item.name for item in cls.stored(db)} + for name, sql in declared.items(): + if name not in present: + db.execute(sql) + + class TableSchema(ABC): """The tables of one SQLite file, and the one reader of the SQL SQLite stored for them. @@ -646,10 +720,23 @@ def create_schema(cls, db: sqlite3.Connection) -> None: for row in cls.schema_rows(): row.create(db) + @classmethod + def derive_schema(cls, db: sqlite3.Connection) -> None: + """Inside the caller's write transaction, the stored schema onto the declaration + wherever the declaration says how: a row's declared earlier table name, an index, + trigger or view rebuilt from its SQL, a declared object the file lacks. What it cannot + say, a table whose columns differ, is refused with the rows untouched.""" + for row in cls.schema_rows(): + row.take_previous_name(db) + SQLiteCatalogObject.carry(db, cls.declared_schema(), + frozenset(item.name for item in cls.stored_schema(db))) + cls.require_schema(db) + @classmethod @abstractmethod def reconcile_schema(cls, db: sqlite3.Connection) -> None: - """The writer's answer to a stored schema that differs from the declaration.""" + """The writer's answer to a stored schema that differs from the declaration, in a + write transaction of its own.""" class RowMutation(Enum): @@ -728,13 +815,17 @@ def reconcile_schema(cls, db: sqlite3.Connection) -> None: class DurableTables(TableSchema): - """Rows nothing else can rebuild: a schema that differs is refused; only a carry moves it.""" + """Rows nothing else can rebuild: the writer carries a stored schema that differs onto + the declaration where the declaration says how, keeping every row, and refuses the rest. + No hand-written step repeats what a declaration already says.""" __slots__ = () @classmethod def reconcile_schema(cls, db: sqlite3.Connection) -> None: - cls.require_schema(db) + if cls.schema_differs(db): + with SQLiteFile.write_transaction(db): + cls.derive_schema(db) @lru_cache(maxsize=4) @@ -772,6 +863,9 @@ class TypedTable(TypedRow, DeclaredFamily, affix="Row", discovery_package="agent checks: ClassVar[tuple[str, ...]] = () indexes: ClassVar[tuple[Index, ...]] = () without_rowid: ClassVar[bool] = False + renamed_from: ClassVar[tuple[str, ...]] = () + """Table names this row's rows were stored under before; the writer renames the one it + finds, so a renamed declaration needs no hand-written step.""" @classmethod def _member_declarations(cls) -> tuple[type[DeclaredFamily], ...]: @@ -914,6 +1008,23 @@ def drop(cls, db: sqlite3.Connection) -> None: """Drop the declared table; SQLite drops its indexes with it.""" db.execute(f"DROP TABLE IF EXISTS {_identifier(cls.declared_name)}") + @classmethod + def take_previous_name(cls, db: sqlite3.Connection) -> None: + """Rows stored under a declared earlier name move to the declared one; rows under both + are refused, since nothing says which to keep.""" + stored = {table.name for table in SQLiteTableName.stored(db)} + match [name for name in cls.renamed_from if name in stored]: + case []: + pass + case [previous] if cls.declared_name not in stored: + db.execute(f"ALTER TABLE {_identifier(previous)} RENAME TO " + f"{_identifier(cls.declared_name)}") + case found: + holders = sorted({*found, cls.declared_name} & stored) + raise RelationViolationError( + f"{cls.declared_name}: tables {holders} could each hold its rows;" + " refusing to choose") + @classmethod def stored_columns(cls, db: sqlite3.Connection) -> tuple[str, ...]: """The columns the stored table has now, which a carry compares with the declaration.""" @@ -1243,7 +1354,7 @@ def _open_writer(self, *, check_same_thread: bool = True) -> sqlite3.Connection: with self.write_transaction(connection): if not self.stored_schema(connection): self.create_schema(connection) - # Under the attachment boundary: a derived file is rebuilt, a durable one refused. + # Under the attachment boundary: a derived file is rebuilt, a durable one carried. self.reconcile_schema(connection) if SQLiteJournalMode.read(connection.execute("PRAGMA journal_mode=WAL")) != [ SQLiteJournalMode("wal") diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index c5e7cabf34..23e83bcb01 100644 --- a/tests/test_commander_forks.py +++ b/tests/test_commander_forks.py @@ -238,9 +238,9 @@ def assign(name): print(refused.stderr.strip()) -def test_retained_store_carry_preserves_original_identities(fork_bus): - from cutover.carry_problem_numbers import ProblemNumbers - +def test_a_retained_store_takes_its_declared_schema_on_the_next_write(fork_bus): + """The writer's open derives what the declaration says: a missing table and indexes are + created, and rows stored under ProblemAllocation's declared earlier name move to it.""" comms, source, order = fork_bus store = comms.goals.todos retained = store.get('P40') @@ -248,25 +248,15 @@ def test_retained_store_carry_preserves_original_identities(fork_bus): db.execute(f'DROP TABLE "{ProblemAllocation.declared_name}"') db.execute(f'DROP INDEX "{Todo.unit_index.name(Todo)}"') db.execute(f'DROP INDEX "{Todo.problem_index.name(Todo)}"') - assert ProblemNumbers.carry(comms.root, apply=False).changed == 3 - assert ProblemNumbers.carry(comms.root, apply=True).changed == 3 - assert store.get('P40') == retained - with store.snapshot() as db: - assert ProblemAllocation.select(db) == [] assert store.allocate_problem() == 'P41' - assert ProblemNumbers.carry(comms.root, apply=True).changed == 0 - # Upgrade the installed old schema: keep unsent P41, delete only the P40 copy. + assert store.get('P40') == retained with store.writing() as db: db.execute(f'ALTER TABLE "{ProblemAllocation.declared_name}" RENAME TO problem_number') - db.execute('INSERT INTO problem_number(number) VALUES (40)') db.execute(f'DROP INDEX "{Todo.problem_index.name(Todo)}"') - assert ProblemNumbers.carry(comms.root, apply=False).changed == 2 - assert ProblemNumbers.carry(comms.root, apply=True).changed == 2 + assert store.allocate_problem() == 'P42' with store.snapshot() as db: - assert [row.number for row in ProblemAllocation.select(db)] == [41] + assert [row.number for row in ProblemAllocation.select(db)] == [41, 42] assert store.get('P40') == retained - assert store.allocate_problem() == 'P42' - assert ProblemNumbers.carry(comms.root, apply=True).changed == 0 def test_created_ids_live_only_on_todos_and_allocation_uses_numeric_index(fork_bus): diff --git a/tests/test_typed_store_snapshot.py b/tests/test_typed_store_snapshot.py index f765324ef7..0353528099 100644 --- a/tests/test_typed_store_snapshot.py +++ b/tests/test_typed_store_snapshot.py @@ -79,3 +79,25 @@ def test_existing_undeclared_schema_is_not_treated_as_missing(tmp_path): with store.snapshot(): pytest.fail("An existing invalid file must not become an empty store") assert store.path.stat().st_size == 0 + + +def test_a_durable_writer_derives_declared_objects_and_refuses_an_undeclared_column(tmp_path): + """A durable file's writer rebuilds what its declaration says (indexes and other + row-less objects, missing ones created); a table whose columns differ keeps its rows and is + refused, since the declaration cannot say what the extra column meant.""" + store = InputDispositions(tmp_path) + assert store.record("first", seq=None, owner="kid", claim_scope=1, target="kid", text="first") + declared = store.declared_schema() + with store.writing() as connection: + for name, sql in declared.items(): + if not sql.lstrip().upper().startswith("CREATE TABLE"): + connection.execute(f'DROP INDEX IF EXISTS "{name}"') + with store.writing() as connection: + assert not store.schema_differs(connection) + assert store.lookup("first").source_text == "first" + with store.writing() as connection: + connection.execute( + f'ALTER TABLE "{InputDispositionRow.declared_name}" ADD COLUMN extra TEXT DEFAULT "kept"') + with pytest.raises(RelationViolationError): + with store.writing(): + pytest.fail("A table whose columns differ from the declaration must be refused") diff --git a/tools/cutover/carry_declared_schema.py b/tools/cutover/carry_declared_schema.py index 8adf8c6ce4..44775e7c6a 100644 --- a/tools/cutover/carry_declared_schema.py +++ b/tools/cutover/carry_declared_schema.py @@ -21,16 +21,16 @@ from importlib import import_module from pathlib import Path from pkgutil import walk_packages -from typing import ClassVar import agent_comms from agent_comms.coordination_database import CoordinationStore -from agent_comms.declared_family import DeclaredFamily, UnknownFamilyMember +from agent_comms.declared_family import UnknownFamilyMember from agent_comms.errors import RelationViolationError from agent_comms.installer.carry import Carried, Carry from agent_comms.locked_store import DurableFile, SQLiteFile from agent_comms.typed_table import ( DerivedTables, + SQLiteCatalogObject, SQLiteSchemaObject, SQLiteUserVersion, TypedRow, @@ -62,45 +62,14 @@ def unnamed(sql: str) -> str: return _INDEX_NAME.sub(r"\1\2", sql, count=1) -class CatalogKind(DeclaredFamily, affix="Kind"): - """What SQLite's catalog calls an entry; each kind owns how the carry treats it.""" - - holds_rows: ClassVar[bool] = False - named_by_coverage: ClassVar[bool] = False - - -class TableKind(CatalogKind): - holds_rows = True - - -class IndexKind(CatalogKind): - named_by_coverage = True - - -class TriggerKind(CatalogKind): - pass - - -class ViewKind(CatalogKind): - pass - - @dataclass(frozen=True) -class StoredObject(TypedRow): - """One catalog entry: SQLite's own type, name and table.""" - - type: type[CatalogKind] - name: str - tbl_name: str - sql: str +class StoredObject(SQLiteCatalogObject): + """One catalog entry, as the legacy carry reads it: retired sealed copies included.""" @property def retired(self) -> bool: return self.type.holds_rows and self.name in RETIRED_TABLES - def drop(self, db) -> None: - db.execute(f'DROP {self.type.declared_name.upper()} IF EXISTS "{self.name}"') - def declaration(self, path: Path) -> type[TypedTable]: try: return TypedTable.member_named(self.name) @@ -119,12 +88,6 @@ def row(self, path: Path) -> type[TypedTable]: def render(self) -> str: return f"{self.type.declared_name} {self.name} on {self.tbl_name}: {self.sql}" - @classmethod - def stored(cls, db) -> tuple[StoredObject, ...]: - return tuple(cls.read(db.execute( - "SELECT type,name,tbl_name,sql FROM sqlite_master " - "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' ORDER BY type,name"))) - def derived(row: type[TypedTable]) -> bool: return any(row in kind.schema_rows() for kind in DerivedTables.schema_members()) @@ -213,14 +176,10 @@ def carry(path: Path) -> FileCarried: declared = {name: sql for row in rows for name, sql in row.schema_objects().items()} counts = RowCount.of(db, kept) retired = RowCount.of(db, [item for item in tables if item.retired]) - for item in before: - if not item.type.holds_rows and declared.get(item.name) != item.sql: - item.drop(db) for item in tables: if item.retired: item.drop(db) - for row in rows: - row.create(db, if_not_exists=True) + StoredObject.carry(db, declared, frozenset(item.name for item in before)) if differences := SQLiteSchemaObject.differences(TypedTable.objects_of(db, rows), declared): raise RelationViolationError(f"{path}: still differs: {differences}") if RowCount.of(db, kept) != counts: diff --git a/tools/cutover/carry_problem_numbers.py b/tools/cutover/carry_problem_numbers.py deleted file mode 100644 index c98808ba5b..0000000000 --- a/tools/cutover/carry_problem_numbers.py +++ /dev/null @@ -1,60 +0,0 @@ -"""Index original Todo identities and install allocation for unsent numbers. - -Delete this step once every retained bus has been carried. -""" - -from contextlib import closing -from pathlib import Path - -from agent_comms.coordination_database import CoordinationStore -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import SQLiteFile -from agent_comms.problem_number import ProblemNumber -from agent_comms.todos import ProblemAllocation, Todo, TodoStore - - -class ProblemNumbers(Carry): - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - paths = tuple(path for path in SQLiteFile.files(TodoStore.path_in(root)) if path.exists()) - for path in paths: - cls.require_confined(root, path) - return paths - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - path = TodoStore.path_in(root) - cls.require_confined(root, path) - if not path.exists(): - return Carried(0) - with closing(CoordinationStore.read_snapshot(path)) as db: - stored = {row.name for row in TodoStore.stored_schema(db)} - numbers_missing = ProblemAllocation.declared_name not in stored - migrate_numbers = numbers_missing and "problem_number" in stored - missing_indexes = tuple( - index - for index in (Todo.unit_index, Todo.problem_index) - if index.name(Todo) not in stored - ) - changed = int(numbers_missing) + len(missing_indexes) - if not changed: - return Carried(0) - created = tuple(ProblemNumber.parse(todo.id) for todo in Todo.stream(db)) - if not apply: - return Carried(changed) - with closing(CoordinationStore.writer(path)) as db, SQLiteFile.write_transaction(db): - if migrate_numbers: - # SQLite renames its sequence too; even unsent issued IDs stay reserved. - db.execute( - f'ALTER TABLE problem_number RENAME TO "{ProblemAllocation.declared_name}"' - ) - for number in created: - ProblemAllocation.delete(db, number=number.number) - elif numbers_missing: - ProblemAllocation.create(db) - for index in missing_indexes: - db.execute(index.sql(Todo)) - TodoStore.require_schema(db) - return Carried( - changed, rewritten=tuple(str(file.relative_to(root)) for file in SQLiteFile.files(path)) - ) From f33a348d82405a86a6616da80fe5f5131bd2fada Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 19:02:58 -0400 Subject: [PATCH 2/3] Delete the hand-written declared-schema step; proof journals convert through their declaration carry_declared_schema was a second owner of 'carry a file onto its declaration' beside DurableTables, and every bus has run it; its dry run also refused tables it did not know (unit-filebus 17008: a batch-derived table broke CarryChain.pending). Its test goes with it. carry_native_input_proofs now converts differing proof journals with NativeContextJournal.reconcile_schema, and its dry run counts them, since a step's dry run is what decides whether it runs. CatalogKind's two boolean flags had no reader left. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/typed_table.py | 7 +- tests/test_declared_schema_carry.py | 202 -------------------- tools/cutover/carry_declared_schema.py | 209 --------------------- tools/cutover/carry_native_input_proofs.py | 34 ++-- 4 files changed, 22 insertions(+), 430 deletions(-) delete mode 100644 tests/test_declared_schema_carry.py delete mode 100644 tools/cutover/carry_declared_schema.py diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index a7f8ba0b8c..1acdfd23ca 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -591,9 +591,6 @@ def drop(self, db: sqlite3.Connection) -> None: class CatalogKind(DeclaredFamily, affix="Kind"): """What SQLite's catalog calls an entry; each kind owns how a schema carry treats it.""" - holds_rows: ClassVar[bool] = False - named_by_coverage: ClassVar[bool] = False - @classmethod def drop(cls, db: sqlite3.Connection, name: str) -> None: db.execute(f"DROP {cls.declared_name.upper()} IF EXISTS {_identifier(name)}") @@ -608,8 +605,6 @@ def derive(cls, db: sqlite3.Connection, item: SQLiteCatalogObject, class TableKind(CatalogKind): - holds_rows = True - @classmethod def derive(cls, db: sqlite3.Connection, item: SQLiteCatalogObject, declared: Mapping[str, str]) -> None: @@ -618,7 +613,7 @@ def derive(cls, db: sqlite3.Connection, item: SQLiteCatalogObject, class IndexKind(CatalogKind): - named_by_coverage = True + pass class TriggerKind(CatalogKind): diff --git a/tests/test_declared_schema_carry.py b/tests/test_declared_schema_carry.py deleted file mode 100644 index 9bc0219c62..0000000000 --- a/tests/test_declared_schema_carry.py +++ /dev/null @@ -1,202 +0,0 @@ -"""The carry gives every durable file kind its declared schema and keeps every row.""" - -from contextlib import closing -import json -import re -import sqlite3 -from pathlib import Path - -import pytest - -from order_scope_fixture import FIX_SCOPE - -from agent_comms.change_report import FileLine -from agent_comms.errors import RelationViolationError -from agent_comms.goals import OrderCheck -from agent_comms.installer.carry import CarryChain -from agent_comms.native_pi import NativeContextJournal -from agent_comms.native_session_files import NativeSessionFiles, PrivateSession -from agent_comms.recorded_time import RecordedEventTime -from agent_comms.registration import Registration -from agent_comms.repository_name import RepositoryName -from agent_comms.todos import GoalRef, PendingProblemOrder, Todo, TodoStore -from agent_comms.order_runs import MechanicalOrderRun -from agent_comms.typed_table import DurableTables, SharedFileTables -from cutover.carry_declared_schema import carry - -REPO = Path(__file__).resolve().parents[1] - - -def _file_rows(): - """Each durable file's rows: the shared coordinator file is one file of many kinds.""" - kinds = {kind for kind in DurableTables.schema_members() - if not any(issubclass(base, DurableTables) and base.declares_schema() - for base in kind.__bases__ if base is not DurableTables)} - shared = tuple({row for kind in kinds if issubclass(kind, SharedFileTables) - for row in kind.schema_rows()}) - files = {kind.__name__: kind.schema_rows() for kind in kinds - if not issubclass(kind, SharedFileTables)} - return {"coordinator": shared, **files} - - -FILES = _file_rows() -_INDEX = re.compile(r'^(CREATE (?:UNIQUE )?INDEX )"[^"]+"( ON )') -# Each column's stored type answers a value STRICT accepts; checks are not enforced while -# the old file is written, so one row reaches every table whose generated columns allow it. -_VALUES = {"TEXT": "x", "INTEGER": 1, "REAL": 1.0, "BLOB": b"x", "": 1, "ANY": 1} -_EXTRACT = re.compile(r'"\w+" (\w+) GENERATED ALWAYS AS \(json_extract\((\w+), ?\'\$\.([\w.]+)\'\)\)') - - -def _stand_in(sql, column, storage): - """A TEXT column a generated column reads as JSON holds each path it reads.""" - document = {"kind": "x"} - for kind, source, path in _EXTRACT.findall(sql): - if source == column: - *parents, leaf = path.split(".") - node = document - for name in parents: - node = node.setdefault(name, {}) - node[leaf] = _VALUES[kind] - return json.dumps(document) if storage == "TEXT" else _VALUES[storage] - - -def _old_file(path, rows): - """The file as positional names and a sealed schema copy left it, with rows.""" - with closing(sqlite3.connect(path)) as db: - db.execute("PRAGMA ignore_check_constraints=1") - for row in rows: - db.execute(row.schema_objects()[row.declared_name]) - for row in rows: - columns = [column for column in db.execute(f'PRAGMA table_xinfo("{row.declared_name}")') - if column[6] == 0] - try: - db.execute( - f'INSERT INTO "{row.declared_name}" ({",".join(chr(34) + c[1] + chr(34) for c in columns)})' - f' VALUES ({",".join("?" for _ in columns)})', [_stand_in(row.schema_objects()[row.declared_name], c[1], c[2]) for c in columns]) - except sqlite3.IntegrityError: - pass # a generated NOT NULL column refuses the stand-in value - for row in rows: - indexes = [sql for name, sql in row.schema_objects().items() if name != row.declared_name - and _INDEX.match(sql)] - for ordinal, sql in enumerate(indexes): - db.execute(_INDEX.sub(rf'\1"{row.declared_name}_{ordinal}_idx"\2', sql)) - for name, sql in row.schema_objects().items(): - if name != row.declared_name and not _INDEX.match(sql): - db.execute(sql) - db.execute("CREATE TABLE cohort_schema_meta (singleton INTEGER PRIMARY KEY, ddl_digest TEXT)") - db.execute("INSERT INTO cohort_schema_meta VALUES (1, 'digest')") - db.execute("CREATE TRIGGER cohort_meta_update_guard BEFORE UPDATE ON cohort_schema_meta " - "BEGIN SELECT RAISE(ABORT, 'frozen'); END") - db.execute("PRAGMA user_version=11") - db.commit() - - -def _rows(path, rows): - with closing(sqlite3.connect(path)) as db: - return {row.declared_name: sorted(map(repr, db.execute(f'SELECT * FROM "{row.declared_name}"'))) - for row in rows} - - -@pytest.mark.parametrize("file", sorted(FILES)) -def test_carry_opens_an_old_file_with_identical_rows(tmp_path, file): - rows = FILES[file] - path = tmp_path / "old.sqlite3" - _old_file(path, rows) - stored = _rows(path, rows) - assert any(stored.values()) - carried = carry(path) - assert set(carried.renamed().values()) == { - name for row in rows for name, sql in row.schema_objects().items() if _INDEX.match(sql)} - assert carried.retired == {"cohort_schema_meta": 1} - with closing(sqlite3.connect(path)) as db: - assert not db.execute( - "SELECT name FROM sqlite_master WHERE name LIKE 'cohort_meta%' OR name LIKE '%schema_meta'" - ).fetchall() - assert db.execute("PRAGMA user_version").fetchone() == (0,) - for row in rows: - assert {item.name: item.sql for item in row.stored_objects(db)} == row.schema_objects() - assert _rows(path, rows) == stored - again = carry(path) - assert not again.renamed() and again.before == again.after - assert _rows(path, rows) == stored - - -def _refused(path, statement, match): - with closing(sqlite3.connect(path)) as db: - db.execute(statement) - db.commit() - before = path.read_bytes() - with pytest.raises(RelationViolationError, match=match): - carry(path) - assert path.read_bytes() == before - - -def test_carry_refuses_an_undeclared_table(tmp_path): - _refused(tmp_path / "f.sqlite3", "CREATE TABLE unknown_rows (x INTEGER)", "undeclared table") - - -def test_carry_refuses_a_table_whose_columns_differ(tmp_path): - _refused(tmp_path / "f.sqlite3", 'CREATE TABLE "binding" ("root" TEXT NOT NULL) STRICT', - "columns differ") - - -def test_carry_refuses_a_derived_file(tmp_path): - from agent_comms.bus_page_index import BusPageIndex - - row = BusPageIndex.rows[0] - _refused(tmp_path / "f.sqlite3", row.schema_objects()[row.declared_name], "derived") - - -def _positional_indexes(db, row): - """Rename a row's declared indexes to the positional names older files hold.""" - indexes = [(name, sql) for name, sql in row.schema_objects().items() if _INDEX.match(sql)] - for ordinal, (name, sql) in enumerate(indexes): - db.execute(f'DROP INDEX "{name}"') - db.execute(_INDEX.sub(rf'\1"{row.declared_name}_{ordinal}_idx"\2', sql)) - - -def test_the_schema_carry_runs_after_the_review_timer_carry(): - """The todo file's deadline column is renamed before its schema is compared.""" - names = [step.name() for step in CarryChain.at(REPO).steps] - assert names.index("carry_review_timers") < names.index("carry_declared_schema") - - -def test_a_bus_with_a_todo_deadline_and_old_proofs_carries_through_the_whole_chain(tmp_path): - lead = Registration(tmp_path / "registry.json").new_thread( - "lead", frozenset(), str(tmp_path), created_at=1.0) - order = PendingProblemOrder( - id="checked-order", owner=FileLine(__file__, 1), scope=FIX_SCOPE, run=MechanicalOrderRun(), - check=OrderCheck(str(tmp_path), "true"), edits=("Fix the problem",), - banned=("No extra store",), report="Report the delivered commit", - at=RecordedEventTime(10.0), first_call_deadline=RecordedEventTime(130.0)) - created = TodoStore(tmp_path).create( - "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", - creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", - review_at=RecordedEventTime(1000.0), order=order) - with closing(sqlite3.connect(TodoStore.path_in(tmp_path))) as db: - db.execute('ALTER TABLE "todo" RENAME COLUMN "review_at" TO "deadline"') - db.commit() - from agent_comms.native_session_reopen import NativeSessionIdentity - - # Schema conversion owns the durable proof even without a surviving transcript. - files = NativeSessionFiles.for_identity( - NativeSessionIdentity('s', str(PrivateSession.path_in(tmp_path, session_id='s'))), - root=tmp_path, owner=lead.incarnation.require_recorded(), - ) - proof = files.input_proof - files.prepare_private_directory(proof.parent) - proof.touch(mode=0o600) - with closing(sqlite3.connect(proof)) as db: - NativeContextJournal.create(db) - _positional_indexes(db, NativeContextJournal) - db.commit() - assert NativeContextJournal.schema_differs(db) - - for step in CarryChain.at(REPO).steps: - step.carry(tmp_path, apply=True) - - assert TodoStore(tmp_path).get("P1").review_at == created.review_at - with closing(sqlite3.connect(TodoStore.path_in(tmp_path))) as db: - assert "deadline" not in Todo.stored_columns(db) - with closing(sqlite3.connect(proof)) as db: - assert not NativeContextJournal.schema_differs(db) diff --git a/tools/cutover/carry_declared_schema.py b/tools/cutover/carry_declared_schema.py deleted file mode 100644 index 44775e7c6a..0000000000 --- a/tools/cutover/carry_declared_schema.py +++ /dev/null @@ -1,209 +0,0 @@ -"""Carry durable SQLite files onto their declared schema once; keep every row. - -Files written before index names were derived from what each index covers hold -positional names (delivery_0_idx), and files written before every file kind was -compared with its declaration hold a sealed copy of it: a meta table with a -digest or version, triggers freezing it, and PRAGMA user_version. Each file is -converted in place inside one write transaction: indexes, triggers and views -come from the declarations of the tables it holds, the sealed copies go, and -every table keeps its rows. - -The files are every durable SQLite file the bus declares and every saved session's -input proof under the bus root; a file whose tables all rebuild themselves is left to rebuild. Delete -once every bus has run it. -""" - -from __future__ import annotations - -import re -from contextlib import closing -from dataclasses import dataclass -from importlib import import_module -from pathlib import Path -from pkgutil import walk_packages - -import agent_comms -from agent_comms.coordination_database import CoordinationStore -from agent_comms.declared_family import UnknownFamilyMember -from agent_comms.errors import RelationViolationError -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import DurableFile, SQLiteFile -from agent_comms.typed_table import ( - DerivedTables, - SQLiteCatalogObject, - SQLiteSchemaObject, - SQLiteUserVersion, - TypedRow, - TypedTable, -) - -# The sealed copies the declarations replaced; their classes are deleted, so only -# their stored names remain. -RETIRED_TABLES = frozenset({ - "schema_meta", - "cohort_schema_meta", - "awareness_schema_meta", - "native_runtime_schema_meta", - "response_schema_meta", - "goal_attempt_schema", -}) - -# Every declaration, so a stored table is looked up among all of them and an object -# another file kind's row declares is never mistaken for a stray. -for _module in walk_packages(agent_comms.__path__, "agent_comms."): - import_module(_module.name) - - -_INDEX_NAME = re.compile(r'^(CREATE (?:UNIQUE )?INDEX )(?:"(?:[^"]|"")+"|\S+)( ON )') - - -def unnamed(sql: str) -> str: - """An index's SQL without its name: what it covers.""" - return _INDEX_NAME.sub(r"\1\2", sql, count=1) - - -@dataclass(frozen=True) -class StoredObject(SQLiteCatalogObject): - """One catalog entry, as the legacy carry reads it: retired sealed copies included.""" - - @property - def retired(self) -> bool: - return self.type.holds_rows and self.name in RETIRED_TABLES - - def declaration(self, path: Path) -> type[TypedTable]: - try: - return TypedTable.member_named(self.name) - except UnknownFamilyMember: - raise RelationViolationError(f"{path}: undeclared table {self.name}") from None - - def row(self, path: Path) -> type[TypedTable]: - """The declaration of this stored table, refusing one the carry cannot keep.""" - row = self.declaration(path) - if derived(row): - raise RelationViolationError(f"{path}: {self.name} is derived; it rebuilds, never carried") - if self.sql != row.schema_objects()[row.declared_name]: - raise RelationViolationError(f"{path}: table {self.name} columns differ") - return row - - def render(self) -> str: - return f"{self.type.declared_name} {self.name} on {self.tbl_name}: {self.sql}" - - -def derived(row: type[TypedTable]) -> bool: - return any(row in kind.schema_rows() for kind in DerivedTables.schema_members()) - - -@dataclass(frozen=True) -class RowCount(TypedRow): - rows: int - - @classmethod - def of(cls, db, tables) -> dict[str, int]: - return {table.name: cls.read(db.execute(f'SELECT COUNT(*) AS rows FROM "{table.name}"'))[0].rows - for table in tables} - - -@dataclass(frozen=True) -class FileCarried: - """What one file's carry changed: the catalog before and after, and the rows kept.""" - - path: Path - before: tuple[StoredObject, ...] - after: tuple[StoredObject, ...] - kept: dict[str, int] - retired: dict[str, int] - - def renamed(self) -> dict[str, str]: - """Old index name to declared name, matched by what the index covers.""" - after = {unnamed(item.sql): item.name for item in self.after if item.type.named_by_coverage} - names = {item.name for item in self.after} - return {item.name: after[unnamed(item.sql)] for item in self.before - if item.type.named_by_coverage and item.name not in names and unnamed(item.sql) in after} - - -@dataclass(frozen=True) -class StoredFile: - """One durable file as its catalog reads, before anything is written.""" - - path: Path - objects: tuple[StoredObject, ...] - user_version: int - - @classmethod - def read(cls, path: Path) -> StoredFile: - with closing(CoordinationStore.read_snapshot(path)) as db: - (version,) = SQLiteUserVersion.read(db.execute("PRAGMA user_version")) - return cls(path, StoredObject.stored(db), version.user_version) - - @property - def rebuilds(self) -> bool: - """Every table it holds is derived, so its owner rebuilds it on open.""" - tables = [item for item in self.objects if item.type.holds_rows and not item.retired] - return bool(tables) and all(derived(item.declaration(self.path)) for item in tables) - - @property - def pending(self) -> bool: - """Whether its catalog differs from what its tables' declarations derive.""" - rows = [item.row(self.path) for item in self.objects - if item.type.holds_rows and not item.retired] - declared = {name: sql for row in rows for name, sql in row.schema_objects().items()} - return (self.user_version != 0 - or {item.name: item.sql for item in self.objects} != declared) - - -def stores(root: Path) -> tuple[StoredFile, ...]: - """Every durable SQLite file the bus declares, and every saved session's input proof - under it. A step writes only inside the root it is given (practice.py), so the - registry's absolute paths, which on a practice copy name the live bus, are never read.""" - declared = {path for owner, directory in SQLiteFile.declarations_in(root) - if issubclass(owner, DurableFile) for path in owner.matching_in(directory)} - return tuple(map(StoredFile.read, sorted(declared))) - - -def pending(root: Path) -> tuple[StoredFile, ...]: - return tuple(store for store in stores(root) if not store.rebuilds and store.pending) - - -def carry(path: Path) -> FileCarried: - if not path.is_file(): - raise RelationViolationError(f"{path} does not exist; nothing to carry") - with closing(CoordinationStore.writer(path, isolation_level=None)) as db: - with SQLiteFile.write_transaction(db): - before = StoredObject.stored(db) - tables = [item for item in before if item.type.holds_rows] - kept = [item for item in tables if not item.retired] - rows = tuple(item.row(path) for item in kept) - declared = {name: sql for row in rows for name, sql in row.schema_objects().items()} - counts = RowCount.of(db, kept) - retired = RowCount.of(db, [item for item in tables if item.retired]) - for item in tables: - if item.retired: - item.drop(db) - StoredObject.carry(db, declared, frozenset(item.name for item in before)) - if differences := SQLiteSchemaObject.differences(TypedTable.objects_of(db, rows), declared): - raise RelationViolationError(f"{path}: still differs: {differences}") - if RowCount.of(db, kept) != counts: - raise RelationViolationError(f"{path}: row counts changed") - db.execute("PRAGMA user_version=0") - after = StoredObject.stored(db) - return FileCarried(path, before, after, counts, retired) - - -class DeclaredSchema(Carry): - """Durable files and session input proofs keep every row on their declared schema.""" - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - return tuple(path for store in pending(root) for path in SQLiteFile.files(store.path) - if path.exists()) - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - stale = pending(root) - if not apply: - return Carried(len(stale)) - carried = tuple(carry(store.path) for store in stale) - return Carried(len(carried), { - "renamed": sum(len(item.renamed()) for item in carried), - "retired_rows": sum(sum(item.retired.values()) for item in carried), - }, tuple(str(item.path.relative_to(root)) for item in carried)) diff --git a/tools/cutover/carry_native_input_proofs.py b/tools/cutover/carry_native_input_proofs.py index 936943e1a5..879c493a5e 100644 --- a/tools/cutover/carry_native_input_proofs.py +++ b/tools/cutover/carry_native_input_proofs.py @@ -25,7 +25,6 @@ from agent_comms.native_session_files import NativeInputProof, NativeSessionFiles from agent_comms.native_session_reopen import NativeSessionIdentity from agent_comms.private_path import NativeSessionFileRole, fsync_directory -from cutover import carry_declared_schema @dataclass(frozen=True, slots=True, kw_only=True) @@ -292,24 +291,33 @@ def backup_files(cls, root: Path) -> tuple[Path, ...]: for path in SQLiteFile.files(primary) if path.exists() ) + @classmethod + def differing(cls, root: Path) -> tuple[Path, ...]: + """Proof journals whose stored schema is not their declaration; a dry run counts them + too, since a step's dry run is what says whether it runs.""" + found = [] + for path in sorted(NativeInputProof.matching_in(root)): + cls.require_confined(root, path) + NativeSessionFileRole.require(path.lstat()) + with closing(CoordinationStore.recovering_snapshot(path)) as db: + if NativeContextJournal.schema_differs(db): + found.append(path) + return tuple(found) + @classmethod def carry(cls, root: Path, *, apply: bool) -> Carried: root = root.resolve(strict=True) plans = cls.relocations(root) if not apply: - return Carried(len(plans)) + return Carried(len(plans) + len(cls.differing(root))) written = {path for plan in plans for path in plan.relocate()} - converted = 0 - for path in sorted(NativeInputProof.matching_in(root)): - cls.require_confined(root, path) - NativeSessionFileRole.require(path.lstat()) - with closing(CoordinationStore.recovering_snapshot(path)) as db: - differs = NativeContextJournal.schema_differs(db) - if differs: - carry_declared_schema.carry(path) - converted += 1 - written.update(SQLiteFile.files(path)) - return Carried(len(plans) + converted, {"proofs": len(plans), "schemas": converted}, + converted = cls.differing(root) + for path in converted: + with closing(CoordinationStore.writer(path, isolation_level=None)) as db: + NativeContextJournal.reconcile_schema(db) + written.update(SQLiteFile.files(path)) + return Carried(len(plans) + len(converted), + {"proofs": len(plans), "schemas": len(converted)}, tuple(str(path.relative_to(root)) for path in sorted(written))) From 16106347c6811bd5d8f5ad8df706ea53bd32b71b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 19:06:21 -0400 Subject: [PATCH 3/3] Delete every carry step the one bus has run: there are no legacy cases There is one bus, and install 47 reported 'Pending carries: none' at 7dc1894e2, so every step on main had run on it. Keeping them as 'legacy' code was slop: it can only run against a bus that does not exist, and each one's dry run now decides installs. Deleted: carry_context_store, carry_goal_execution, carry_goal_work_decision, carry_install_schedule, carry_native_input_proofs, carry_order_scopes, carry_proof_build_commits, carry_review_timers, the one-shot P380 connector_browser_trust tool, and the tests that exercised them (including test_order_scopes' carry test, which still called the deleted pending(applied) signature). carry_retired_ledger is the only step left, so the RetiredFiles base with one user folds into it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/installer/carry.py | 21 -- tests/test_goal_execution.py | 25 +- tests/test_holder_standing_rules.py | 68 +--- tests/test_native_input_proof_carry.py | 250 --------------- tests/test_order_scopes.py | 47 +-- tests/test_proof_build.py | 22 -- tools/cutover/carry_context_store.py | 271 ---------------- tools/cutover/carry_goal_execution.py | 76 ----- tools/cutover/carry_goal_work_decision.py | 57 ---- tools/cutover/carry_install_schedule.py | 9 - tools/cutover/carry_native_input_proofs.py | 353 --------------------- tools/cutover/carry_order_scopes.py | 83 ----- tools/cutover/carry_proof_build_commits.py | 96 ------ tools/cutover/carry_retired_ledger.py | 19 +- tools/cutover/carry_review_timers.py | 119 ------- tools/cutover/connector_browser_trust.py | 34 -- 16 files changed, 19 insertions(+), 1531 deletions(-) delete mode 100644 tests/test_native_input_proof_carry.py delete mode 100644 tools/cutover/carry_context_store.py delete mode 100644 tools/cutover/carry_goal_execution.py delete mode 100644 tools/cutover/carry_goal_work_decision.py delete mode 100644 tools/cutover/carry_install_schedule.py delete mode 100644 tools/cutover/carry_native_input_proofs.py delete mode 100644 tools/cutover/carry_order_scopes.py delete mode 100644 tools/cutover/carry_proof_build_commits.py delete mode 100644 tools/cutover/carry_review_timers.py delete mode 100644 tools/cutover/connector_browser_trust.py diff --git a/src/agent_comms/installer/carry.py b/src/agent_comms/installer/carry.py index 3925e985da..e5dd1d6d86 100644 --- a/src/agent_comms/installer/carry.py +++ b/src/agent_comms/installer/carry.py @@ -111,24 +111,3 @@ def pending(self, dry_run: Callable[[type[Carry]], Carried]) -> tuple[type[Carry return self.steps[position:] return () - -class RetiredFiles(Carry): - """A step that deletes files the bus no longer reads.""" - - @classmethod - @abstractmethod - def retired(cls) -> tuple[str, ...]: - """The files' names relative to the bus root.""" - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - return tuple(root / name for name in cls.retired()) - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - paths = tuple(path for path in cls.backup_files(root) if path.exists()) - for path in paths: - cls.require_confined(root, path) - if apply: - path.unlink() - return Carried(len(paths), {}, tuple(path.name for path in paths) if apply else ()) diff --git a/tests/test_goal_execution.py b/tests/test_goal_execution.py index b7e97e0a1c..c58926e091 100644 --- a/tests/test_goal_execution.py +++ b/tests/test_goal_execution.py @@ -2,17 +2,14 @@ import os import sqlite3 from contextlib import AsyncExitStack, asynccontextmanager, closing -from dataclasses import replace import pytest -from cutover.carry_goal_execution import FAILED_BACKEND_REASON, BackendGoalBlocks from agent_comms import agent_events as events from agent_comms.acp import CommsAgent from agent_comms.child_process import ProcessIdentity from agent_comms.comms import Comms from agent_comms.goal_actions import ( - BlockedGoalAction, PausedGoalAction, RetryGoalAction, SetGoalAction, @@ -23,10 +20,9 @@ RunnableExecution, RunningExecution, ) -from agent_comms.goal_states import ActiveGoal, BlockedGoal +from agent_comms.goal_states import ActiveGoal from agent_comms.owned_turn import OwnedTurn from agent_comms.turn_terminal import OwnerStop -from test_goal_answers import copied_goal_bus @pytest.fixture @@ -128,22 +124,3 @@ async def test_verified_productive_turn_recovers_blocked_generation(execution_bu assert comms.registry.require("worker").goal.state == ActiveGoal() assert failure_count(comms) == 1 # The old failure remains evidence, never replayed. - -def test_cutover_retires_backend_text_and_preserves_owner_block(execution_bus): - comms = execution_bus - owner = comms.registry.require("worker") - comms.registry.register(replace(owner, goal=replace( - owner.goal.require(), - state=BlockedGoal(FAILED_BACKEND_REASON), - ))) - comms.registry.declare(comms.registry.new_thread("explicit", frozenset(), owner.worktree)) - comms.goals.update_goal("explicit", SetGoalAction(text="Needs owner decision")) - decided = comms.goals.update_goal("explicit", BlockedGoalAction(progress="Waiting for owner approval")) - dry = BackendGoalBlocks.carry(comms.root, apply=False) - assert dry.detail == {"migrated": 1, "preserved": 1} - assert comms.registry.require("worker").goal.state != ActiveGoal() - carried = BackendGoalBlocks.carry(comms.root, apply=True) - assert carried.detail == dry.detail - assert comms.registry.require("worker").goal.state == ActiveGoal() - assert comms.registry.require("explicit").goal == decided - assert BackendGoalBlocks.carry(comms.root, apply=True).changed == 0 diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index c6b9a49ca9..c076894894 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -5,14 +5,11 @@ import subprocess import sys import time -from contextlib import closing from dataclasses import replace from functools import partial -from pathlib import Path from uuid import uuid4 import pytest -from cutover.carry_goal_work_decision import RETIRED_TABLE, GoalWorkDecisions from agent_comms import agent_events as events from agent_comms.bus_publication import AgentOrigin @@ -28,22 +25,17 @@ ) from agent_comms.commander_heartbeat import CommanderHeartbeat from agent_comms.comms import Comms -from agent_comms.coordination_database import CoordinationStore from agent_comms.coordinator import Coordination -from agent_comms.errors import RelationViolationError from agent_comms.field_codec import FieldCodec from agent_comms.goal_actions import ( CreateProblemGoalAction, DirectiveGoalAction, ReissuedOrderGoalAction, SetGoalAction, ) -from agent_comms.goal_attempts import GoalAttemptStore, GoalWorkDecision from agent_comms.goal_presentation import RunnableExecution from agent_comms.goal_states import ActiveGoal -from agent_comms.goals import CountCheckResult, Goal, OrderCheck, RequestGoalOrder +from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder from agent_comms.input_disposition import InputDispositions from agent_comms.installer.backup import StoreBackup -from agent_comms.installer.carry import CarryChain from agent_comms.launch_root import LaunchRoot -from agent_comms.locked_store import SQLiteFile from agent_comms.message_page import MessagePageRequest from agent_comms.messages import InfoKind, Message from agent_comms.native_arguments import GivenBinary @@ -400,64 +392,6 @@ def test_detached_and_goal_blocks_are_both_in_standing_rule_scan(held_bus, insta assert all(rule in broken for rule in family) -def legacy_decision_table(comms): - store = comms.goals.open_goal_store() - goal = comms.registry.require("worker").goal.require() - generation = store.require_generation(goal.id) - with closing(GoalAttemptStore.database(store.path)) as db: - with SQLiteFile.write_transaction(db): - GoalWorkDecision(goal.id, "recorded-decision", generation.number).claim(db) - db.execute(f'ALTER TABLE "{GoalWorkDecision.declared_name}" RENAME TO "{RETIRED_TABLE}"') - return store.path - - -def test_cutover_renames_copied_ledger_once_and_preserves_rows(held_bus, comms): - path = legacy_decision_table(held_bus) - original = comms.goals.open_goal_store().path.read_bytes() - with closing(CoordinationStore.read_snapshot(path)) as db: - before = tuple(db.execute(f'SELECT * FROM "{RETIRED_TABLE}"')) - assert GoalWorkDecisions.carry(held_bus.root, apply=False).changed == 1 - carried = GoalWorkDecisions.carry(held_bus.root, apply=True) - assert carried.changed == 1 - assert carried.rewritten == (str(path.relative_to(held_bus.root)),) - assert GoalWorkDecisions.carry(held_bus.root, apply=True).changed == 0 - GoalAttemptStore(path.parent) # the installed reader accepts the carried schema - with closing(CoordinationStore.read_snapshot(path)) as db: - assert tuple(db.execute(f'SELECT * FROM "{GoalWorkDecision.declared_name}"')) == before - assert comms.goals.open_goal_store().path.read_bytes() == original - assert held_bus.goals.open_goal_store().require_generation( - held_bus.registry.require("worker").goal.require().id, - ).number == 1 - assert path in GoalWorkDecisions.backup_files(held_bus.root) - - -def test_cutover_refuses_two_tables_without_choosing_rows(held_bus): - path = legacy_decision_table(held_bus) - with closing(GoalAttemptStore.database(path)) as db: - GoalWorkDecision.create(db) - db.commit() - before = path.read_bytes() - with pytest.raises(RelationViolationError): - GoalWorkDecisions.carry(held_bus.root, apply=True) - assert path.read_bytes() == before - - -def test_install_discovers_the_decision_cutover_from_its_dry_run_and_runs_it_once(held_bus): - chain = CarryChain.at(Path(__file__).resolve().parents[1]) - assert GoalWorkDecisions in chain.steps - - def dry_run(step): - return step.carry(held_bus.root, apply=False) - - assert chain.pending(dry_run) == () - legacy_decision_table(held_bus) - assert chain.pending(dry_run)[0] is GoalWorkDecisions - assert GoalWorkDecisions.carry(held_bus.root, apply=True).changed == 1 - second = GoalWorkDecisions.carry(held_bus.root, apply=True) - assert second.changed == 0 and second.rewritten == () - assert chain.pending(dry_run) == () - - def test_startup_heartbeat_enforces_and_publishes_both_new_rule_lines(held_bus, installed_runtime): blocked_holder(held_bus) question = publish(held_bus, "worker", "lead", time.time() - 901) diff --git a/tests/test_native_input_proof_carry.py b/tests/test_native_input_proof_carry.py deleted file mode 100644 index 111f76beb4..0000000000 --- a/tests/test_native_input_proof_carry.py +++ /dev/null @@ -1,250 +0,0 @@ -"""Actual SQLite resources move inside the copied root, without copying proof bytes.""" -from contextlib import closing -import hashlib -import json -from pathlib import Path -import sqlite3 -import subprocess -import sys - -import pytest - -from agent_comms.coordination_database import CoordinationStore -from agent_comms.coordination_tables.participants import Participants -from agent_comms.errors import RelationViolationError -from agent_comms.fresh_private_session import create_fresh_private_session -from agent_comms.locked_store import SQLiteFile -from agent_comms.native_claim_epoch import RecordedNativeClaim -from agent_comms.native_input_record import TriageNativeExecution, UnrecordedNativeColumn -from agent_comms.native_pi import NativeContextJournal, NativeContextProof -from agent_comms.native_runtime_input import NativeRuntimeInput -from agent_comms.native_session_files import NativeSessionFiles -from agent_comms.native_session_reopen import NativeSessionIdentity -from agent_comms.thread_identity import RecordedThreadIncarnation -from cutover.carry_native_input_proofs import NativeInputProofs - -OWNER = RecordedThreadIncarnation('proof-owner', 1.0) - - -def source(root: Path, mode='DELETE', owners=(OWNER,)): - root.mkdir(mode=0o700) - identity = create_fresh_private_session(root, worktree=root) - with identity.path.open('ab') as transcript: - transcript.write((json.dumps({ - 'type':'message','timestamp':'2026-10-01T00:00:00.000Z','id':'entry', - 'message':{'role':'user','inputId':'a'*32,'inputDigest':'b'*64, - 'content':[{'type':'text','text':'retained source'}]}, - })+'\n').encode()) - with closing(sqlite3.connect(CoordinationStore.path_in(root))) as db: - Participants.create(db) - NativeRuntimeInput.create(db) - for ordinal, owner in enumerate(owners): - Participants(participant_lookup=owner.lookup, display_name=owner.name, committed=True).insert(db) - NativeRuntimeInput( - input_id=f'{ordinal+1:032x}', stage=TriageNativeExecution, - execution_id=UnrecordedNativeColumn(), attempt_ordinal=UnrecordedNativeColumn(), - owner_lookup=owner.lookup, - owner_thread=owner.name, owner_generation=1, owner_token_digest='e'*64, - sent_owner_claimed_generation=RecordedNativeClaim(1), session_id=identity.session_id, - # A copied record names the original; the carry never follows it. - session_file='/original/bus/sessions/'+identity.path.name, - ).insert(db) - db.commit() - CoordinationStore.path_in(root).chmod(0o600) - legacy = identity.path.with_name(identity.path.name+'.input-proof') - # A real abruptly closed writer leaves WAL/SHM without a live borrower. - completed = subprocess.run([sys.executable, '-c', ''' -import os,sqlite3,sys -from agent_comms.native_pi import NativeContextJournal -path,sid,mode=sys.argv[1:] -os.close(os.open(path,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)) -db=sqlite3.connect(path) -db.execute('PRAGMA journal_mode='+mode) -NativeContextJournal.create(db) -NativeContextJournal(input_id='a'*32, session_id=sid, session_entry_id='entry', - request_generation=1,llm_context_digest='c'*64,committed_at_ms=1791964800001, - schema=2,type='context_committed').insert(db) -db.commit() -os.chmod(path,0o600) -os._exit(0) -''', str(legacy),identity.session_id,mode], timeout=15, capture_output=True) - assert completed.returncode == 0, completed.stderr.decode() - return NativeSessionFiles.for_identity(identity,root=root,owner=OWNER), legacy - - -@pytest.mark.parametrize('mode', ['DELETE','WAL']) -def test_move_keeps_inode_rows_and_all_sqlite_companions(tmp_path, mode): - root = tmp_path/'copy' - files, legacy = source(root, mode) - before = {path: path.stat().st_ino for path in SQLiteFile.files(legacy) if path.exists()} - if mode == 'WAL': - assert len(before) == 3 - transcript = files.session.read_bytes() - assert NativeInputProofs.carry(root,apply=False).changed == 1 - assert all(path.exists() for path in before) - moved = NativeInputProofs.carry(root,apply=True) - assert moved.detail['proofs'] == 1 - assert files.input_proof.stat().st_ino == before[legacy] - assert not any(path.exists() for path in before) - assert files.session.read_bytes() == transcript - with NativeContextJournal.open_evidence(files) as db: - rows = NativeContextJournal.select(db) - assert len(rows)==1 and rows[0].session_id==files.session_id - assert NativeInputProofs.carry(root,apply=True).changed == 0 - - -def test_scope_and_rename_follow_native_identity_not_filename(tmp_path): - root=tmp_path/'copy' - files,legacy=source(root) - NativeInputProofs.carry(root,apply=True) - with files.session.open('rb') as transcript: - transcript.readline() - start=transcript.tell() - renamed=files.session.with_name('renamed.jsonl') - files.session.rename(renamed) - current=NativeSessionFiles.for_identity(NativeSessionIdentity.locate(str(renamed)),root=root,owner=OWNER) - assert current.input_proof==files.input_proof - observed=NativeContextProof.read_evidence(current,'a'*32,session_entry_start=start) - assert observed.session_id==files.session_id and observed.input_id=='a'*32 - other_owner=NativeSessionFiles.for_identity(current,root=root, - owner=RecordedThreadIncarnation('other',2.0)) - other_bus=NativeSessionFiles.for_identity(current,root=tmp_path/'other-bus',owner=OWNER) - assert current.input_proof!=other_owner.input_proof - assert current.input_proof!=other_bus.input_proof - - -@pytest.mark.parametrize('owners', [(),(OWNER,RecordedThreadIncarnation('other',2.0))]) -def test_unknown_or_ambiguous_owner_refuses_before_any_move(tmp_path, owners): - root=tmp_path/'copy' - files,legacy=source(root,owners=owners) - before=legacy.read_bytes() - with pytest.raises(RelationViolationError,match='no unique recorded owner'): - NativeInputProofs.carry(root,apply=True) - assert legacy.read_bytes()==before and not files.input_proof.exists() - - -def test_collision_never_overwrites_either_proof(tmp_path): - root=tmp_path/'copy' - files,legacy=source(root) - files.input_proof.parent.mkdir(parents=True,mode=0o700) - files.input_proof.write_bytes(b'other proof') - before=legacy.read_bytes() - with pytest.raises(RelationViolationError,match='already exists'): - NativeInputProofs.carry(root,apply=True) - assert legacy.read_bytes()==before and files.input_proof.read_bytes()==b'other proof' - - -@pytest.mark.parametrize('redirect', ['proof','transcript','coordinator']) -def test_copy_alias_never_changes_the_original(tmp_path,redirect): - root=tmp_path/'copy' - files,legacy=source(root) - paths={'proof':legacy,'transcript':files.session,'coordinator':CoordinationStore.path_in(root)} - selected=paths[redirect] - original=tmp_path/'original' - selected.rename(original) - selected.symlink_to(original) - digest=hashlib.sha256(original.read_bytes()).digest() - with pytest.raises(RelationViolationError,match='unconfined'): - NativeInputProofs.carry(root,apply=True) - assert hashlib.sha256(original.read_bytes()).digest()==digest - assert not files.input_proof.exists() - - -def test_unrecognized_companion_is_preserved_and_refused(tmp_path): - root=tmp_path/'copy' - files,legacy=source(root) - staged=legacy.with_name(legacy.name+'.initializing-123') - staged.write_bytes(b'unresolved publication') - with pytest.raises(RelationViolationError,match='unrecognized companions'): - NativeInputProofs.carry(root,apply=True) - assert staged.read_bytes()==b'unresolved publication' and legacy.exists() - assert not files.input_proof.exists() - - -def external_source(tmp_path, name): - bus = tmp_path / (name + '-bus') - files, legacy = source(bus) - journals = tmp_path / (name + '-journals') - journals.mkdir(mode=0o700) - transcript = journals / files.session.name - files.session.rename(transcript) - external = journals / legacy.name - legacy.rename(external) - files = NativeSessionFiles.for_identity( - NativeSessionIdentity.locate(str(transcript)), root=bus, owner=OWNER, - ) - return files, external, journals - - -def create_target(files, row): - files.prepare_private_directory(files.input_proof.parent) - files.input_proof.touch(mode=0o600) - with closing(sqlite3.connect(files.input_proof)) as db: - NativeContextJournal.create_schema(db) - row.insert(db) - db.commit() - - -def test_external_history_merge_is_atomic_and_preserves_originals(tmp_path): - files, legacy, journals = external_source(tmp_path, 'missing') - # Different journal lifetimes can reuse a generation for different inputs; - # immutable evidence is keyed by the declared input/generation pair. - create_target(files, NativeContextJournal( - input_id='f'*32, session_id=files.session_id, session_entry_id='new-entry', - request_generation=1, llm_context_digest='e'*64, committed_at_ms=1791964800002, - schema=2, type='context_committed', - )) - original = hashlib.sha256(legacy.read_bytes()).digest() - inode = files.input_proof.stat().st_ino - plan, = NativeInputProofs.histories(files.root, journals) - preview = plan.inspect(journals) - assert (preview.difference.missing_rows, preview.difference.conflicting_keys) == (1, 0) - assert plan.apply(journals).inserted_rows == 1 - assert files.input_proof.stat().st_ino == inode - with NativeContextJournal.open_evidence(files) as db: - assert NativeContextJournal.count(db) == 2 - assert not NativeContextJournal.schema_differs(db) - # A normal writer still uses the unmodified append-only contract afterward. - with closing(sqlite3.connect(files.input_proof)) as db: - NativeContextJournal( - input_id='d'*32, session_id=files.session_id, session_entry_id='later-entry', - request_generation=2, llm_context_digest='d'*64, committed_at_ms=1791964800003, - schema=2, type='context_committed', - ).insert(db) - db.commit() - target_hash = hashlib.sha256(files.input_proof.read_bytes()).digest() - assert plan.apply(journals).inserted_rows == 0 - assert hashlib.sha256(files.input_proof.read_bytes()).digest() == target_hash - assert hashlib.sha256(legacy.read_bytes()).digest() == original - - conflict, old, conflict_root = external_source(tmp_path, 'conflict') - with closing(sqlite3.connect(old)) as db: - NativeContextJournal( - input_id='b'*32, session_id=conflict.session_id, session_entry_id='another-entry', - request_generation=2, llm_context_digest='d'*64, committed_at_ms=1791964800003, - schema=2, type='context_committed', - ).insert(db) - db.commit() - create_target(conflict, NativeContextJournal( - input_id='a'*32, session_id=conflict.session_id, session_entry_id='entry', - request_generation=1, llm_context_digest='e'*64, committed_at_ms=1791964800001, - schema=2, type='context_committed', - )) - old_hash = hashlib.sha256(old.read_bytes()).digest() - target_hash = hashlib.sha256(conflict.input_proof.read_bytes()).digest() - refused, = NativeInputProofs.histories(conflict.root, conflict_root) - preview = refused.inspect(conflict_root) - assert (preview.difference.missing_rows, preview.difference.conflicting_keys) == (1, 1) - with pytest.raises(RelationViolationError, match='conflicting keys'): - refused.apply(conflict_root) - assert hashlib.sha256(conflict.input_proof.read_bytes()).digest() == target_hash - assert hashlib.sha256(old.read_bytes()).digest() == old_hash - with NativeContextJournal.open_evidence(conflict) as db: - assert NativeContextJournal.count(db) == 1 - - skipped, preserved, skipped_root = external_source(tmp_path, 'missing-target') - preserved_hash = hashlib.sha256(preserved.read_bytes()).digest() - untouched, = NativeInputProofs.histories(skipped.root, skipped_root) - assert untouched.apply(skipped_root).inserted_rows == 0 - assert not skipped.input_proof.exists() - assert hashlib.sha256(preserved.read_bytes()).digest() == preserved_hash diff --git a/tests/test_order_scopes.py b/tests/test_order_scopes.py index 00054bdf18..8f83ca7d20 100644 --- a/tests/test_order_scopes.py +++ b/tests/test_order_scopes.py @@ -2,7 +2,6 @@ from dataclasses import replace import json import os -import sqlite3 import subprocess import sys from pathlib import Path @@ -15,22 +14,17 @@ from agent_comms.change_report import FileLine, MissingSourceLine, PullRequestHeadRevision from agent_comms.check_root import CheckRoot from agent_comms.commander.order import OrderFile -from agent_comms.coordination_database import CoordinationStore from agent_comms.field_codec import FieldCodec, MalformedRecord from agent_comms.goal_actions import CreateProblemGoalAction, SetGoalAction from agent_comms.goals import RequestGoalOrder -from agent_comms.installer.backup import StoreBackup -from agent_comms.installer.carry import Carried, CarryChain -from agent_comms.locked_store import SQLiteFile from agent_comms.recorded_time import RecordedEventTime from agent_comms.repository_name import RepositoryName from agent_comms.tool_caller import ThreadCaller from agent_comms.tools import CommsOrderDoneTool, CommsTodoCreateTool from agent_comms.todos import ( - CommitDelivery, DoneTodoState, FixScope, FixShape, FoldedProblemOrder, OrderScope, - PendingProblemOrder, ReportDelivery, ReportedProblemOrder, TodoStore, TriageScope, + CommitDelivery, DoneTodoState, FixShape, OrderScope, + PendingProblemOrder, ReportDelivery, ReportedProblemOrder, TriageScope, ) -from cutover.carry_order_scopes import TRIAGE_PROBLEMS, OrderScopes @pytest.mark.parametrize('scope', (TriageScope(), FIX_SCOPE)) @@ -128,40 +122,3 @@ def test_report_returns_through_the_real_cli_on_a_selected_history_copy(comms, t assert copied.goals.todos.get(current.id).state is DoneTodoState assert comms.goals.todos.get('P1').pending - -def test_carry_scopes_current_nested_and_prior_orders_on_a_copy(comms, tmp_path, source_repository): - assert OrderScopes in CarryChain.at(Path(__file__).resolve().parents[1]).pending(()) - _, original = assigned(comms, FIX_SCOPE, source_repository) - store = comms.goals.todos - lead = comms.registry.require('lead') - worker = comms.registry.require('worker') - for problem in TRIAGE_PROBLEMS: - store.create(problem, original.repo, 'Classify the source', creator=lead, owner=worker, - assigner=lead.incarnation, - generation=original.held().generation, review_at=original.review_at, - goal=original.goal, order=replace(order(source_repository), scope=TriageScope())) - copied = tmp_path / 'copy' - StoreBackup.copy_tree(comms.root, copied) - path = TodoStore.path_in(copied) - with sqlite3.connect(path) as db: - db.execute('UPDATE todo SET "order"=json_remove("order", ?)', ('$.scope',)) - prior = replace(original.order.checked(), id='prior-order') - db.execute('UPDATE todo SET prior_orders=json_array(json_remove(?, ?)) WHERE id=?', - (json.dumps(FieldCodec.encode(prior)), '$.scope', original.id)) - folded = FoldedProblemOrder(original=original.order, into='P9', why='Merged findings') - db.execute('UPDATE todo SET "order"=json_remove(?, ?) WHERE id=?', - (json.dumps(FieldCodec.encode(folded)), '$.original.scope', original.id)) - before = CoordinationStore.digest(path) - dry_run = OrderScopes.carry(copied, apply=False) - assert dry_run == Carried(len(TRIAGE_PROBLEMS) + 2) - assert CoordinationStore.digest(path) == before - applied = OrderScopes.carry(copied, apply=True) - assert applied == Carried(len(TRIAGE_PROBLEMS) + 2, rewritten=tuple( - str(file.relative_to(copied)) for file in SQLiteFile.files(path))) - converted = TodoStore(copied) - for problem in TRIAGE_PROBLEMS: - assert converted.get(problem).order.checked().scope == TriageScope() - for declaration in converted.get(original.id).order_history: - assert declaration.scope == FixScope(FixShape(parents=(declaration.owner,), members=())) - assert OrderScopes.carry(copied, apply=True) == Carried(0) - assert comms.goals.todos.get(original.id) == original diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py index 385e17810e..f6e8c01fbf 100644 --- a/tests/test_proof_build.py +++ b/tests/test_proof_build.py @@ -21,7 +21,6 @@ from agent_comms.field_codec import FieldCodec from agent_comms.goal_actions import CreateProblemGoalAction, SetGoalAction from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder -from agent_comms.installer.carry import CarryChain from agent_comms.installer.runtime_resolution import ( ArchiveRecord, CommitRecord, @@ -51,7 +50,6 @@ from agent_comms.tool_caller import ThreadCaller from agent_comms.tools import CommsOrderDoneTool, CommsProblemCloseTool, CommsTodoCreateTool from order_scope_fixture import FIX_SCOPE -from cutover.carry_install_schedule import InstallSchedule from source_revision_fixture import SourceRepository @@ -411,23 +409,3 @@ def test_install_waits_for_merges_to_settle_into_one_batch( assert graph.newer in due.why assert "oldest 30 min" in due.why - -def test_install_schedule_carry_is_discovered_and_deletes_only_its_file(tmp_path): - root = tmp_path / "bus" - root.mkdir() - schedule = root / "install-deadline.json" - schedule.write_text('{"at": 0, "installer": "retired"}') - history = root / "saved-session" - history.write_bytes(b"preserved original history") - assert InstallSchedule in CarryChain.at(Path(__file__).resolve().parents[1]).steps - assert InstallSchedule.backup_files(root) == (schedule,) - assert InstallSchedule.carry(root, apply=False).changed == 1 - assert schedule.exists() - - result = InstallSchedule.carry(root, apply=True) - - assert result.changed == 1 - assert schedule.name in result.rewritten - assert not schedule.exists() - assert history.read_bytes() == b"preserved original history" - assert InstallSchedule.carry(root, apply=True).changed == 0 diff --git a/tools/cutover/carry_context_store.py b/tools/cutover/carry_context_store.py deleted file mode 100644 index 5a1ed4c2a3..0000000000 --- a/tools/cutover/carry_context_store.py +++ /dev/null @@ -1,271 +0,0 @@ -"""Move historical context rows once; retain every message and sealed claim row.""" - -from __future__ import annotations - -import os -import sqlite3 -import tempfile -from abc import abstractmethod -from contextlib import ExitStack, closing -from dataclasses import dataclass, field, replace -from pathlib import Path -from typing import Annotated, ClassVar - -from agent_comms.context_store import ( - AbsentContext, - ContextEntry, - ContextStore, - ContextStoreInfo, - RecordedContextState, -) -from agent_comms.declared_family import DeclaredFamily -from agent_comms.errors import RelationViolationError -from agent_comms.field_codec import FieldCodec -from agent_comms.private_bus_checkpoint import ( - _SEED, - CheckedPrefix, - WritableCheckpoint, - _chain, - _connect, - _directory_sync, -) -from agent_comms.thread_identity import ThreadIncarnation -from agent_comms.turn_context import ( - ContextEntryManifest, - ContextEntryReference, - ContextManifest, - RecordedContextManifest, - SegmentManifest, -) -from agent_comms.typed_table import Column, Index, TypedTable -from agent_comms.wire_log import WireLog -from agent_comms.wire_record import WireRecord - -from agent_comms.installer.carry import Carried, Carry - - -# This old declaration exists only at this one-time format boundary: its table marks a bus -# whose context records still sit in the log. -@dataclass(frozen=True) -class RecordedContextManifestSources(TypedTable): - offset: Annotated[int, Column(check="offset>=0")] - length: Annotated[int, Column(check="length>0")] - thread: ThreadIncarnation - indexes = (Index(("thread", "offset")), Index(("offset",), unique=True)) - - -class ContextObservation(DeclaredFamily, affix="Observation"): - @abstractmethod - def move(self, move: ContextMove, offset: int) -> None: ... - - def recorded(self) -> RecordedContextManifest: - raise RelationViolationError("Context predecessor is not a recorded request") - - -@dataclass(frozen=True) -class ContextEntryManifestObservation(ContextObservation): - entries: tuple[ContextEntryManifest, ...] - - def move(self, move: ContextMove, offset: int) -> None: - for entry in self.entries: - move.entry(entry) - - -@dataclass(frozen=True) -class RecordedContextManifestObservation(ContextObservation): - manifest: RecordedContextManifest - - def recorded(self) -> RecordedContextManifest: - return self.manifest - - def move(self, move: ContextMove, offset: int) -> None: - move.request(self.manifest, offset) - - -@dataclass(frozen=True) -class ContextEnvelope: - observation: ContextObservation - strict_fields = True - - -class ContextMove: - """Decode old positions once; the context store owns recording and predecessors. - - Rows arrive in log order, so every predecessor and every entry a request names - was resolved earlier in the same scan; each is kept by its offset or reference - and never read back from the log or the store.""" - - def __init__(self, context_db: sqlite3.Connection, store: ContextStore) -> None: - self.context_db, self.store = context_db, store - self.source = ContextStoreInfo.source_in(context_db).initialized(context_db).source - self.entries: dict[ContextEntryReference, SegmentManifest] = {} - self.requests: dict[int, RecordedContextManifest] = {} - self.values: dict[int, tuple[tuple[SegmentManifest, tuple[ContextEntryReference, ...]], ...]] = {} - self.priors: dict[ThreadIncarnation, RecordedContextState] = {} - self.count = 0 - - def entry(self, entry: ContextEntryManifest) -> None: - # The first value of a journal identity wins, as insert_if_absent keeps it. - if entry.reference not in self.entries: - self.entries[entry.reference] = entry.manifest - ContextEntry.capture(entry).insert_if_absent(self.context_db) - - def original(self, reference: ContextEntryReference) -> SegmentManifest: - if reference not in self.entries: - raise RelationViolationError("Recorded context names an entry the log never held") - return self.entries[reference] - - def value( - self, - record: RecordedContextManifest, - position: int, - offset: int, - ) -> tuple[SegmentManifest, tuple[ContextEntryReference, ...]]: - part = record.segments[position] - header, references = part.manifest, part.entries - for predecessor in part.predecessors(offset): - if predecessor not in self.requests: - raise RelationViolationError("Context predecessor is not a recorded request") - if self.requests[predecessor].thread != record.thread: - raise RelationViolationError("Context predecessor belongs to another owner") - old, old_references = self.values[predecessor][position] - if part.inherited_entries > len(old_references): - raise RelationViolationError("Recorded context inherited entries are incomplete") - references = (*old_references[: part.inherited_entries], *references) - if not header.captured_text and old.sha256 == header.sha256: - header = replace(header, captured_text=old.captured_text) - return header, tuple(references) - - def request(self, record: RecordedContextManifest, offset: int) -> None: - values = tuple( - self.value(record, position, offset) for position in range(len(record.segments)) - ) - self.requests[offset], self.values[offset] = record, values - segments = tuple( - replace(header, contributors=tuple(self.original(ref) for ref in references)) - if references - else header - for header, references in values - ) - self.priors[record.thread] = self.store.record_after( - self.context_db, - self.source, - self.priors.get(record.thread, AbsentContext()), - ContextManifest(record.thread, record.turn, segments, record.counter, record.request_id), - ) - self.count += 1 - - -class ContextManifests(Carry): - staging_reserve_bytes: ClassVar[int] = 1024**3 - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - return ( - root / "bus.jsonl", - root / "bus_meta.json", - root / "private_bus_checkpoint.sqlite3", - ContextStore.path_in(root), - *cls.derived_files(root), - ) - - @classmethod - def derived_files(cls, root: Path) -> tuple[Path, ...]: - # Exact old disposable formats, never sessions, annotations or claims. - patterns = ( - "bus_page_index.sqlite3*", - "wake_candidates.sqlite3*", - "bus_route_counts.v*.sqlite3*", - "bus_activity_*.json", - "bus_display_*.json", - ) - return tuple(path for pattern in patterns for path in root.glob(pattern)) - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - path = root / "private_bus_checkpoint.sqlite3" - if not path.exists(): - return Carried(0) - with closing(_connect(path, readonly=True)) as db: - if not RecordedContextManifestSources.present(db): - return Carried(0) - count = RecordedContextManifestSources.count(db) - if not apply: - return Carried(count + 1) - return cls._rewrite(root) - - @classmethod - def _rewrite(cls, root: Path) -> Carried: - """Convert the selected stopped root under Carry's existing wire lock. - - The old index proves the bus bytes before they move; it is never copied. - The rewritten bus gets a fresh index derived from its own rows, so no old - table, offset or seal survives the conversion. Practice runs select an - already copied root through the normal Carry entrypoint. - """ - log = WireLog(root / "bus.jsonl") - index_path = WritableCheckpoint.path_in(log.path.parent) - required = log.path.stat().st_size + index_path.stat().st_size + cls.staging_reserve_bytes - filesystem = os.statvfs(root) - available = filesystem.f_bavail * filesystem.f_frsize - if required > available: - raise OSError( - f"Context conversion requires {required} bytes; " - f"available {available} bytes on {root}" - ) - if ContextStore.path_in(root).exists(): - raise RelationViolationError("Context conversion needs an absent destination store") - marker = log.private_marker_unlocked() - with closing(_connect(index_path, readonly=True)) as source_db: - witness = CheckedPrefix.read_witness(source_db) - with ExitStack() as lifetime: - stream = lifetime.enter_context(log.path.open("rb")) - witness.require_source(marker, os.fstat(stream.fileno()), stream) - stage = Path( - lifetime.enter_context( - tempfile.TemporaryDirectory(prefix=".move-context-", dir=root) - ) - ) - # Failed preparation must not leave a partial destination store. - # Carry's installer backup owns recovery once publication starts. - store = ContextStore(stage) - output_path = stage / "bus.jsonl" - digest = _SEED - with store.writing() as context_db, output_path.open("w+b") as output: - os.chmod(output_path, 0o600) - move = ContextMove(context_db, store) - stream.seek(0) - for offset, raw in WireRecord.rows(stream, end=witness.offset): - digest = _chain(digest, raw) - value = FieldCodec.decode(dict[str, object], FieldCodec.parse(raw)) - if FieldCodec.record_key(ContextEnvelope, "observation") in value: - FieldCodec.decode(ContextEnvelope, value).observation.move(move, offset) - else: - # Keep every original message byte, including its kind label. - output.write(raw) - if digest.hex() != witness.digest: - raise RelationViolationError( - "Context conversion did not preserve the complete proved prefix" - ) - output.flush() - os.fsync(output.fileno()) - size = output.tell() - derived = cls.derived_files(root) - for path in derived: - path.unlink() - index_path.unlink() - os.replace(output_path, log.path) - os.replace(store.path, ContextStore.path_in(root)) - _directory_sync(log.path) - CheckedPrefix._install(log, marker) - return Carried( - move.count + 1, - {"bus_bytes": size, "requests": move.count}, - ( - "bus.jsonl", - "bus_meta.json", - "private_bus_checkpoint.sqlite3", - store.filename, - *(path.name for path in derived), - ), - ) diff --git a/tools/cutover/carry_goal_execution.py b/tools/cutover/carry_goal_execution.py deleted file mode 100644 index 1484d28804..0000000000 --- a/tools/cutover/carry_goal_execution.py +++ /dev/null @@ -1,76 +0,0 @@ -"""Remove backend-produced registry blocks once; explicit decisions stay stored. - -The retired texts belong only to this cutover. Delete this step after rollout; -no turn, retry, timer or installation hook uses them to authorize execution. -""" - -from __future__ import annotations - -from dataclasses import replace -from pathlib import Path - -from agent_comms.goal_states import ActiveGoal, BlockedGoal -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import RegistryFile -from agent_comms.registry_store import RegistryStore - -FAILED_BACKEND_REASON = "Backend turn failed; inspect local diagnostics before resuming." -BACKEND_REASONS = frozenset( - { - FAILED_BACKEND_REASON, - ( - "Backend reported success without assistant output or tool activity; inspect " - "the session before resuming." - ), - "Goal turn ended without verified terminal progress.", - "Backend turn ended without a result; inspect local diagnostics.", - "Goal origin turn did not finish successfully.", - ( - "Standby wake launch authority unavailable; inspect the UNKNOWN input before " - "explicit Retry. No input was replayed." - ), - ( - "The interrupted goal attempt is unresolved. Inspect it, then use Retry to " - "authorize a new attempt. Your messages can still be sent." - ), - "Goal attempt unresolved; inspect diagnostics before Retry.", - } -) - - -class BackendGoalBlocks(Carry): - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - store = RegistryStore(RegistryFile.path_in(root)) - paths = tuple(path for path in store.written if path.exists()) - for path in paths: - cls.require_confined(root, path) - return paths - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - store = RegistryStore(RegistryFile.path_in(root)) - cls.require_confined(root, store.path) - if not store.path.exists(): - return Carried(0) - migrated = preserved = 0 - with store.editing() as edit: - for owner in tuple(edit.document.threads.values()): - for goal in owner.goal: - if isinstance(goal.state, BlockedGoal): - if goal.state.block_reason in BACKEND_REASONS: - migrated += 1 - active = replace(goal, state=ActiveGoal(), revision=goal.revision + 1) - edit.document.install_thread(replace(owner, goal=active)) - else: - preserved += 1 - if apply and migrated: - edit.commit() - rewritten = ( - tuple(str(path.relative_to(root)) for path in store.written) - if apply and migrated - else () - ) - return Carried( - migrated, detail={"migrated": migrated, "preserved": preserved}, rewritten=rewritten - ) diff --git a/tools/cutover/carry_goal_work_decision.py b/tools/cutover/carry_goal_work_decision.py deleted file mode 100644 index dab6d476e4..0000000000 --- a/tools/cutover/carry_goal_work_decision.py +++ /dev/null @@ -1,57 +0,0 @@ -"""Rename the durable goal decision table once; keep its rows and launch fences. - -The old spelling exists only at this carry boundary. Core derives the new name -from GoalWorkDecision. Delete this step after every retained bus has run it. -""" -from __future__ import annotations - -from contextlib import closing -from pathlib import Path - -from agent_comms.errors import RelationViolationError -from agent_comms.goal_attempts import GoalAttemptStore, GoalWorkDecision -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import SQLiteFile -from agent_comms.typed_table import SqlValue - - -RETIRED_TABLE = "goal_human_decision" - - -class GoalWorkDecisions(Carry): - """Carry the table after old owners stop, before the new runtime reads it.""" - - @staticmethod - def ledger_path(root: Path) -> Path: - return GoalAttemptStore.path_in(root / GoalAttemptStore.root_directory) - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - paths = tuple(path for path in SQLiteFile.files(cls.ledger_path(root)) if path.exists()) - for path in paths: - cls.require_confined(root, path) - return paths - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - path = cls.ledger_path(root) - cls.require_confined(root, path) - if not path.exists(): - return Carried(0) - with closing(GoalAttemptStore.database(path)) as db: - with SQLiteFile.write_transaction(db): - retired_present = SqlValue.for_annotation(bool).read(db.execute( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?)", - (RETIRED_TABLE,), - ))[0] - current_present = GoalWorkDecision.present(db) - if retired_present and current_present: - raise RelationViolationError("Both goal decision tables exist; refuse to choose rows") - if not retired_present: - if not current_present: - raise RelationViolationError("The goal ledger has no decision table") - return Carried(0) - if apply: - db.execute(f'ALTER TABLE "{RETIRED_TABLE}" RENAME TO "{GoalWorkDecision.declared_name}"') - rewritten = (str(path.relative_to(root)),) if apply else () - return Carried(1, rewritten=rewritten) diff --git a/tools/cutover/carry_install_schedule.py b/tools/cutover/carry_install_schedule.py deleted file mode 100644 index 676734a741..0000000000 --- a/tools/cutover/carry_install_schedule.py +++ /dev/null @@ -1,9 +0,0 @@ -"""Delete the retired install schedule; installed build identity now answers what is due.""" - -from agent_comms.installer.carry import RetiredFiles - - -class InstallSchedule(RetiredFiles): - @classmethod - def retired(cls) -> tuple[str, ...]: - return ("install-deadline.json",) diff --git a/tools/cutover/carry_native_input_proofs.py b/tools/cutover/carry_native_input_proofs.py deleted file mode 100644 index 879c493a5e..0000000000 --- a/tools/cutover/carry_native_input_proofs.py +++ /dev/null @@ -1,353 +0,0 @@ -"""Carry bus-contained resources; reconcile explicitly selected external history. - -The source root never supplies coordinator or target authority. Original external -journals are read-only; only the bus root owns every target path. -""" -from __future__ import annotations - -import argparse -import json -from abc import abstractmethod -from contextlib import ExitStack, closing, contextmanager -from dataclasses import dataclass -from pathlib import Path -import sqlite3 -from typing import Annotated, Iterator, Literal - -from agent_comms.coordination_database import CoordinationStore -from agent_comms.coordination_tables.participants import OwnerGenerations -from agent_comms.errors import RelationViolationError -from agent_comms.field_codec import FieldCodec, PathText -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import SQLiteFile -from agent_comms.native_pi import NativeContextJournal -from agent_comms.native_runtime_input import NativeRuntimeInput -from agent_comms.native_session_files import NativeInputProof, NativeSessionFiles -from agent_comms.native_session_reopen import NativeSessionIdentity -from agent_comms.private_path import NativeSessionFileRole, fsync_directory - - -@dataclass(frozen=True, slots=True, kw_only=True) -class LegacyNativeProof(NativeSessionFiles): - """The retired spelling and its original coordinator-owned participant.""" - - owner: OwnerGenerations - - @property - def source(self) -> Path: - return self.session.with_name(self.session.name + ".input-proof") - - @property - def source_files(self) -> tuple[Path, ...]: - return tuple(path for path in SQLiteFile.files(self.source) if path.exists()) - - def require_sources(self, journal_root: Path) -> None: - for source in (*self.source_files, self.session): - Carry.require_confined(journal_root, source) - NativeSessionFileRole.require(source.lstat()) - - def require_targets(self) -> None: - for target in SQLiteFile.files(self.input_proof): - Carry.require_confined(self.root, target) - - -class ProofRelocation(LegacyNativeProof): - def require_movable(self) -> None: - self.require_sources(self.root) - self.require_targets() - for destination in SQLiteFile.files(self.input_proof): - if destination.exists() or destination.is_symlink(): - raise RelationViolationError(f"Native proof destination already exists: {destination}") - - def relocate(self) -> tuple[Path, ...]: - """Move bus-contained companions before the primary; roll back failures.""" - present = self.source_files - if self.source not in present: - raise FileNotFoundError(self.source) - self.prepare_private_directory(self.input_proof.parent) - moves = tuple(zip(SQLiteFile.files(self.source), SQLiteFile.files(self.input_proof), strict=True)) - moved: list[Path] = [] - with ExitStack() as rollback: - for source, destination in reversed(moves): - if source in present: - source.rename(destination) - rollback.callback(fsync_directory, destination.parent) - rollback.callback(fsync_directory, source.parent) - rollback.callback(destination.rename, source) - moved.append(source) - fsync_directory(destination.parent) - fsync_directory(source.parent) - rollback.pop_all() - return (*moved, *(path for path in SQLiteFile.files(self.input_proof) if path.exists())) - - -@dataclass(frozen=True) -class NativeProofDifference: - missing_rows: int - conflicting_keys: int - - def require_compatible(self) -> None: - if self.conflicting_keys: - raise RelationViolationError( - f"Native proof history has {self.conflicting_keys} conflicting keys" - ) - - -@dataclass(frozen=True) -class ProofHistoryReport: - source: Annotated[Path, PathText] - target: Annotated[Path, PathText] - owner: OwnerGenerations - difference: NativeProofDifference - inserted_rows: int = 0 - message: str = "existing target" - - -class ProofHistory(LegacyNativeProof): - def report(self, difference: NativeProofDifference, *, inserted: int = 0, - message: str = "existing target") -> ProofHistoryReport: - return ProofHistoryReport(self.source, self.input_proof, self.owner, - difference, inserted, message) - - @abstractmethod - def inspect(self, journal_root: Path) -> ProofHistoryReport: ... - - @abstractmethod - def apply(self, journal_root: Path) -> ProofHistoryReport: ... - - -class MissingProofTarget(ProofHistory): - def inspect(self, journal_root: Path) -> ProofHistoryReport: - self.require_sources(journal_root) - self.require_targets() - with closing(CoordinationStore.read_snapshot(self.source)) as db: - rows = NativeContextJournal.count(db) - return self.report(NativeProofDifference(rows, 0), message="skipped: target absent; original preserved") - - def apply(self, journal_root: Path) -> ProofHistoryReport: - return self.inspect(journal_root) - - -class ExistingProofTarget(ProofHistory): - history_schema = "native_history" - - @contextmanager - def connection(self, journal_root: Path, mode: Literal["ro", "rw"]) -> Iterator[sqlite3.Connection]: - """SQLite URI modes are its external protocol, not domain dispatch.""" - self.require_sources(journal_root) - self.require_targets() - NativeSessionFileRole.require(self.input_proof.lstat()) - with closing(sqlite3.connect(self.input_proof.as_uri() + f"?mode={mode}", uri=True, - timeout=30)) as db: - db.execute("PRAGMA synchronous=EXTRA") - NativeContextJournal.require_schema(db) - db.execute(f"ATTACH DATABASE ? AS {self.history_schema}", - (self.source.as_uri() + "?mode=ro",)) - yield db - - def difference(self, db: sqlite3.Connection) -> NativeProofDifference: - """Derive every key and compared column from the existing declaration.""" - table = NativeContextJournal - same_key = " AND ".join( - f'past."{item.wire_name}" = present."{item.wire_name}"' - for item in table._fields() if item.column.primary_key - ) - different = " OR ".join( - f'past."{item.wire_name}" IS NOT present."{item.wire_name}"' - for item in table._fields() - ) - past = f"{self.history_schema}.{table.declared_name}" - present = f"main.{table.declared_name}" - missing = db.execute( - f"SELECT COUNT(*) FROM {past} AS past WHERE NOT EXISTS " - f"(SELECT 1 FROM {present} AS present WHERE {same_key})" - ).fetchone()[0] - conflicts = db.execute( - f"SELECT COUNT(*) FROM {past} AS past JOIN {present} AS present " - f"ON {same_key} WHERE {different}" - ).fetchone()[0] - return NativeProofDifference(missing, conflicts) - - def validate(self, db: sqlite3.Connection) -> None: - """Validate immutable historical rows through the existing schema in memory. - - Independent journal lifetimes can reuse generations. The live append - policy is not a rule for restoring older keyed evidence. No input runs. - """ - self.difference(db).require_compatible() - table = NativeContextJournal - columns = table._column_list(tuple(table.column_projection())) - rows = table.read(db.execute( - f"SELECT {columns} FROM {self.history_schema}.{table.declared_name} UNION " - f"SELECT {columns} FROM main.{table.declared_name}" - )) - with closing(sqlite3.connect(":memory:")) as validation: - table.create_schema(validation) - for name in table.trigger_bodies(): - validation.execute(f'DROP TRIGGER "{name}"') - for row in rows: - if row.session_id != self.session_id: - raise RelationViolationError("Native proof history belongs to another session") - row.insert(validation) - - def merge(self, db: sqlite3.Connection) -> int: - # Decode outside the writer lock; recheck every key inside the transaction. - self.validate(db) - table = NativeContextJournal - with db: - db.execute("BEGIN IMMEDIATE") - difference = self.difference(db) - difference.require_compatible() - if not difference.missing_rows: - return 0 - # Transactional DDL restores triggers on rollback too. Only the live - # append policy is bypassed, never the declared row checks/keys. - triggers = table.trigger_bodies() - for name in triggers: - db.execute(f'DROP TRIGGER "{name}"') - columns = table._column_list(tuple(table.column_projection())) - inserted = db.execute( - f"INSERT INTO {table.declared_name} ({columns}) " - f"SELECT {columns} FROM {self.history_schema}.{table.declared_name} " - f"EXCEPT SELECT {columns} FROM main.{table.declared_name}" - ).rowcount - declarations = table.schema_objects() - for name in triggers: - db.execute(declarations[name]) - return inserted - - def inspect(self, journal_root: Path) -> ProofHistoryReport: - with self.connection(journal_root, "ro") as db: - difference = self.difference(db) - return self.report(difference) - - def apply(self, journal_root: Path) -> ProofHistoryReport: - with self.connection(journal_root, "rw") as db: - difference = self.difference(db) - inserted = self.merge(db) - return self.report(difference, inserted=inserted) - - -class NativeInputProofs(Carry): - @classmethod - def sources(cls, root: Path, journal_root: Path) -> tuple[LegacyNativeProof, ...]: - found = set(journal_root.rglob("*.input-proof*")) - primaries = sorted(path for path in found if path.name.endswith(".input-proof")) - coordinator = CoordinationStore.path_in(root) - for path in SQLiteFile.files(coordinator): - cls.require_confined(root, path) - plans: list[LegacyNativeProof] = [] - if primaries: - with closing(CoordinationStore.read_snapshot(coordinator)) as db: - for primary in primaries: - cls.require_confined(journal_root, primary) - source = primary.with_name(primary.name.removesuffix(".input-proof")) - cls.require_confined(journal_root, source) - NativeSessionFileRole.require(source.lstat()) - identity = NativeSessionIdentity.locate(str(source)) - inputs = NativeRuntimeInput.select(db, session_id=identity.session_id) - owners = {item.owner_identity.lookup: item.owner_identity for item in inputs} - if len(owners) != 1: - raise RelationViolationError( - f"Native proof session has no unique recorded owner: {source} ({len(owners)} owners)" - ) - plan = LegacyNativeProof.for_identity(identity, root=root, owner=next(iter(owners.values()))) - plan.require_sources(journal_root) - plan.require_targets() - plans.append(plan) - declared = {path for plan in plans for path in plan.source_files} - if stray := found - declared: - raise RelationViolationError(f"Native proof has unrecognized companions: {sorted(stray)}") - targets = [plan.input_proof for plan in plans] - if len(set(targets)) != len(targets): - raise RelationViolationError("Multiple native proofs name the same participant/session") - return tuple(plans) - - @classmethod - def relocations(cls, root: Path) -> tuple[ProofRelocation, ...]: - root = root.resolve(strict=True) - plans = tuple(ProofRelocation.for_identity(source, root=root, owner=source.owner) - for source in cls.sources(root, root)) - for plan in plans: - plan.require_movable() - return plans - - @classmethod - def histories(cls, root: Path, journal_root: Path) -> tuple[ProofHistory, ...]: - root = root.resolve(strict=True) - journal_root = journal_root.resolve(strict=True) - plans: list[ProofHistory] = [] - for source in cls.sources(root, journal_root): - # Decode filesystem presence once; consumers use the resulting capability. - kind = ExistingProofTarget if source.input_proof.exists() else MissingProofTarget - plans.append(kind.for_identity(source, root=root, owner=source.owner)) - return tuple(plans) - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - root = root.resolve(strict=True) - return tuple(path for plan in cls.relocations(root) for path in plan.source_files) + tuple( - path for primary in NativeInputProof.matching_in(root) - for path in SQLiteFile.files(primary) if path.exists() - ) - - @classmethod - def differing(cls, root: Path) -> tuple[Path, ...]: - """Proof journals whose stored schema is not their declaration; a dry run counts them - too, since a step's dry run is what says whether it runs.""" - found = [] - for path in sorted(NativeInputProof.matching_in(root)): - cls.require_confined(root, path) - NativeSessionFileRole.require(path.lstat()) - with closing(CoordinationStore.recovering_snapshot(path)) as db: - if NativeContextJournal.schema_differs(db): - found.append(path) - return tuple(found) - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - root = root.resolve(strict=True) - plans = cls.relocations(root) - if not apply: - return Carried(len(plans) + len(cls.differing(root))) - written = {path for plan in plans for path in plan.relocate()} - converted = cls.differing(root) - for path in converted: - with closing(CoordinationStore.writer(path, isolation_level=None)) as db: - NativeContextJournal.reconcile_schema(db) - written.update(SQLiteFile.files(path)) - return Carried(len(plans) + len(converted), - {"proofs": len(plans), "schemas": len(converted)}, - tuple(str(path.relative_to(root)) for path in sorted(written))) - - -class PreviewHistory: - @classmethod - def run(cls, root: Path, journal_root: Path) -> tuple[ProofHistory, ...]: - plans = NativeInputProofs.histories(root, journal_root) - for plan in plans: - print(json.dumps(FieldCodec.encode(plan.inspect(journal_root))), flush=True) - return plans - - -class ApplyHistory(PreviewHistory): - @classmethod - def run(cls, root: Path, journal_root: Path) -> tuple[ProofHistory, ...]: - plans = super().run(root, journal_root) # Print every comparison before any write. - for plan in plans: - print(json.dumps(FieldCodec.encode(plan.apply(journal_root))), flush=True) - return plans - - -def main() -> None: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--bus-root", type=Path, required=True) - parser.add_argument("--journal-root", type=Path, required=True) - parser.add_argument("--apply", dest="operation", action="store_const", const=ApplyHistory, - default=PreviewHistory) - args = parser.parse_args() - args.operation.run(args.bus_root, args.journal_root) - - -if __name__ == "__main__": - main() diff --git a/tools/cutover/carry_order_scopes.py b/tools/cutover/carry_order_scopes.py deleted file mode 100644 index 501e37cec7..0000000000 --- a/tools/cutover/carry_order_scopes.py +++ /dev/null @@ -1,83 +0,0 @@ -"""Carry order scopes through agent-comms install, after old owners stop. - -Delete this step once every bus has run it. Historical owner references are -preserved, not reclassified against today's source; fresh fix orders validate -their class declarations. No resident compatibility reader is installed. -""" -from __future__ import annotations - -import json -from contextlib import closing, nullcontext -from dataclasses import dataclass -from pathlib import Path -from typing import Annotated - -from agent_comms.change_report import FileLine -from agent_comms.coordination_database import CoordinationStore -from agent_comms.field_codec import FieldCodec -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import SQLiteFile -from agent_comms.todos import CheckedProblemOrder, FixScope, FixShape, ProblemOrder, Todo, TodoStore, TriageScope -from agent_comms.typed_table import Column, JsonStorage, TextStorage, TypedRow - -TRIAGE_PROBLEMS = frozenset({"P362", "P377", "P378", "P379"}) - - -@dataclass(frozen=True) -class StoredOrder(TypedRow): - """SQLite decodes one pre-scope declaration selected by its family's tag.""" - - problem: str - path: str - owner: Annotated[FileLine, Column(storage=TextStorage)] - - def scope_json(self) -> str: - scope = (TriageScope() if self.problem in TRIAGE_PROBLEMS - else FixScope(FixShape(parents=(self.owner,), members=()))) - return json.dumps(FieldCodec.encode(scope)) - - -class OrderScopes(Carry): - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - paths = tuple(path for path in SQLiteFile.files(TodoStore.path_in(root)) if path.exists()) - for path in paths: - cls.require_confined(root, path) - return paths - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - path = TodoStore.path_in(root) - cls.require_confined(root, path) - if not path.exists(): - return Carried(0) - kinds = tuple(member.declared_name for member in ProblemOrder.members_with(CheckedProblemOrder)) - placeholders = ",".join("?" for _ in kinds) - table = Todo.declared_name - changed = 0 - identity = Todo._column_list(("id",)) - connection = CoordinationStore.writer(path) if apply else CoordinationStore.read_snapshot(path) - with closing(connection) as db: - with SQLiteFile.write_transaction(db) if apply else nullcontext(db): - for declared in Todo._fields(): - if not issubclass(declared.storage, JsonStorage): - continue - column = Todo._column_list((declared.name,)) - orders = StoredOrder.read(db.execute( - f'SELECT todo.{identity} AS problem, entry.path AS path, ' - f"json_extract(todo.{column}, entry.path || '.owner') AS owner " - f'FROM "{table}" AS todo, json_tree(todo.{column}) AS entry ' - f"WHERE entry.key='kind' AND entry.value IN ({placeholders}) " - f"AND json_type(todo.{column}, entry.path || '.scope') IS NULL", kinds)) - for order in orders: - if apply: - db.execute( - f'UPDATE "{table}" SET {column}=json_set({column}, ?, json(?)) WHERE {identity}=?', - (order.path + ".scope", order.scope_json(), order.problem)) - changed += 1 - if apply: - # Decode every resulting canonical row before committing any conversion. - for _ in Todo.stream(db): - pass - rewritten = tuple(str(file.relative_to(root)) for file in SQLiteFile.files(path)) if apply and changed else () - return Carried(changed, rewritten=rewritten) diff --git a/tools/cutover/carry_proof_build_commits.py b/tools/cutover/carry_proof_build_commits.py deleted file mode 100644 index 3ea3f00ff6..0000000000 --- a/tools/cutover/carry_proof_build_commits.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Clear incomplete old proof pairs once; never infer their Textual revision. - -Run through agent-comms install after owners stop. Delete this step once every -bus has run it; no old-proof reader remains in production. -""" - -from __future__ import annotations - -import json -from contextlib import closing, nullcontext -from dataclasses import dataclass -from pathlib import Path - -from agent_comms.build_identity import BuildIdentity, RecordedBuildIdentity, UnrecordedBuildIdentity -from agent_comms.coordination_database import CoordinationStore -from agent_comms.field_codec import FieldCodec -from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import SQLiteFile -from agent_comms.todos import CheckedDelivery, OrderDelivery, Todo, TodoStore -from agent_comms.typed_table import JsonStorage, TypedRow - - -@dataclass(frozen=True) -class StoredProofBuild(TypedRow): - problem: str - path: str - incomplete: bool - - -class ProofBuildCommits(Carry): - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - paths = tuple(path for path in SQLiteFile.files(TodoStore.path_in(root)) if path.exists()) - for path in paths: - cls.require_confined(root, path) - return paths - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - path = TodoStore.path_in(root) - cls.require_confined(root, path) - if not path.exists(): - return Carried(0, {"cleared": 0, "unchanged": 0}) - cleared = unchanged = 0 - table = Todo.declared_name - identity = Todo._column_list(("id",)) - kinds = OrderDelivery.members_with(CheckedDelivery) - placeholders = ",".join("?" for _ in kinds) - unrecorded = json.dumps(FieldCodec.encode(UnrecordedBuildIdentity())) - connection = ( - CoordinationStore.writer(path) if apply else CoordinationStore.read_snapshot(path) - ) - with closing(connection) as db: - with SQLiteFile.write_transaction(db) if apply else nullcontext(db): - for declared in Todo._fields(): - if not issubclass(declared.storage, JsonStorage): - continue - column = Todo._column_list((declared.name,)) - proofs = tuple( - StoredProofBuild.read( - db.execute( - f"SELECT todo.{identity} AS problem, entry.fullkey AS path, " - f"(json_extract(entry.value, '$.kind') IS ? " - f"AND json_type(entry.value, '$.core') IS 'text' " - f"AND json_type(entry.value, '$.toad') IS 'text') AS incomplete " - f'FROM "{table}" AS todo, json_tree(todo.{column}) AS entry ' - f"WHERE entry.key='proof_build' " - f"AND json_extract(todo.{column}, entry.path || '.kind') IN ({placeholders})", - ( - FieldCodec.encode(RecordedBuildIdentity, type[BuildIdentity]), - *( - FieldCodec.encode(member, type[OrderDelivery]) - for member in kinds - ), - ), - ) - ) - ) - for proof in proofs: - cleared += proof.incomplete - unchanged += not proof.incomplete - if apply and proof.incomplete: - db.execute( - f'UPDATE "{table}" SET {column}=json_set({column}, ?, json(?)) WHERE {identity}=?', - (proof.path, unrecorded, proof.problem), - ) - if apply: - # Reject the whole transaction if any resulting canonical row cannot decode. - for _ in Todo.stream(db): - pass - rewritten = ( - tuple(str(file.relative_to(root)) for file in SQLiteFile.files(path)) - if apply and cleared - else () - ) - return Carried(cleared, {"cleared": cleared, "unchanged": unchanged}, rewritten) diff --git a/tools/cutover/carry_retired_ledger.py b/tools/cutover/carry_retired_ledger.py index 8d5bb39d11..166911282d 100644 --- a/tools/cutover/carry_retired_ledger.py +++ b/tools/cutover/carry_retired_ledger.py @@ -1,9 +1,20 @@ """Delete the carry ledger: each step's own dry run now says whether a bus needs it.""" -from agent_comms.installer.carry import RetiredFiles +from pathlib import Path +from agent_comms.installer.carry import Carried, Carry + + +class RetiredLedger(Carry): + @classmethod + def backup_files(cls, root: Path) -> tuple[Path, ...]: + return (root / "carries.json",) -class RetiredLedger(RetiredFiles): @classmethod - def retired(cls) -> tuple[str, ...]: - return ("carries.json",) + def carry(cls, root: Path, *, apply: bool) -> Carried: + paths = tuple(path for path in cls.backup_files(root) if path.exists()) + for path in paths: + cls.require_confined(root, path) + if apply: + path.unlink() + return Carried(len(paths), {}, tuple(path.name for path in paths) if apply else ()) diff --git a/tools/cutover/carry_review_timers.py b/tools/cutover/carry_review_timers.py deleted file mode 100644 index 37e9f00fb4..0000000000 --- a/tools/cutover/carry_review_timers.py +++ /dev/null @@ -1,119 +0,0 @@ -"""Todo rows trade the order deadline for the commander's review timer, once. - -The deadline column becomes ``review_at`` with the same instant: the first review is when the -order used to be due. The clock strikes are deleted from Core, so a recorded stall, missed -deadline or missed first call becomes the plain strike it was recorded as, with its reason, -time, actor and evidence intact; a void of one still names the same strike. Core's watcher -decodes under the name its class declares, so its old spelling is rewritten by the same pass. -Delete once every bus has run it. -""" - -from __future__ import annotations - -from contextlib import closing -from dataclasses import dataclass, fields, replace -from functools import singledispatch -from pathlib import Path -from types import NoneType - -from agent_comms.coordination_database import CoordinationStore -from agent_comms.installer.carry import Carried, Carry -from agent_comms.todos import Assignment, StrikeTodo, Todo, TodoStore - - -# The column the order deadline was stored in; the review timer keeps its instant. -RETIRED_COLUMN = "deadline" - - -# These old declarations exist only at this one-time format boundary: the strike kinds Core -# no longer declares, so their stored rows decode before they are carried. -@dataclass(frozen=True, kw_only=True) -class ClockStrike(StrikeTodo, declared_name=None): - """A strike Core used to record from a clock; it is carried as the plain strike it was.""" - - -@dataclass(frozen=True, kw_only=True) -class StallStrike(ClockStrike): - pass - - -@dataclass(frozen=True, kw_only=True) -class MissedDeadlineStrike(ClockStrike): - order_id: str - - -@dataclass(frozen=True, kw_only=True) -class MissedFirstCallStrike(ClockStrike): - order_id: str - - -@singledispatch -def assigned(assignment: Assignment) -> Assignment: - """An assignment flattens its parent to a name; the parent's decoded identity names it.""" - return replace(assignment, parent=assignment.parent_identity.name) - - -@assigned.register -def _(unassigned: NoneType) -> None: - """A row nobody holds carries no assignment.""" - - -@singledispatch -def carried(strike: StrikeTodo) -> StrikeTodo: - """Every strike Core still declares keeps its kind.""" - return replace(strike, previous=assigned(strike.previous)) - - -@carried.register -def _(strike: ClockStrike) -> StrikeTodo: - return carried(StrikeTodo(**{item.name: getattr(strike, item.name) - for item in fields(StrikeTodo)})) - - -def carried_todo(todo: Todo) -> Todo: - return replace( - todo, assignment=assigned(todo.assignment), - strikes=tuple(map(carried, todo.strikes)), - voided=tuple(replace(void, strike=carried(void.strike)) for void in todo.voided), - released=tuple(replace(release, previous=assigned(release.previous)) - for release in todo.released)) - - -def carry(root: Path, *, apply: bool) -> int: - path = TodoStore.path_in(root) - if not path.exists(): - return 0 - with closing(CoordinationStore.writer(path)) as db: - db.execute("BEGIN IMMEDIATE") - try: - if RETIRED_COLUMN not in Todo.stored_columns(db): - db.rollback() - return 0 - Todo.rename_stored_column(db, RETIRED_COLUMN, "review_at") - todos = tuple(map(carried_todo, Todo.stream(db))) - Todo.drop(db) - Todo.create(db) - for todo in todos: - todo.upsert(db) - if apply: - db.commit() - else: - db.rollback() - return len(todos) - except BaseException: - db.rollback() - raise - - -class ReviewTimers(Carry): - """Todo rows keep their instant as the review timer; clock strikes become plain strikes.""" - - @classmethod - def backup_files(cls, root: Path) -> tuple[Path, ...]: - return tuple(root.glob(f"{TodoStore.path_in(root).name}*")) - - @classmethod - def carry(cls, root: Path, *, apply: bool) -> Carried: - changed = carry(root, apply=apply) - rewritten = tuple(path.name for path in root.glob(f"{TodoStore.path_in(root).name}*")) - return Carried(changed, {}, rewritten if changed and apply else ()) diff --git a/tools/cutover/connector_browser_trust.py b/tools/cutover/connector_browser_trust.py deleted file mode 100644 index 442340d722..0000000000 --- a/tools/cutover/connector_browser_trust.py +++ /dev/null @@ -1,34 +0,0 @@ -"""One-shot P380 cutover, with the connector stopped and grants.json backed up. - -PYTHONPATH=src python tools/cutover/connector_browser_trust.py PATH/TO/grants.json - -The installer operator deletes this tool after the live cutover. It is deliberately -not a resident reader and does not infer browser trust from an old client approval. -""" -from __future__ import annotations - -import argparse -from pathlib import Path - -from agent_comms.connector_store import ConnectorGrants, ConnectorStore -from agent_comms.field_codec import FieldCodec - - -def convert(store: ConnectorStore) -> None: - """Discard waiting requests, retaining every client/code/access/refresh grant.""" - with store.locked(): - grants = store._read_using( - lambda raw: FieldCodec.decode_document(ConnectorGrants, {**raw, "requests": {}}) - ) - store._publish_unlocked(grants) - - -def main() -> None: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("path", type=Path) - arguments = parser.parse_args() - convert(ConnectorStore(arguments.path)) - - -if __name__ == "__main__": - main()