Durable stores derive their declared schema on the writer's open

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:

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:

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 ())

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:

src/agent_comms/typed_table.py

@@ -588,6 +588,75 @@ 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."""
+
+    @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):
+    @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):
+    pass
+
+
+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 +715,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 +810,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 +858,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 +1003,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 +1349,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")

tests/test_commander_forks.py

@@ -238,35 +238,25 @@ 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')
     with store.writing() as db:
         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):