Share table capabilities and migrate claim fields in retained stores

src/agent_comms/acp_extension.py

@@ -201,7 +201,7 @@ def require_live(self):
 class LiveAttachmentScope(AttachmentScope):
     """Logical ACP attachment and executor evidence have separate identities."""
     session_id: str
-    claim_scope: ClaimedIdentity
+    claimed_identity: ClaimedIdentity
     owner: ProcessIdentity
 
     @property
@@ -210,11 +210,11 @@ def logical_key(self) -> tuple[str, ...]: ...
 
     @property
     def owner_created_at(self):
-        return self.claim_scope.incarnation.require_recorded().created_at
+        return self.claimed_identity.incarnation.require_recorded().created_at
 
     @property
     def claimed_generation(self):
-        return self.claim_scope.claimed_generation
+        return self.claimed_identity.claimed_generation
 
     def to_wire(self):
         return FieldCodec.encode_record(self)
@@ -302,15 +302,15 @@ class NoCursorScope(CursorScope):
 class LiveCursorScope(CursorScope, LiveAttachmentScope):
     session_id: str
     wire_root_id: str
-    claim_scope: Annotated[ClaimedIdentity, WireName("admission")]
+    claimed_identity: ClaimedIdentity
     owner: ProcessIdentity
 
     def __post_init__(self):
         Id128Text.decode(self.wire_root_id)
-        self.claim_scope.incarnation.require_recorded()
+        self.claimed_identity.incarnation.require_recorded()
         if (
             not self.session_id
-            or self.claim_scope.claimed_generation <= 0
+            or self.claimed_identity.claimed_generation <= 0
         ):
             raise ValueError("Invalid native cursor scope")
 
@@ -370,7 +370,7 @@ def validate_scope(self, scope: CursorScope) -> None:
         cursor = self.cursor
         if (
             cursor.wire_root_id != scope.wire_root_id
-            or cursor.owner_thread != scope.claim_scope.incarnation.name
+            or cursor.owner_thread != scope.claimed_identity.incarnation.name
             or cursor.owner_claimed_generation != scope.claimed_generation
             or not 0 <= cursor.injected_seq <= cursor.covered_seq
         ):
@@ -446,7 +446,7 @@ class NoQueueScope(QueueScope):
 @dataclass(frozen=True)
 class LiveQueueScope(QueueScope, LiveAttachmentScope):
     session_id: str
-    claim_scope: Annotated[ClaimedIdentity, WireName("admission")]
+    claimed_identity: ClaimedIdentity
     owner: ProcessIdentity
 
     @property
@@ -549,7 +549,7 @@ def without_scope(self) -> None:
         raise ValueError('Available input queue has no original attachment scope.')
 
     def capture_human_input(self, comms, acquire_scope: Callable[[], QueueScope]) -> HumanInputOrigin:
-        return HumanInputOrigin.capture(comms, acquire_scope().claim_scope)
+        return HumanInputOrigin.capture(comms, acquire_scope().claimed_identity)
 
 
 @dataclass(frozen=True)

src/agent_comms/channel_coding_tools.py

@@ -100,7 +100,7 @@ def for_original(self, assignment, operation_id):
 @dataclass(frozen=True, slots=True)
 class SelectedCodingToolOwner(CodingToolOwner):
     store: Coordination
-    claim_scope: WakeClaim
+    resource_scope: WakeClaim
     session_dir: Path
     input_id: str
     goal_permission: TurnGoalPermission = field()
@@ -115,8 +115,8 @@ def for_original(self, assignment, operation_id) -> CodingToolOwner:
                 self.store.session._connection, self.input_id
             ):
                 raise SelectedToolDenied("Tool original is absent from the reserved native batch")
-        return replace(self, claim_scope=replace(
-            self.claim_scope,
+        return replace(self, resource_scope=replace(
+            self.resource_scope,
             source_seq=assignment.wire_seq,
             source_message_id=assignment.message_id,
             wake_assignment_id=assignment.assignment_id,
@@ -127,11 +127,11 @@ def for_original(self, assignment, operation_id) -> CodingToolOwner:
     def verify(self, store, turn, snapshot):
         if self.input_id != turn.command.input_id:
             raise SelectedToolDenied("Native tool input changed")
-        verify_sent_full_input(store, self.claim_scope, self.owner_name, self.input_id)
-        verify_selected_wake(self.comms, store, self.claim_scope, self.owner_name)
+        verify_sent_full_input(store, self.resource_scope, self.owner_name, self.input_id)
+        verify_selected_wake(self.comms, store, self.resource_scope, self.owner_name)
 
     def claim_resource(self, store, turn, resource, slot):
-        scope = replace(self.claim_scope, operation_id=slot)
+        scope = replace(self.resource_scope, operation_id=slot)
         return publish_selected_resource_claim(self.comms, store, scope, self.owner_name, resource)
 
 

src/agent_comms/cli_protocol.py

@@ -20,6 +20,7 @@
 from .field_codec import TimestampText
 from .recorded_time import UnrecordedTimeError
 from .external_transcript import RequestedEarlierPage
+from .turn_item import TurnItem
 
 if TYPE_CHECKING:
     from .child_process import AttachedChild
@@ -222,11 +223,9 @@ async def handle(self, bridge: CliBridge, child: AttachedChild) -> None:
         await bridge.accept(self.message)
 
 
-class EarlierItems(WatchLine):
+class EarlierItems(WatchLine, TurnItem):
     """The pager's outstanding continuation, handled only by its socket-owning bridge."""
     wire_tag = 'earlier-items'
-    turn_id: str
-    item_id: str
     request: RequestedEarlierPage
 
     @classmethod

src/agent_comms/cohort_schema.py

@@ -32,6 +32,7 @@
     NoDelete,
     SharedFileTables,
     TypedTable,
+    WithoutRowid,
 )
 
 if TYPE_CHECKING:
@@ -75,7 +76,7 @@ class ClaimBatchAcknowledgements(
     GuardedColumns,
     NoDelete,
     CohortTable,
-    TypedTable,
+    WithoutRowid,
     declared_name="claim_batch_receipts",
 ):
     transition_column = "sealed"
@@ -135,7 +136,6 @@ class ClaimBatchAcknowledgements(
     policy_version: Annotated[str, FrozenColumn(check="length(policy_version) BETWEEN 1 AND 256")]
     accepted_at_ms: Annotated[int, FrozenColumn(check="accepted_at_ms >= 0")]
     sealed: bool = False
-    without_rowid = True
 
     @classmethod
     def decision_hash(cls, delivery: Delivered) -> str:
@@ -177,7 +177,7 @@ def require_selected_claim(self, assignment: WakeAssignment) -> None:
             raise ProjectionUnavailableError("selected source acknowledgement is unsealed or changed")
 
 @dataclass(frozen=True, kw_only=True)
-class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, TypedTable):
+class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, WithoutRowid):
 
     insert_condition = """((SELECT sealed FROM claim_batch_receipts r
         WHERE r.wire_root_id = NEW.wire_root_id AND r.wire_seq = NEW.wire_seq) != 0)
@@ -198,7 +198,6 @@ class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, TypedTable):
         ("claim_id",),
         ("wire_root_id", "wire_seq", "recipient_lookup"),
     )
-    without_rowid = True
 
     @classmethod
     def references(cls):
@@ -216,7 +215,7 @@ class CohortDeliveries(
     GuardedInserts,
     ImmutableRows,
     CohortTable,
-    TypedTable,
+    WithoutRowid,
     declared_name="cohort_delivery_receipts",
 ):
     """Sealed deliveries, including observers that receive no execution claim."""
@@ -255,7 +254,6 @@ class CohortDeliveries(
         ("wire_root_id", "wire_seq", "claim_id"),
     )
     indexes = (Index(("recipient_lookup", "wire_seq")),)
-    without_rowid = True
 
     @classmethod
     def sealed_sequences_after(

src/agent_comms/context_store.py

@@ -11,7 +11,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Annotated, ClassVar, Literal, Self, cast
+from typing import Annotated, ClassVar, Self, cast
 
 from .errors import RelationViolationError, StoredHistoryUnreadable
 from .field_codec import (
@@ -35,7 +35,7 @@
     UnrecordedSegment,
 )
 
-from .typed_table import Column, DurableStore, Generated, Index, TypedTable
+from .typed_table import Column, DurableStore, Generated, Index, SingletonRow, TypedTable
 
 
 @dataclass(frozen=True)
@@ -91,8 +91,7 @@ def page(
 
 
 @dataclass(frozen=True)
-class ContextStoreInfo(ContextSourceState, TypedTable):
-    singleton: Annotated[Literal[1], Column(primary_key=True)]
+class ContextStoreInfo(ContextSourceState, SingletonRow):
     source: ContextStoreIdentity
 
     @classmethod

src/agent_comms/coordination_response.py

@@ -53,7 +53,7 @@
     ImmutableRows,
     SqlValue,
     TypedRow,
-    TypedTable,
+    WithoutRowid,
 )
 
 from .locked_store import RegistryFile
@@ -65,7 +65,7 @@ class ResponseTable(ImmutableRows, PrivateCoordinatorTable):
     schema_error = PublicationActivationBlocked
 
 @dataclass(frozen=True)
-class PublicationAppendDispatches(ResponseTable, TypedTable):
+class PublicationAppendDispatches(ResponseTable, WithoutRowid):
     execution_id: Annotated[str, Column(primary_key=True)]
     exact_target: Annotated[
         str, Column(primary_key=True, check="length(exact_target) BETWEEN 1 AND 256")
@@ -76,7 +76,6 @@ class PublicationAppendDispatches(ResponseTable, TypedTable):
     owner_generation: Annotated[int, Column(check="owner_generation>0")]
     attempt_ordinal: Annotated[int, Column(check="attempt_ordinal>0")]
     dispatched_at_ms: Annotated[int, Column(check="dispatched_at_ms>=0")]
-    without_rowid = True
 
     @classmethod
     def references(cls):

src/agent_comms/coordination_snapshot.py

@@ -355,7 +355,7 @@ def release_resources(self, boundary: NativeBoundary, outcome: TerminalAttempt)
         current = bus.bus.claim_projection().live_in(boundary.registry)
         owned = tuple(
             claim for claim in current.values()
-            if claim.claim_scope.belongs_to(attempt)
+            if claim.resource_scope.belongs_to(attempt)
         )
         if owned:
             bus.publisher.release_resources(boundary, owned, outcome.release_wording)

src/agent_comms/envelope_claim_transitions.py

@@ -169,9 +169,7 @@ def held_at(self, claimed_generation: int) -> bool:
 class RecordedScope(ClaimScope, declared_name=None):
     """Every acquisition published today records the input generation and operation it serves."""
 
-    owner_claimed_generation: Annotated[
-        int, WireName("owner_admission_generation")
-    ]
+    owner_claimed_generation: int
     operation_id: Annotated[str, Id128Text]
     turn_id: str
 
@@ -274,7 +272,7 @@ class ClaimIdentity:
 @dataclass(frozen=True)
 class AcquiringClaim(ClaimIdentity):
     generation: Annotated[str, Id128Text]
-    claim_scope: Annotated[ClaimScope, WireName("admission")]
+    resource_scope: ClaimScope
 
 
 @dataclass(frozen=True)
@@ -351,7 +349,7 @@ class Acquisition(ClaimTransition, declared_name=None):
         default=(), kw_only=True
     )
     generation: Annotated[str, Id128Text] = field(kw_only=True)
-    claim_scope: ClassVar[ClaimScope]
+    resource_scope: ClassVar[ClaimScope]
 
     def __post_init__(self) -> None:
         super().__post_init__()
@@ -372,7 +370,7 @@ def claimed(self, resource: str) -> ClaimOwner:
 
     @property
     def operation_ids(self) -> tuple[str, ...]:
-        return self.claim_scope.operation_ids
+        return self.resource_scope.operation_ids
 
     def require_holders(self, previous: ClaimProjection) -> None:
         for resource in self.claims:
@@ -390,7 +388,7 @@ def project(self, previous: ClaimProjection) -> ClaimProjection:
 class WakeClaims(RecordMember, Acquisition, AcquiringClaim):
     """An acquisition published today; its row records the scope it serves."""
 
-    claim_scope: Annotated[RecordedScope, WireName("admission")]
+    resource_scope: RecordedScope
 
 
 @dataclass(frozen=True)
@@ -400,7 +398,7 @@ class UnscopedClaims(RecordMember, Acquisition):
     Only bus history holds this row; no writer publishes one.
     """
 
-    claim_scope: ClassVar[ClaimScope] = UnrecordedScope()
+    resource_scope: ClassVar[ClaimScope] = UnrecordedScope()
 
 
 @dataclass(frozen=True)
@@ -447,7 +445,7 @@ def __post_init__(self) -> None:
     def acquire(
         self, scope: RecordedScope, registry: RegistrySnapshot, publish: Callable[[], ClaimOwner]
     ) -> ClaimOwner:
-        return self if self.claim_scope.reuses(scope) and self.live(registry) else publish()
+        return self if self.resource_scope.reuses(scope) and self.live(registry) else publish()
 
     def require_holder(self, owner: str, incarnation: str) -> None:
         if (self.owner, self.incarnation) != (owner, incarnation):
@@ -457,7 +455,7 @@ def live(self, registry: RegistrySnapshot) -> bool:
         incarnation = RecordedThreadIncarnation(self.owner, float(self.incarnation))
         return any(
             thread.process_alive
-            and self.claim_scope.held_at(registry.claim_generation(thread.name))
+            and self.resource_scope.held_at(registry.claim_generation(thread.name))
             for thread in registry.active_threads(registry.threads_for(incarnation))
         )