Avoid replaying bus history for catalog-only schema changes

plans/commander-native-plan.md

@@ -235,12 +235,47 @@ takes 76 SQLite instructions after index reconciliation versus 45,088 before;
 both return no message. All exact-pair queries use the declared covering index.
 Reconciliation preserves every bus sequence and both bus/address row counts;
 integrity and foreign-key checks pass, and a repeated reconciliation is a no-op.
-Its current unconditional projection rebuild costs 39.83 seconds even for this
-index-only change; that installer/schema defect remains open with the commander.
+The baseline unconditional projection rebuild costs 39.83 seconds even for this
+index-only change; the separate schema repair below removes it.
 Installed full-session/parity acceptance and the complete control-plane phase
 remain open. The source census has zero parse omissions; unprojected declarations
 remain visible.
 
+The schema follow-through, draft #2629, is stacked on #2624. `ComposedSchemas`
+now derives preceding row-write counts from SQLite's existing connection;
+`DerivedTables` owns whether changed source rows or its own changed table shape
+require replay. Catalog-only changes use the existing schema derivation without
+resetting projection contents. On the same retained bus copy, adding the index
+now takes 24.845 milliseconds with zero row writes; every source/projection count
+is preserved, integrity and foreign-key checks pass, and a second reconciliation
+is a no-op. Four focused controls cover source/projection indexes, source-row
+conversion with unchanged projection schema, changed projection tables and
+explicit content repair. Two existing held-CLI cases confirm both composed
+schemas reconcile through the actual installer command path.
+
+An open invalidation defect remains: installed decoder/projector behavior changes
+with identical stored DDL can silently retain stale projections. No production
+path guarantees the explicit reset that the earlier wording assumed. Planned
+reconciliation checks DDL only; recorded build identity has no projection
+invalidation consumer. Current explicit reset callers are schema carry and
+tests, not an installation step for behavior changes. Combining a producer
+behavior change with an index change exposes a regression: the old composed
+carry rebuilt those projections, while #2629 can skip them. The row counter
+does not establish unchanged producing behavior. That premise is invalidated;
+#2629 remains draft, unaccepted, and outside the current installation.
+
+No actual unchanged-DDL stale-output case was found in the checked batch:
+#2624 to #2629 changes only schema carry, tests and this plan; the installed
+`d9ce9cd95` to integration `883223ed4` transition creates the SQLite bus and its
+projection schemas through the history conversion. Its core projector and
+response-policy methods do not change between the published owner-request
+parent and that integration head. This does not establish generic skip safety.
+The same schema-repair owner is tracing the persisted response-policy answer
+and its live role owner to remove the competing decision. The automatic
+heartbeat/index change remains separate from this rejected optimization route.
+The earlier heartbeat checkpoint remains independently reviewable; installed
+qualification belongs to the integration owner.
+
 ### Phase 8: cutover
 
 **Changes.**

src/agent_comms/typed_table.py

@@ -789,6 +789,15 @@ def stored_tables(cls, db: sqlite3.Connection) -> list[SQLiteTableName]:
     def schema_differs(cls, db: sqlite3.Connection) -> bool:
         return bool(SQLiteSchemaObject.differences(cls.stored_schema(db), cls.declared_schema()))
 
+    @classmethod
+    def table_schema_differs(cls, db: sqlite3.Connection) -> bool:
+        """Whether stored row shapes differ, independently of indexes, triggers and views."""
+        tables = frozenset(table.name for table in cls.stored_tables(db))
+        return bool(SQLiteSchemaObject.differences(
+            [item for item in cls.stored_schema(db) if item.name in tables],
+            {row.declared_name: row._table_sql() for row in cls.schema_rows()},
+        ))
+
     @classmethod
     def require_schema(cls, db: sqlite3.Connection) -> None:
         """Readers use only the declared schema, whichever kind of file this is."""
@@ -845,8 +854,8 @@ def install_schema(cls, db: sqlite3.Connection, change: SchemaChange) -> None:
 
     @classmethod
     @abstractmethod
-    def carry_schema(cls, db: sqlite3.Connection) -> None:
-        """Inside the reconcile's transaction: this kind of file's way onto its declaration."""
+    def carry_schema(cls, db: sqlite3.Connection, *, source_changes: int = 0) -> None:
+        """Carry this family's schema after the preceding families' row writes."""
 
 
 class ComposedSchemas:
@@ -863,9 +872,11 @@ def schema_rows(cls) -> tuple[type[TypedTable], ...]:
         ))
 
     @classmethod
-    def carry_schema(cls, db: sqlite3.Connection) -> None:
+    def carry_schema(cls, db: sqlite3.Connection, *, source_changes: int = 0) -> None:
+        """Carry in declaration order; later families see preceding row writes."""
+        before = db.total_changes - source_changes
         for family in cls.schema_families():
-            family.carry_schema(db)
+            family.carry_schema(db, source_changes=db.total_changes - before)
 
 
 class RowMutation(Enum):
@@ -1330,8 +1341,11 @@ def reset_schema(cls, db: sqlite3.Connection) -> None:
         cls.create_schema(db)
 
     @classmethod
-    def carry_schema(cls, db: sqlite3.Connection) -> None:
-        cls.reset_schema(db)
+    def carry_schema(cls, db: sqlite3.Connection, *, source_changes: int = 0) -> None:
+        if source_changes or cls.table_schema_differs(db):
+            cls.reset_schema(db)
+        else:
+            cls.derive_schema(db)
 
 
 class DurableTables(TableSchema):
@@ -1342,7 +1356,7 @@ class DurableTables(TableSchema):
     __slots__ = ()
 
     @classmethod
-    def carry_schema(cls, db: sqlite3.Connection) -> None:
+    def carry_schema(cls, db: sqlite3.Connection, *, source_changes: int = 0) -> None:
         cls.derive_schema(db)
 
 

tests/test_bus.py

@@ -15,9 +15,11 @@
 )
 from agent_comms.bus_publication import Delivered, HumanOrigin, KeyedPublication
 from agent_comms.bus_tables import (
+    Addressed,
     AnsweredSource,
     BusProjectionTables,
     BusRow,
+    BusSchema,
     BusTables,
     DirectReplySource,
     MessageIdentitySource,
@@ -41,7 +43,9 @@
 from agent_comms.response_acknowledgement import (
     canonical_publication_key,
 )
-from agent_comms.typed_table import RowNotFoundError, SQLiteTableName, TypedTable
+from agent_comms.typed_table import (
+    RowNotFoundError, SQLiteSchemaObject, SQLiteTableName, TypedTable, WriterOwnedFile,
+)
 from bus_record_fixture import publication_processes, record_from
 
 
@@ -84,6 +88,49 @@ def test_install_refuses_existing_schema_changes_until_an_explicit_rebuild(bus):
     assert bus.page(3, 2) == (second, record)
 
 
+def test_schema_carry_preserves_metadata_only_rows_and_rebuilds_invalidated_projections(bus):
+    record = bus.append(lambda seq, db: PublicRecord.from_message(
+        Message("author", "worker", "preserved source", InfoKind, seq=seq),
+    ))
+    with bus.writer_connection() as db:
+        # Both source and projection indexes are metadata, not reasons to replay history.
+        for table in (Addressed.table, MessageIdentitySource):
+            db.execute(f'DROP INDEX "{table.indexes[0].name(table)}"')
+            before = db.total_changes
+            assert BusSchema.reconcile_schema(db, WriterOwnedFile())
+            assert db.total_changes == before
+            assert MessageIdentitySource.require(db, seq=record.seq).message_id == record.message_id
+
+        # An unchanged projection schema still needs replay after a real source-row carry.
+        projected_schema = MessageIdentitySource.stored_objects(db)
+        db.execute(Message.table._sql(
+            f'ALTER TABLE "{Message.table.declared_name}" DROP COLUMN {{report}}'))
+        before = db.total_changes
+        assert BusSchema.reconcile_schema(db, WriterOwnedFile())
+        assert db.total_changes - before > Message.table.count(db)
+        assert not SQLiteSchemaObject.differences(
+            MessageIdentitySource.stored_objects(db),
+            {item.name: item.sql for item in projected_schema},
+        )
+        assert MessageIdentitySource.require(db, seq=record.seq).message_id == record.message_id
+
+        # A changed projection table needs its source even when no source row was written.
+        db.execute(f'ALTER TABLE "{MessageIdentitySource.declared_name}" '
+                   'ADD COLUMN retired_value TEXT')
+        before = db.total_changes
+        assert BusSchema.reconcile_schema(db, WriterOwnedFile())
+        assert db.total_changes > before
+        assert MessageIdentitySource.require(db, seq=record.seq).message_id == record.message_id
+        assert not BusSchema.reconcile_schema(db, WriterOwnedFile())
+
+        # Explicit content repair remains an unconditional rebuild.
+        MessageIdentitySource.update(db, where="{seq}=?", parameters=(record.seq,),
+                                     sender="outdated-author")
+        BusProjectionTables.reset_schema(db)
+        assert MessageIdentitySource.require(db, seq=record.seq).sender == record.sender
+    assert bus.lookup(record.seq) == record
+
+
 @pytest.mark.parametrize("declaration", BusRecord.members_with(BusRecord))
 @pytest.mark.parametrize(
     "role",