Integration batch two: owned dispatch, closure state, field directives, retained waits, install speed, import tax

src/agent_comms/acp_extension.py

@@ -6,10 +6,8 @@
 
 from __future__ import annotations
 
-import json
 from abc import abstractmethod
 from dataclasses import dataclass, field, replace
-from hashlib import sha256
 from typing import TYPE_CHECKING, Annotated, Callable, ClassVar
 
 
@@ -201,7 +199,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 +208,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 +300,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 +368,7 @@ def validate_scope(self, scope: CursorScope) -> None:
         cursor = self.cursor
         if (
             cursor.wire_root_id != scope.wire_root_id
-            or cursor.recipient_lookup != scope.claim_scope.incarnation.lookup
+            or cursor.recipient_lookup != scope.claimed_identity.incarnation.lookup
             or cursor.owner_claimed_generation != scope.claimed_generation
             or not 0 <= cursor.injected_seq <= cursor.covered_seq
         ):
@@ -398,9 +396,7 @@ def same_observation(self, other: CursorEnvelope) -> bool:
 
     @property
     def digest(self) -> str:
-        return sha256(
-            json.dumps(FieldCodec.encode(self), sort_keys=True, separators=(",", ":")).encode()
-        ).hexdigest()
+        return FieldCodec.digest(self)
 
 
 @dataclass(frozen=True)
@@ -446,7 +442,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 +545,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)
@@ -570,9 +566,7 @@ def status(self) -> str | None:
 
     @property
     def digest(self) -> str:
-        return sha256(
-            json.dumps(FieldCodec.encode(self), sort_keys=True, separators=(",", ":")).encode()
-        ).hexdigest()
+        return FieldCodec.digest(self)
 
 
 @dataclass(frozen=True)

src/agent_comms/agent_backend.py

@@ -22,7 +22,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, ClassVar, Protocol, Self
+from typing import TYPE_CHECKING, Any, ClassVar, Self
 
 from .declared_family import DeclaredFamily
 from .pi_counts import PiCount
@@ -157,23 +157,25 @@ class InputContent:
     contributions: tuple[InputContributionCoordinates, ...] = ()
 
 
-class TurnInboxReceiver(Protocol):
-    """What a running turn does with each record its inbox carries."""
+class TurnInboxItem(ABC):
+    """An inbox item carries its input identities; the forwarding owner dispatches it."""
+
+    @property
+    def carried_inputs(self) -> tuple[InputId, ...]:
+        return ()
 
-    def place(self, turn: object, item: InputRequest) -> Awaitable[bool]: ...
-    def send_now(self, turn: object, item: SendNow) -> Awaitable[bool]: ...
-    def use_settings(self, turn: object, item: UseSettings) -> Awaitable[bool]: ...
-    def clear_queue(self, turn: object, item: ClearQueue) -> Awaitable[bool]: ...
+    def refuse(self, inputs, turn):
+        inputs.refuse(turn, self.carried_inputs)
 
 
-class TurnInbox(asyncio.Queue):
+class TurnInbox(asyncio.Queue[TurnInboxItem]):
     """The live inbox parks its head until the existing observer sees READY."""
     def __init__(self):
         super().__init__()
         self.ready = asyncio.Event()
         self.ready.set()
 
-    async def get(self):
+    async def get(self) -> TurnInboxItem:
         await self.ready.wait()
         return await super().get()
 
@@ -186,30 +188,16 @@ def has_ready_work(self) -> bool:
     def all_written_started(self):
         return not any(item.carried_inputs for item in self._queue)
 
-    def begin(self, item):
+    def begin(self, item: TurnInboxItem) -> None:
         self._queue.appendleft(item)
         self._wakeup_next(self._getters)
 
-    def hold(self, item):
+    def hold(self, item: TurnInboxItem) -> None:
         # Park under the shared wire lock, before READY can be published.
         self.ready.clear()
         self.begin(item)
 
 
-class TurnInboxItem(ABC):
-    """An inbox item owns its carried inputs and delivery to a running turn."""
-
-    @property
-    def carried_inputs(self) -> tuple[InputId, ...]:
-        return ()
-
-    def refuse(self, inputs, turn):
-        inputs.refuse(turn, self.carried_inputs)
-
-    @abstractmethod
-    def deliver_to(self, receiver: TurnInboxReceiver, turn: object) -> Awaitable[bool]: ...
-
-
 @dataclass(frozen=True, kw_only=True)
 class InputRequest(TurnInboxItem):
     """One queued user input handed to a backend: identity, content and busy policy.
@@ -225,9 +213,6 @@ class InputRequest(TurnInboxItem):
     def carried_inputs(self) -> tuple[InputId, ...]:
         return (self.input_id,)
 
-    def deliver_to(self, receiver: TurnInboxReceiver, turn: object) -> Awaitable[bool]:
-        """Hand this input to the running turn's receiver; False ends forwarding."""
-        return receiver.place(turn, self)
 
 
 @dataclass(frozen=True)
@@ -239,8 +224,6 @@ class SendNow(TurnInboxItem):
 
     input_ids: tuple[InputId, ...]
 
-    def deliver_to(self, receiver: TurnInboxReceiver, turn: object) -> Awaitable[bool]:
-        return receiver.send_now(turn, self)
 
 
 @dataclass(frozen=True)
@@ -254,8 +237,6 @@ class UseSettings(TurnInboxItem):
     model: ThreadModel
     thinking_level: ThinkingLevelField
 
-    def deliver_to(self, receiver: TurnInboxReceiver, turn: object) -> Awaitable[bool]:
-        return receiver.use_settings(turn, self)
 
 
 @dataclass(frozen=True)
@@ -264,8 +245,6 @@ class ClearQueue(TurnInboxItem):
 
     input_ids: tuple[InputId, ...]
 
-    def deliver_to(self, receiver: TurnInboxReceiver, turn: object) -> Awaitable[bool]:
-        return receiver.clear_queue(turn, self)
 
 
 class CompactionOutcome(DeclaredFamily, DeclaredReply, affix="Compaction"):

src/agent_comms/assignment_states.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
-import json
 from abc import abstractmethod
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, ClassVar
 
 from .coordination_errors import IdentityConflict, IntegrityViolationError
 from .declared_family import DeclaredFamily
+from .field_codec import FieldCodec
 from .lifecycle import LifecycleState
 from .lifecycle_capability import Active, Complete, Deferred, Failed, Pending, Terminal
 from .notification_progress import (
@@ -363,14 +363,12 @@ def wake_frame(self, source, obligation) -> tuple[str, str]:
             source.exact_reply_target,
         ):
             raise IdentityConflict("full wake frame requires the current response obligation")
-        return self.mode.full_expectation(), "you owe a response: " + json.dumps(
+        return self.mode.full_expectation(), "you owe a response: " + FieldCodec.compact(
             {
                 "target": obligation.exact_target,
                 "source_seq": source.seq,
                 "execution_id": obligation.execution_id,
-            },
-            ensure_ascii=True,
-            separators=(",", ":"),
+            }, ensure_ascii=True,
         )
 
     def require_engagement(self) -> Engagement:

src/agent_comms/audience_manifest.py

@@ -9,7 +9,6 @@
 from __future__ import annotations
 
 import hashlib
-import json
 import math
 import re
 from collections.abc import Sequence
@@ -19,7 +18,7 @@
 from agent_comms.coordination_contracts import MAX_IDENTIFIER_CHARS
 
 from .channel_targets import exact_stored_target
-from .field_codec import MalformedRecord, Sha256Text
+from .field_codec import FieldCodec, MalformedRecord, Sha256Text
 from .messages import Message
 
 AUDIENCE_VERSION: Final = 1
@@ -43,8 +42,8 @@ def _bounded_text(value: str, field: str, *, thread_name: bool = False) -> None:
 
 def _canonical_json(value: object, *, limit: int) -> bytes:
     try:
-        encoded = json.dumps(
-            value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
+        encoded = FieldCodec.compact(
+            value, sort_keys=True, ensure_ascii=False, allow_nan=False,
         ).encode("utf-8")
     except ValueError as error:  # NaN, or a lone surrogate the UTF-8 encoding refuses
         raise MalformedRecord("audience data cannot be canonically encoded") from error

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_bridges/shared.py

@@ -799,7 +799,7 @@ async def prepare_delivery(self) -> None:
     async def accept(self, message: messages.Message) -> None:
         conversation = message.type.conversation_message
         outcome = UnconfirmedOutcome if conversation else FilteredOutcome
-        delivery = DeliveryRow(message.message_id, json.dumps(message.to_wire(), separators=(',', ':')) + '\n', outcome)
+        delivery = DeliveryRow(message.message_id, FieldCodec.compact(message.to_wire()) + '\n', outcome)
         with self.db:
             first = delivery.insert_if_absent(self.db).rowcount
         if not first:

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(