diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 67e1c07fa8..f16d520b94 100644 --- a/plans/commander-native-plan.md +++ b/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.** diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 99b27d4bfa..c4a8d9c151 100644 --- a/src/agent_comms/typed_table.py +++ b/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) diff --git a/tests/test_bus.py b/tests/test_bus.py index 6264c2d67e..19b0338a9d 100644 --- a/tests/test_bus.py +++ b/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",