Derive table mutation guards from capabilities

src/agent_comms/attempt_recovery.py

@@ -80,14 +80,14 @@ def observe_native_release(
         is insufficient.
         """
         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

@@ -59,7 +59,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")

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 CohortDeliveryReceipts, CohortTable
 from .coordination_cohort import accept_delivery_cohort
 from .errors import RelationViolationError
 from .message_bus import MessageBus
@@ -39,7 +39,7 @@ 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)
+                CohortTable.require_installed(resource.session._connection)
                 sealed = CohortDeliveryReceipts.sealed_sequences_after(
                     resource.session._connection,
                     wire_root_id=root_id,

src/agent_comms/coordinated_runtime_schema.py

@@ -2,27 +2,15 @@
 
 from __future__ import annotations
 
-import sqlite3
 
-from agent_comms.coordination_errors import PublicationActivationBlocked
 from agent_comms.coordinator import Coordination
 
 from .native_runtime_input import NativeRuntimeTable
-from .typed_table import SQLiteForeignKeys
-
-
-def assert_native_runtime_schema(db: sqlite3.Connection) -> None:
-    if not NativeRuntimeTable.stored_schema(db):
-        raise PublicationActivationBlocked("native runtime schema is not installed")
-    if NativeRuntimeTable.schema_differs(db) or SQLiteForeignKeys.read(
-        db.execute("PRAGMA foreign_keys")
-    ) != [SQLiteForeignKeys(True)]:
-        raise PublicationActivationBlocked("native runtime schema has drifted")
 
 
 def install_native_runtime_schema(store: Coordination) -> None:
     """Explicit fresh schema install. Existing state must match the declared schema."""
     with store.session.writing() as db:
         if not NativeRuntimeTable.stored_schema(db):
             NativeRuntimeTable.create_schema(db)
-        assert_native_runtime_schema(db)
+        NativeRuntimeTable.require_installed(db)

src/agent_comms/coordination_cohort.py

@@ -15,14 +15,7 @@
 
 from agent_comms.assignment_states import AssignmentState, PendingNotification
 from agent_comms.bus_publication import CommittedDelivery
-from agent_comms.cohort_schema import (
-    AwarenessClaimGenerations,
-    ClaimBatchAcknowledgements,
-    ClaimBatchMembers,
-    CohortDeliveryReceipts,
-    assert_cohort_schema,
-    assert_optional_awareness_schema,
-)
+from agent_comms.cohort_schema import AwarenessClaimGenerations, ClaimBatchAcknowledgements, ClaimBatchMembers, CohortDeliveryReceipts, CohortTable, assert_optional_awareness_schema
 from agent_comms.coordination_contracts import POLICY_VERSION, RESOLVER_VERSION
 from agent_comms.coordination_errors import (
     IdentityConflict,
@@ -214,8 +207,8 @@ def _record_optional_owner_generations(
                 claim_id=assignment.assignment_id,
                 wire_root_id=wire_root_id,
                 wire_seq=wire_seq,
-                recipient_lookup=assignment.recipient_lookup,
-                canonical_thread=assignment.recipient,
+                owner_lookup=assignment.recipient_lookup,
+                owner_thread=assignment.recipient,
                 owner_generation=current.generation,
             ).insert(db)
     except (sqlite3.Error, SchemaVersionError, IdentityConflict, MalformedRecord) as error:
@@ -245,7 +238,7 @@ def accept_delivery_cohort(
         if wire_seq <= marker.claimed_after_seq:
             raise IdentityConflict("historical source precedes the current claim_scope floor")
         initial = source.delivery(wire_seq)
-        assert_cohort_schema(db)
+        CohortTable.require_installed(db)
         acknowledgement = ClaimBatchAcknowledgements.one(db, wire_root_id=wire_root_id, wire_seq=wire_seq)
         if acknowledgement is not None:
             return AlreadyApplied(_claim_record(db, initial))
@@ -360,7 +353,7 @@ def sealed_cohort_assignments(
         raise ValueError("acknowledgement-backed page bounds are invalid")
     with store.session.read():
         db = store.session._connection
-        assert_cohort_schema(db)
+        CohortTable.require_installed(db)
         selected_states = AssignmentState.selection_predicate(state_capability, "c")
         return tuple(
             WakeAssignment.read(

src/agent_comms/coordination_response.py

@@ -15,14 +15,13 @@
 
 from __future__ import annotations
 
-import sqlite3
 import time
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
 from typing import Annotated, Type
 
-from agent_comms.cohort_schema import assert_cohort_schema
+from agent_comms.cohort_schema import CohortTable
 from agent_comms.coordination_errors import (
     IdentityConflict,
     PublicationActivationBlocked,
@@ -52,8 +51,8 @@
     Column,
     DurableTables,
     ForeignKey,
+    ImmutableRows,
     SharedFileTables,
-    SQLiteForeignKeys,
     SqlValue,
     TypedRow,
     TypedTable,
@@ -62,23 +61,15 @@
 from .locked_store import RegistryFile
 
 
-class ResponseTable(SharedFileTables, DurableTables, PrivateRuntimeSchema):
+class ResponseTable(ImmutableRows, SharedFileTables, DurableTables, PrivateRuntimeSchema):
     """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)
 
-    @classmethod
-    def trigger_bodies(cls) -> dict[str, str]:
-        return {
-            f"{cls.declared_name}_{operation.lower()}_guard": " "
-            f"BEFORE {operation} ON {cls.declared_name} "
-            "BEGIN SELECT RAISE(ABORT,'response evidence is frozen'); END"
-            for operation in ("UPDATE", "DELETE")
-        }
-
-
 @dataclass(frozen=True)
 class PublicationAppendDispatches(ResponseTable, TypedTable):
     execution_id: Annotated[str, Column(primary_key=True)]
@@ -235,21 +226,12 @@ def require_source(self, initial, assignment: WakeAssignment) -> None:
             raise IdentityConflict("response sealed route differs from committed source")
 
 
-def _assert_response_schema(db: sqlite3.Connection) -> None:
-    if not ResponseTable.stored_schema(db):
-        raise PublicationActivationBlocked("private response schema is not installed")
-    if ResponseTable.schema_differs(db) or SQLiteForeignKeys.read(
-        db.execute("PRAGMA foreign_keys")
-    ) != [SQLiteForeignKeys(True)]:
-        raise PublicationActivationBlocked("private response schema has drifted")
-
-
 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)
-        _assert_response_schema(db)
+        ResponseTable.require_installed(db)
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -351,8 +333,8 @@ def _require_cohort_assignments(
     Caller already holds the bus file lock before the SQL transaction.
     """
     db = store.session._connection
-    _assert_response_schema(db)
-    assert_cohort_schema(db)
+    ResponseTable.require_installed(db)
+    CohortTable.require_installed(db)
     snapshot.require_wire_responses()
     metadata = bus.log.private_marker_unlocked()
     if metadata.root_id != wire_root_id:
@@ -555,7 +537,7 @@ def _settle_fenced_response(
             # a first attempt from a lost response/crash. If it commits but the
             # process dies before appending, no automatic retry is authorized.
             with store.session.writing() as db:
-                _assert_response_schema(db)
+                ResponseTable.require_installed(db)
                 terminal = _terminal_replay(store, fence, bus, wire_root_id, owner_witness, exact_target)
                 if terminal is not None:
                     return terminal
@@ -575,7 +557,7 @@ def _settle_fenced_response(
                 elif not prior.matches(wire_root_id, fence, exact_target):
                     raise IdentityConflict("response dispatch belongs to another root or owner")
         with store.session.writing() as db:
-            _assert_response_schema(db)
+            ResponseTable.require_installed(db)
             terminal = _terminal_replay(store, fence, bus, wire_root_id, owner_witness, exact_target)
             if terminal is not None:
                 return terminal

src/agent_comms/coordination_schema.py

@@ -66,18 +66,6 @@ def _sql_members(family, predicate=lambda member: True):
     )
 
 
-def _sql_edges(family, column):
-    return (
-        " OR ".join(
-            f"({column.format(row='OLD')} = {_sql_values((name,))} "
-            f"AND {column.format(row='NEW')} IN ({_sql_values(sorted(edges))}))"
-            for name, edges in family.transition_table().items()
-            if edges
-        )
-        or "0"
-    )
-
-
 _CONTEXT_FAMILIES: Final = (
     ExecutionState, AttemptState, AssignmentState, WakePolicy, ResponseState, RecoveryCondition,
 )
@@ -93,10 +81,6 @@ def _schema_context(*members):
         wake_names=_sql_members(WakePolicy),
         response_names=_sql_members(ResponseState),
         recovery_names=_sql_members(RecoveryCondition),
-        execution_edges=_sql_edges(ExecutionState, "json_extract({row}.lifecycle, '$.kind')"),
-        attempt_edges=_sql_edges(AttemptState, "json_extract({row}.lifecycle, '$.kind')"),
-        assignment_edges=_sql_edges(AssignmentState, "json_extract({row}.lifecycle, '$.kind')"),
-        obligation_edges=_sql_edges(ResponseState, "json_extract({row}.lifecycle, '$.kind')"),
         terminal_attempt_names=_sql_members(AttemptState, lambda member: member.terminal),
         failed_attempt_phases=_sql_members(AttemptState, lambda member: member.failed),
         retry_attempt_phases=_sql_values(

src/agent_comms/coordination_tables/__init__.py

@@ -2,3 +2,23 @@
 
 Importing one row must not import unrelated annotation and UI consumers.
 """
+
+from agent_comms.typed_table import ForeignKey, GuardedAfterWrites
+
+
+class ExecutionReference:
+    """An execution-owned row cannot outlive its original execution."""
+
+    @classmethod
+    def references(cls) -> tuple[ForeignKey, ...]:
+        from .executions import ExecutionRecord
+
+        return (ForeignKey(("execution_id",), ExecutionRecord, ("execution_id",), on_delete="RESTRICT"),)
+
+
+class RetryPartition(GuardedAfterWrites):
+    """A failed execution cannot retain authorization to retry."""
+
+    after_write_condition = """EXISTS (SELECT 1 FROM retry_disposition_basis b JOIN executions e ON e.execution_id = b.execution_id WHERE
+        b.execution_id = NEW.execution_id AND b.authorized = 1 AND e.status = 'failed' AND e.current_attempt_ordinal
+        IS NOT NULL)"""