Own notification context and coordinator carry through checked registry privacy

src/agent_comms/bus_tables.py

@@ -26,6 +26,7 @@
 from agent_comms.typed_table import (
     AppendOnlyTables,
     Column,
+    ComposedSchemas,
     DerivedTables,
     DurableTables,
     ForeignKey,
@@ -57,21 +58,9 @@ def create_schema(cls, db: sqlite3.Connection) -> None:
                 header.kind.read_from(db, header.seq).project_into(db)
 
 
-class BusSchema(BusTables, BusProjectionTables):
+class BusSchema(ComposedSchemas, BusTables, BusProjectionTables):
     """One file reconciles its declared families using each family's carry policy."""
 
-    @classmethod
-    def schema_rows(cls) -> tuple[type[TypedTable], ...]:
-        return tuple(dict.fromkeys(
-            row for family in cls.__bases__ for row in family.schema_rows()
-        ))
-
-    @classmethod
-    def carry_schema(cls, db: sqlite3.Connection) -> None:
-        for family in cls.__bases__:
-            family.carry_schema(db)
-
-
 @dataclass(frozen=True)
 class ParticipantSource(TypedTable, BusProjectionTables, declared_name=None):
     lookup: Annotated[str, Column(primary_key=True)]

src/agent_comms/cli_commands.py

@@ -1812,6 +1812,10 @@ def answer(self, root: Path) -> ReconciledStoresResult:
     def apply(self, ctx: Comms) -> ReconciledStoresResult:
         return self.answer(ctx.root)
 
+    def execute(self, route: CommsRoute, output: TextIO) -> int:
+        """Reconciliation reads its declared files before opening runtime consumers."""
+        return self.written(self.answer(route.root), output)
+
 
 @dataclass(frozen=True, kw_only=True)
 class ReconcileStoresCliCommand(StoreReconcileCliCommand, declared_name="reconcile-stores"):
@@ -1830,14 +1834,9 @@ class PlannedReconcileCliCommand(StoreReconcileCliCommand, declared_name="planne
 
     @staticmethod
     def changes(owner: type[ReconciledFile], path: Path) -> bool:
+        schema = owner.schema_at(path)
         with closing(owner.read_snapshot(path)) as db:
-            return owner.table_schema().schema_differs(db)
-
-    def execute(self, route: CommsRoute, output: TextIO) -> int:
-        """Opening the bus opens stores, and opening a store reconciles its schema: before
-        the backup that write would have no copy, so only the stores themselves are read."""
-        return self.written(self.answer(route.root), output)
-
+            return schema.schema_differs(db)
 
 @dataclass(frozen=True, kw_only=True)
 class CheckpointCliCommand(NameArgument, CliCommand, declared_name="checkpoint"):

src/agent_comms/cohort_schema.py

@@ -9,11 +9,9 @@
 from typing import TYPE_CHECKING, Annotated, ClassVar, Literal
 
 from agent_comms.coordination_errors import IdentityConflict, SchemaVersionError
-from agent_comms.coordination_schema import CoordinatorTable
 from agent_comms.coordination_tables.assignments import WakeAssignment
 from agent_comms.coordination_tables.participants import OwnerGenerations, OwnerRecord, Participants
-from agent_comms.coordinator import Coordination
-from agent_comms.private_runtime_schema import PrivateRuntimeSchema
+from agent_comms.private_runtime_schema import PrivateCoordinatorTable
 
 from .audience_manifest import MAX_AUDIENCE_BYTES
 from .coordination_contracts import POLICY_VERSION, RESOLVER_VERSION
@@ -33,7 +31,6 @@
     InitialOneWayColumn,
     NoDelete,
     SharedFileTables,
-    SQLiteJournalMode,
     TypedTable,
 )
 
@@ -44,23 +41,33 @@
 _LOG = logging.getLogger(__name__)
 
 
-class CohortTable(PrivateRuntimeSchema):
+class CohortTable(PrivateCoordinatorTable):
     """Mandatory sealed N/K acknowledgement declarations."""
 
     __slots__ = ()
     schema_error = SchemaVersionError
 
-    @classmethod
-    def install(cls, store) -> None:
-        install_private_cohort_schema(store)
-
 
 class AwarenessTable(SharedFileTables, DurableTables):
     """Optional provenance captured only during the unsealed acceptance transaction."""
 
     __slots__ = ()
     schema_error = SchemaVersionError
 
+    @classmethod
+    def install_optional(cls, db: sqlite3.Connection) -> None:
+        """Optional provenance cannot roll back mandatory native preparation."""
+        db.execute("SAVEPOINT optional_awareness_install")
+        try:
+            if not cls.stored_schema(db):
+                cls.create_schema(db)
+            assert_optional_awareness_schema(db)
+        except (sqlite3.Error, SchemaVersionError, MalformedRecord) as error:
+            db.execute("ROLLBACK TO optional_awareness_install")
+            _LOG.warning("Optional awareness schema unavailable: %r", error, exc_info=error)
+        finally:
+            db.execute("RELEASE optional_awareness_install")
+
 
 @dataclass(frozen=True, kw_only=True)
 class ClaimBatchAcknowledgements(
@@ -308,14 +315,12 @@ class AwarenessClaimGenerations(
     owner_generation: Annotated[int, Column(check="owner_generation > 0")]
 
     def current(self, expected: OwnerGenerations) -> bool:
-        """A historical capture is omitted; a changed current owner refuses."""
+        """Old generations and pre-rename names are historical, never current context."""
         from .wake_candidate_index import ProjectionUnavailableError
 
-        if self.owner_generation < expected.generation:
-            return False
-        if self.owner_identity != expected:
+        if self.owner_lookup != expected.owner_lookup or self.owner_generation > expected.generation:
             raise ProjectionUnavailableError("captured participant owner changed")
-        return True
+        return self.owner_identity == expected
 
     def require_selection(self, assignment: WakeAssignment, acknowledgement: ClaimBatchAcknowledgements) -> None:
         """Verify this immutable acceptance capture against its original claim."""
@@ -366,32 +371,3 @@ def references(cls):
 def assert_optional_awareness_schema(db: sqlite3.Connection) -> None:
     if not AwarenessTable.stored_schema(db) or AwarenessTable.schema_differs(db):
         raise SchemaVersionError("optional awareness generation schema is unavailable")
-
-
-def _install_optional_awareness_schema(db: sqlite3.Connection) -> None:
-    if not AwarenessTable.stored_schema(db):
-        AwarenessTable.create_schema(db)
-    assert_optional_awareness_schema(db)
-
-
-def install_private_cohort_schema(store: Coordination) -> None:
-    """Create current acknowledgement tables on a fresh coordinator; no old-schema repair."""
-    with store.session.writing() as db:
-        CoordinatorTable.require_schema(db)
-        if not CohortTable.stored_schema(db):
-            CohortTable.create_schema(db)
-        CohortTable.require_installed(db)
-        if SQLiteJournalMode.read(db.execute("PRAGMA journal_mode")) != [
-            SQLiteJournalMode("delete")
-        ]:
-            raise SchemaVersionError("cohort schema requires rollback-journal mode")
-        # Missing/corrupt optional metadata must not roll back mandatory cohort
-        # creation, and no prior claim is retroactively assigned a generation.
-        db.execute("SAVEPOINT optional_awareness_install")
-        try:
-            _install_optional_awareness_schema(db)
-        except (sqlite3.Error, SchemaVersionError, MalformedRecord) as error:
-            db.execute("ROLLBACK TO optional_awareness_install")
-            _LOG.warning("Optional awareness schema unavailable: %r", error, exc_info=error)
-        finally:
-            db.execute("RELEASE optional_awareness_install")

src/agent_comms/coordination_database.py

@@ -18,6 +18,7 @@
     CarriedWriter,
     ReconciledFile,
     StoreWriter,
+    TableSchema,
 )
 
 from .declared_family import DeclaredFamily
@@ -72,6 +73,12 @@ class CoordinationStore(CarriedWriter, ReconciledFile):
     def table_schema(cls) -> type[CoordinatorTable]:
         return CoordinatorTable
 
+    @classmethod
+    def schema_at(cls, path: Path) -> type[TableSchema]:
+        from .private_registry_guard import RegistryPrivacy
+
+        return RegistryPrivacy.capture(path.parent).coordinator_schema()
+
     @staticmethod
     def observed(path: Path) -> StoreFile:
         """Classify through SQLite without dropping another connection's process locks.

src/agent_comms/coordination_response.py

@@ -42,7 +42,7 @@
 from agent_comms.native_input_owner import RegistryOwner
 from agent_comms.obligation_states import PublishedResponse, PublishingResponse
 from agent_comms.owner_fence import OwnerFence
-from agent_comms.private_runtime_schema import PrivateRuntimeSchema
+from agent_comms.private_runtime_schema import PrivateCoordinatorTable
 from agent_comms.private_registry_guard import RegistryPrivacy
 from agent_comms.private_send_rules import RunningTurnRegistryCheck
 from agent_comms.publisher import ResponseWrites
@@ -59,15 +59,11 @@
 from .locked_store import RegistryFile
 
 
-class ResponseTable(ImmutableRows, PrivateRuntimeSchema):
+class ResponseTable(ImmutableRows, PrivateCoordinatorTable):
     """An immutable response definition with its write checks and append evidence."""
 
     schema_error = PublicationActivationBlocked
 
-    @classmethod
-    def install(cls, store) -> None:
-        install_private_response_schema(store)
-
 @dataclass(frozen=True)
 class PublicationAppendDispatches(ResponseTable, TypedTable):
     execution_id: Annotated[str, Column(primary_key=True)]
@@ -224,14 +220,6 @@ def require_source(self, initial, assignment: WakeAssignment) -> None:
             raise IdentityConflict("response sealed route differs from committed source")
 
 
-def install_private_response_schema(store: Coordination) -> None:
-    """Install the current response tables on a fresh coordinator only."""
-    with store.session.writing() as db:
-        if not ResponseTable.stored_schema(db):
-            ResponseTable.create_schema(db)
-        ResponseTable.require_installed(db)
-
-
 @dataclass(frozen=True, kw_only=True)
 class LiveResponseOwner(RegistryOwner):
     """One captured registry owner; response finality does not depend on its goal."""

src/agent_comms/coordination_schema.py

@@ -3,10 +3,10 @@
 from __future__ import annotations
 
 from functools import cache
-from importlib import import_module
-from pkgutil import iter_modules
 from typing import Final
 
+from metaclass_registry.discovery import discover_registry_classes
+
 import agent_comms.coordination_tables as coordination_tables
 
 from agent_comms.assignment_states import AssignmentState
@@ -42,8 +42,9 @@ def schema_rows(cls):
 
     @classmethod
     def declarations(cls):
-        for module in iter_modules(coordination_tables.__path__, coordination_tables.__name__ + "."):
-            import_module(module.name)
+        discover_registry_classes(
+            coordination_tables.__path__, coordination_tables.__name__ + ".", CoordinatorTable,
+        )
         return TypedTable.members_with(cls)
 
     @classmethod

src/agent_comms/coordinator.py

@@ -25,7 +25,6 @@
 )
 from .execution_store import ExecutionStore
 from .participant_store import ParticipantStore
-from .private_runtime_schema import PrivateRuntimeSchema
 from .store_files import StoreLock
 from .working_memory_annotations import WorkingMemoryAnnotations
 
@@ -206,19 +205,6 @@ def _bind_tables(self) -> None:
         self.assignments = AssignmentStore(self.session)
         self.annotations = WorkingMemoryAnnotations(self.session)
 
-    def install_private_runtime(self) -> None:
-        """Explicit protocol/owner bootstrap, never invoked by a reader."""
-        # Load the canonical declarations before querying their existing family.
-        from agent_comms import (  # noqa: F401
-            cohort_schema,
-            coordination_response,
-            native_prompt_binding,
-            native_runtime_input,
-        )
-
-        for schema in PrivateRuntimeSchema.__subclasses__():
-            schema.install(self)
-
     def close(self) -> None:
         self.session.close()
 

src/agent_comms/installer/upgrade.py

@@ -689,7 +689,8 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice,
         bridges.require_resumed()
         return result
     except UpgradeRefused as error:
-        report.failed(error)
+        if not rehearse_only:
+            report.failed(error)
         raise
     finally:
         tree.close()

src/agent_comms/native_pi.py

@@ -891,9 +891,9 @@ def read_tracked_input_commitment(
     """
     Id128Text.decode(input_id)
     session_file = Path(session_file).absolute()
-    from .native_entries import NativeInputEvidenceRead
+    from .native_entries import NativeEvidenceRead
 
-    with NativeInputEvidenceRead.borrow(session_file, evidence) as evidence:
+    with NativeEvidenceRead.borrow(session_file, evidence) as evidence:
         evidence.observe()
         try:
             return CommittedInput(evidence.tracked(input_id).message.input_digest.require())

src/agent_comms/native_prompt_binding.py

@@ -18,7 +18,6 @@
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.coordination_tables.assignments import WakeAssignment
 from agent_comms.coordinator import Coordination
-from agent_comms.private_runtime_schema import PrivateRuntimeSchema
 
 from .cohort_schema import CohortTable
 from .field_codec import FieldCodec, Id128Text, MalformedRecord
@@ -34,6 +33,7 @@
 from .threads import Thread
 from .typed_table import (
     Column,
+    DurableTables,
     ImmutableRows,
     IntegerStorage,
     TextStorage,
@@ -52,14 +52,10 @@ def binding_store_path(store: Coordination) -> Path:
 
 @dataclass(frozen=True, slots=True)
 class PromptBinding(
-    ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, TypedTable, PrivateRuntimeSchema
+    ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, TypedTable, DurableTables
 ):
     filename = "native_prompt_bindings.sqlite3"
 
-    @classmethod
-    def install(cls, store) -> None:
-        install_prompt_binding_schema(store)
-
     input_id: Annotated[
         str,
         Column(primary_key=True, check="length(input_id)=32 AND input_id NOT GLOB '*[^0-9a-f]*'"),
@@ -99,11 +95,6 @@ class _BindingRoot(TypedRow):
     wire_root_id: str
 
 
-def install_prompt_binding_schema(store: Coordination) -> None:
-    """Explicit fresh-root install; ordinary callers may rely on _ensure."""
-    _ensure_binding_schema(store)
-
-
 def _ensure_binding_schema(store: Coordination) -> None:
     """Serialized snapshot installation; never repair an uncertain commit."""
     create_sidecar_file(binding_store_path(store), PromptBinding)