Finish table mutation rule ownership (supersedes 2471)

src/agent_comms/attempt_recovery.py

@@ -28,7 +28,8 @@
 )
 from agent_comms.coordination_results import Applied
 from agent_comms.coordination_snapshot import RecoverySnapshot
-from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayAssessments, ReplayFact
+from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayFact
+from agent_comms.coordination_tables.executions import ReplayAssessments
 from agent_comms.coordination_tables.executions import CurrentExecutions
 from agent_comms.coordination_tables.recovery import (
     RecoveryAudit,
@@ -79,14 +80,14 @@ def observe_native_release(
         input still settles UNKNOWN, never unsent or safe to replay.
         """
         from agent_comms.comms import Comms
-        from agent_comms.coordinated_runtime_schema import assert_native_runtime_schema
+        from agent_comms.native_runtime_input import NativeRuntimeTable
 
         if store.session.path.name != CoordinationStore.filename:
             raise RecoveryBlocked("native recovery requires the canonical coordination store")
         comms = Comms(store.session.path.parent)
         with store.session.boundary(comms.bus) as boundary:
             registry = boundary.registry
-            assert_native_runtime_schema(store.session._connection)
+            NativeRuntimeTable.require_installed(store.session._connection)
             snapshot = store.snapshots.get(execution_id)
             attempt = snapshot.attempt
             if attempt is None or not snapshot.is_current:

src/agent_comms/attempt_start.py

@@ -69,8 +69,8 @@ def validate(self, snapshot: RecoverySnapshot, participant: ParticipantSnapshot)
         execution = snapshot.execution
         ordinal = execution.lifecycle.next_attempt_ordinal
         if not participant.committed or (
-            participant.owner_thread,
-            participant.participant_generation,
+            participant.owner_identity.owner_thread,
+            participant.owner_identity.generation,
         ) != (self.owner_thread, self.owner_generation):
             raise StaleFence("owner generation or thread changed")
         if (execution.revision, participant.pointer.pointer_revision) != (

src/agent_comms/attempt_store.py

@@ -18,7 +18,7 @@
 from agent_comms.coordination_session import CoordinationSession, NativeBoundary
 from agent_comms.coordination_snapshot import RecoverySnapshot
 from agent_comms.coordination_tables.assignments import WakeAssignment
-from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayAssessments, ReplayFact
+from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayFact
 from agent_comms.coordination_tables.recovery import (
     ACPClientConnectivity,
     ConnectivityFacet,
@@ -57,7 +57,7 @@ def require_fence(self, fence: OwnerFence) -> tuple[RecoverySnapshot, AttemptRec
         attempt = snapshot.require_current_attempt()
         if (
             attempt.authority != fence.authority
-            or self.participants.get(attempt.owner_lookup).participant_generation
+            or self.participants.get(attempt.owner_lookup).owner_identity.generation
             != attempt.owner_generation
         ):
             raise StaleFence("attempt fence is not current")
@@ -179,18 +179,13 @@ def observe_replay(
             if snapshot.pointer_revision != expected_pointer_revision:
                 raise StaleRevision("pointer revision changed")
             before = snapshot.replay
-            if (before.revision if before else None) != expected_replay_revision:
-                raise StaleRevision("replay assessment revision changed")
-            after = ReplayAssessments(
-                fence.execution_id,
-                facts,
-                replay_safe,
-                side_effects_possible,
-                1 if before is None else before.revision + 1,
-            )
-            if not after.record(self.session._connection, snapshot.replay):
+            before.require_revision(expected_replay_revision)
+            if not before.observe(
+                self.session._connection, fence.execution_id, facts=facts,
+                replay_safe=replay_safe, side_effects_possible=side_effects_possible,
+            ):
                 return AlreadyApplied(snapshot)
-            return Applied(self.snapshots.get(after.execution_id))
+            return Applied(self.snapshots.get(fence.execution_id))
 
     def fail_unknown(
         self,

src/agent_comms/cohort_foreground.py

@@ -8,7 +8,7 @@
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.coordinator import Coordination
 
-from .cohort_schema import CohortDeliveryReceipts, assert_cohort_schema
+from .cohort_schema import CohortDeliveries, CohortTable
 from .coordination_cohort import accept_delivery_cohort
 from .errors import RelationViolationError
 from .message_bus import MessageBus
@@ -39,8 +39,8 @@ def accept(resource: Coordination) -> int:
                 raise IdentityConflict("private initial wire root changed")
             authorized_after = max(after_seq, marker.claimed_after_seq)
             with resource.session.read():
-                assert_cohort_schema(resource.session._connection)
-                sealed = CohortDeliveryReceipts.sealed_sequences_after(
+                CohortTable.require_installed(resource.session._connection)
+                sealed = CohortDeliveries.sealed_sequences_after(
                     resource.session._connection,
                     wire_root_id=root_id,
                     recipient_lookup=lookup,