From af6438aad5cddb1a79528ec59b28b8f3c234d661 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Fri, 23 Oct 2026 04:30:27 -0400
Subject: [PATCH 1/3] Let input forwarding own nominal inbox dispatch through
 the existing MRO

---
 src/agent_comms/agent_backend.py          | 28 ++-------
 src/agent_comms/native_session_prepare.py | 29 +---------
 src/agent_comms/prompt_state.py           | 22 ++++---
 src/agent_comms/turn_inputs.py            | 70 +++++++++++++++++++----
 4 files changed, 77 insertions(+), 72 deletions(-)

diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py
index 743f8a25de..cf11ac8352 100644
--- a/src/agent_comms/agent_backend.py
+++ b/src/agent_comms/agent_backend.py
@@ -23,7 +23,7 @@
 from dataclasses import dataclass
 from enum import Enum
 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
@@ -136,15 +136,6 @@ class InputContent:
     contributions: tuple[InputContributionCoordinates, ...] = ()
 
 
-class TurnInboxReceiver(Protocol):
-    """What a running turn does with each record its inbox carries."""
-
-    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]: ...
-
-
 class TurnInbox(asyncio.Queue):
     """The live inbox parks its head until the existing observer sees READY."""
     def __init__(self):
@@ -176,7 +167,7 @@ def hold(self, item):
 
 
 class TurnInboxItem(ABC):
-    """An inbox item owns its carried inputs and delivery to a running turn."""
+    """An inbox item carries its input identities; the forwarding owner dispatches it."""
 
     @property
     def carried_inputs(self) -> tuple[InputId, ...]:
@@ -185,8 +176,6 @@ def carried_inputs(self) -> tuple[InputId, ...]:
     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)
@@ -204,9 +193,7 @@ 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)
@@ -218,8 +205,7 @@ 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)
@@ -233,8 +219,7 @@ 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)
@@ -243,8 +228,7 @@ 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"):
diff --git a/src/agent_comms/native_session_prepare.py b/src/agent_comms/native_session_prepare.py
index d2f31411c6..1857eb0fb7 100644
--- a/src/agent_comms/native_session_prepare.py
+++ b/src/agent_comms/native_session_prepare.py
@@ -18,40 +18,15 @@
 from .coordinator import Coordination
 from .pi_payloads import StateData
 from .session_fence import session_writer_fence
-from .prompt_state import UnwrittenPrompt
+from . import prompt_state
 
 if TYPE_CHECKING:
     from .pi_native_backend import PersistentPiSession
-    from .following_inputs import FollowingInputs
-
-
-class PreparedSession(UnwrittenPrompt):
-    acknowledged = False
-
-    def permits_retention(self, session: backend.NativeSession) -> bool:
-        return session.output.clean
-
-    async def deliver_to(
-        self, inputs: FollowingInputs, session: backend.NativeSession,
-    ) -> bool:
-        session.native.forget_context()
-        state = session.native.attestation.require_state()
-        identity = session.native.attestation.require_identity()
-        session.launch.session.attest(identity)
-        if (
-            state.is_streaming.value_or(True)
-            or state.is_compacting.value_or(True)
-            or not state.pending_message_count.matches(0)
-        ):
-            raise NativePiUnavailable("Native preparation did not attest the idle saved session")
-        session.active_session_file = identity.session_file
-        await session.stats.request(session)
-        return True
 
 
 class NativeSessionPreparation(backend.NativeSession):
     """Share native launch/attestation with turns; retain an idle child, send no prompt."""
-    prompt_type = PreparedSession
+    prompt_type = prompt_state.PreparedSession
 
     def finish_result(self) -> Iterable[events.AgentEvent]:
         if not self.native_session.child_state.retained:
diff --git a/src/agent_comms/prompt_state.py b/src/agent_comms/prompt_state.py
index 4248556a65..ae201a8b6a 100644
--- a/src/agent_comms/prompt_state.py
+++ b/src/agent_comms/prompt_state.py
@@ -8,11 +8,9 @@
 
 from . import turn_failure as failures
 from .agent_backend import TurnInboxItem
-from .locked_store import RegistryFile
-from .maintenance_barrier import MaintenanceBarrier
 
 if TYPE_CHECKING:
-    from .backend import TurnSession
+    from .backend import NativeSession, TurnSession
     from .pi_events import AgentSettled, InputCommitted, MessageStart, Response
 
 
@@ -64,15 +62,6 @@ class UnwrittenPrompt(PromptState, TurnInboxItem):
     def dispatch(self):
         return UnacknowledgedPrompt()
 
-    async def deliver_to(self, inputs, session):
-        def write():
-            session.grant_prompt()
-            session.native.send(session.command)
-        async with MaintenanceBarrier(RegistryFile.path_in(session.startup.root)).ingress_async() as status:
-            with status.phase.send_new(session.send_boundary, None, session.command.input_id, session.command.message) as outcome:
-                completion = outcome.dispatch(inputs, session, self, write)
-        return await completion
-
     def refuse(self, inputs, session):
         session.output.record_failure(failures.InputMissing("Original input was cleared before send."))
         session.finish_request()
@@ -84,6 +73,15 @@ def raise_native_failure(self, error, attestation) -> Never:
         raise NativePiInputNotSent(error, attestation) from error
 
 
+class PreparedSession(UnwrittenPrompt):
+    """An idle-session preparation owns no user prompt."""
+
+    acknowledged = False
+
+    def permits_retention(self, session: NativeSession) -> bool:
+        return session.output.clean
+
+
 @dataclass(frozen=True)
 class AcknowledgedPrompt(PromptState):
     response: Response
diff --git a/src/agent_comms/turn_inputs.py b/src/agent_comms/turn_inputs.py
index 3d35c99710..9e5c248e90 100644
--- a/src/agent_comms/turn_inputs.py
+++ b/src/agent_comms/turn_inputs.py
@@ -1,12 +1,13 @@
 """Input forwarding owns pending native IDs and never replays written inputs.
 
-Each inbox member delivers itself to this receiver. This owner converts it to
-its Pi command when it writes it to the running child.
+Declared handlers on this owner select each inbox operation by its nominal
+MRO, and convert it to its Pi command at the owning send boundary.
 """
 
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Awaitable, Callable, Iterable
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
@@ -27,9 +28,12 @@
 from .field_codec import Id128Text
 from .locked_store import RegistryFile
 from .maintenance_barrier import MaintenanceBarrier
+from .mro_dispatch import MroDispatch, handles
+from .native_failures import NativePiUnavailable
+from .prompt_state import PreparedSession, UnwrittenPrompt
 
 if TYPE_CHECKING:
-    from .backend import TurnSession
+    from .backend import NativeSession, TurnSession
     from .pi_events import MessageStart, Response
 
 
@@ -38,7 +42,7 @@
 
 
 @dataclass
-class InputForwarding:
+class InputForwarding(MroDispatch):
     queue: TurnInbox = field(default_factory=TurnInbox)
     pending: dict[str, commands.Prompt] = field(default_factory=dict)
     accepted: set[str] = field(default_factory=set)
@@ -82,19 +86,59 @@ def retire(self, native_id: str) -> commands.Prompt:
         self.changed.set()
         return item
 
-    async def forward(self, session: TurnSession) -> None:
+    async def forward(self, session: NativeSession) -> None:
         # Inputs that wait through one tool batch reach the model together,
         # not one model call apiece.
         await self.write(session, commands.SetSteeringMode(mode="all"))
         while True:
             self.current = await self.queue.get()
             try:
-                if not await self.current.deliver_to(self, session):
+                if not await self.dispatch(self.current, session):
                     return
             finally:
                 self.current = None
 
-    async def write(self, session: TurnSession, *written: commands.PiCommand) -> None:
+    async def consume_handlers(
+        self, item: TurnInboxItem,
+        handlers: Iterable[Callable[..., Awaitable[bool]]], session: NativeSession,
+    ) -> bool:
+        # One inbox item performs one operation. PreparedSession must not also
+        # execute its UnwrittenPrompt ancestor and accidentally send a prompt.
+        return await next(iter(handlers))(item, session)
+
+    @handles(TurnInboxItem)
+    async def unhandled(self, item: TurnInboxItem, session: NativeSession) -> bool:
+        raise TypeError(f"No input handler declared for {type(item).__name__}")
+
+    @handles(UnwrittenPrompt)
+    async def original_prompt(self, item: UnwrittenPrompt, session: TurnSession) -> bool:
+        def write():
+            session.grant_prompt()
+            session.native.send(session.command)
+        async with MaintenanceBarrier(RegistryFile.path_in(session.startup.root)).ingress_async() as status:
+            with status.phase.send_new(
+                session.send_boundary, None, session.command.input_id, session.command.message,
+            ) as outcome:
+                completion = outcome.dispatch(self, session, item, write)
+        return await completion
+
+    @handles(PreparedSession)
+    async def prepare_session(self, item: PreparedSession, session: NativeSession) -> bool:
+        session.native.forget_context()
+        state = session.native.attestation.require_state()
+        identity = session.native.attestation.require_identity()
+        session.launch.session.attest(identity)
+        if (
+            state.is_streaming.value_or(True)
+            or state.is_compacting.value_or(True)
+            or not state.pending_message_count.matches(0)
+        ):
+            raise NativePiUnavailable("Native preparation did not attest the idle saved session")
+        session.active_session_file = identity.session_file
+        await session.stats.request(session)
+        return True
+
+    async def write(self, session: NativeSession, *written: commands.PiCommand) -> None:
         for command in written:
             session.native.send(command)
         await session.native.drain()
@@ -113,7 +157,8 @@ async def queued(self, input_ids: tuple[InputId, ...]) -> tuple[str, ...]:
                 return candidates
             await self.changed.wait()
 
-    async def clear_queue(self, session: TurnSession, item: ClearQueue) -> bool:
+    @handles(ClearQueue)
+    async def clear_queue(self, item: ClearQueue, session: TurnSession) -> bool:
         command = commands.ClearQueue(input_ids=await self.queued(item.input_ids))
         if not command.input_ids:
             return True
@@ -131,7 +176,8 @@ async def clear_queue(self, session: TurnSession, item: ClearQueue) -> bool:
         session.rejected_signal.set()
         return True
 
-    async def use_settings(self, session: TurnSession, item: UseSettings) -> bool:
+    @handles(UseSettings)
+    async def use_settings(self, item: UseSettings, session: TurnSession) -> bool:
         # Pi resets thinking to the new model's default, so the level follows the model.
         await self.write(
             session,
@@ -142,7 +188,8 @@ async def use_settings(self, session: TurnSession, item: UseSettings) -> bool:
         )
         return True
 
-    async def send_now(self, session: TurnSession, item: SendNow) -> bool:
+    @handles(SendNow)
+    async def send_now(self, item: SendNow, session: TurnSession) -> bool:
         candidates = await self.queued(item.input_ids)
         if not candidates:
             return True
@@ -156,7 +203,8 @@ async def send_now(self, session: TurnSession, item: SendNow) -> bool:
                 ))
         return await completion
 
-    async def place(self, session: TurnSession, item: InputRequest) -> bool:
+    @handles(InputRequest)
+    async def place(self, item: InputRequest, session: TurnSession) -> bool:
         public_id = item.input_id.value
         native_id = Id128Text.mint()
         text = item.content.text

From d40a65fdd9bd4b16ec6700e2672e95d9f054ea3c Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Fri, 23 Oct 2026 04:40:11 -0400
Subject: [PATCH 2/3] Make FieldCodec own compact JSON layout across its
 consumers

---
 src/agent_comms/acp_extension.py              |  5 ++--
 src/agent_comms/agent_backend.py              | 13 +++-------
 src/agent_comms/assignment_states.py          |  8 +++---
 src/agent_comms/audience_manifest.py          |  7 ++----
 src/agent_comms/cli_bridges/shared.py         |  2 +-
 src/agent_comms/connector_store.py            |  3 +--
 src/agent_comms/context_segments/awareness.py |  5 ++--
 .../context_segments/optional_awareness.py    |  2 +-
 .../context_segments/partner_activity.py      |  2 +-
 .../context_segments/relationships.py         |  2 +-
 src/agent_comms/context_segments/wake.py      | 25 ++++---------------
 src/agent_comms/context_store.py              |  2 +-
 src/agent_comms/exporting.py                  |  3 +--
 src/agent_comms/field_codec.py                | 15 +++++++++++
 src/agent_comms/fresh_private_session.py      |  5 ++--
 src/agent_comms/goal_attempts.py              |  4 +--
 src/agent_comms/goals.py                      |  3 +--
 src/agent_comms/nk_foreground.py              |  3 +--
 src/agent_comms/pi_helper.py                  |  3 +--
 src/agent_comms/plan_phases.py                |  6 ++++-
 src/agent_comms/private_sidecar.py            | 10 +++-----
 src/agent_comms/response_acknowledgement.py   |  7 ++----
 src/agent_comms/retained_task_facts.py        |  3 +--
 src/agent_comms/selected_source_batch.py      |  4 +--
 src/agent_comms/selected_write_plan.py        |  7 ++----
 src/agent_comms/threads.py                    |  3 +--
 src/agent_comms/turn_context.py               |  6 ++---
 src/agent_comms/typed_table.py                |  2 +-
 src/agent_comms/working_memory_labels.py      |  8 +++---
 29 files changed, 73 insertions(+), 95 deletions(-)

diff --git a/src/agent_comms/acp_extension.py b/src/agent_comms/acp_extension.py
index 34fba48c9a..cf22972a85 100644
--- a/src/agent_comms/acp_extension.py
+++ b/src/agent_comms/acp_extension.py
@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-import json
 from abc import abstractmethod
 from dataclasses import dataclass, field, replace
 from hashlib import sha256
@@ -399,7 +398,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()
+            FieldCodec.compact(FieldCodec.encode(self), sort_keys=True).encode()
         ).hexdigest()
 
 
@@ -571,7 +570,7 @@ def status(self) -> str | None:
     @property
     def digest(self) -> str:
         return sha256(
-            json.dumps(FieldCodec.encode(self), sort_keys=True, separators=(",", ":")).encode()
+            FieldCodec.compact(FieldCodec.encode(self), sort_keys=True).encode()
         ).hexdigest()
 
 
diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py
index cf11ac8352..868bba037a 100644
--- a/src/agent_comms/agent_backend.py
+++ b/src/agent_comms/agent_backend.py
@@ -136,14 +136,14 @@ class InputContent:
     contributions: tuple[InputContributionCoordinates, ...] = ()
 
 
-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()
 
@@ -156,11 +156,11 @@ 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)
@@ -177,7 +177,6 @@ def refuse(self, inputs, turn):
         inputs.refuse(turn, self.carried_inputs)
 
 
-
 @dataclass(frozen=True, kw_only=True)
 class InputRequest(TurnInboxItem):
     """One queued user input handed to a backend: identity, content and busy policy.
@@ -195,7 +194,6 @@ def carried_inputs(self) -> tuple[InputId, ...]:
 
 
 
-
 @dataclass(frozen=True)
 class SendNow(TurnInboxItem):
     """Ask the running turn to place these queued steering inputs immediately.
@@ -207,7 +205,6 @@ class SendNow(TurnInboxItem):
 
 
 
-
 @dataclass(frozen=True)
 class UseSettings(TurnInboxItem):
     """Ask the running turn to use the thread's stored model and thinking level.
@@ -221,7 +218,6 @@ class UseSettings(TurnInboxItem):
 
 
 
-
 @dataclass(frozen=True)
 class ClearQueue(TurnInboxItem):
     """Clear only these queued user inputs, never delivered bus inputs."""
@@ -230,7 +226,6 @@ class ClearQueue(TurnInboxItem):
 
 
 
-
 class CompactionOutcome(DeclaredFamily, DeclaredReply, affix="Compaction"):
     """The result of one requested compaction."""
 
diff --git a/src/agent_comms/assignment_states.py b/src/agent_comms/assignment_states.py
index a0112fc050..024c4c216f 100644
--- a/src/agent_comms/assignment_states.py
+++ b/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:
diff --git a/src/agent_comms/audience_manifest.py b/src/agent_comms/audience_manifest.py
index 4bc85f1492..b3cb48f283 100644
--- a/src/agent_comms/audience_manifest.py
+++ b/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,9 +42,7 @@ 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
-        ).encode("utf-8")
+        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
     if len(encoded) > limit:
diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py
index c72a31679f..afcfead392 100644
--- a/src/agent_comms/cli_bridges/shared.py
+++ b/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:
diff --git a/src/agent_comms/connector_store.py b/src/agent_comms/connector_store.py
index 676874a2ac..03d817efef 100644
--- a/src/agent_comms/connector_store.py
+++ b/src/agent_comms/connector_store.py
@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 import hashlib
-import json
 import secrets
 import time
 from collections.abc import Iterator
@@ -66,7 +65,7 @@ class AuthorizationQuery:
     @property
     def request_id(self) -> str:
         """A repeated query reuses its wait, never another client's PKCE challenge."""
-        return digest(json.dumps(FieldCodec.encode(self), sort_keys=True, separators=(",", ":")))
+        return digest(FieldCodec.compact(FieldCodec.encode(self), sort_keys=True))
 
     def redirect(self, code: str) -> str:
         return f"{self.redirect_uri}?{urlencode({'code': code, 'state': self.state})}"
diff --git a/src/agent_comms/context_segments/awareness.py b/src/agent_comms/context_segments/awareness.py
index d89f9815aa..8558894c3d 100644
--- a/src/agent_comms/context_segments/awareness.py
+++ b/src/agent_comms/context_segments/awareness.py
@@ -1,6 +1,5 @@
 """Bounded sequence references to the canonical bus, never delivery authority."""
 
-import json
 from dataclasses import dataclass
 
 from ..field_codec import FieldCodec
@@ -20,9 +19,9 @@ def capture(cls, path, records):
                                bus=str(path), sources=sources)
 
     def values(self):
-        return {"pointers": json.dumps({
+        return {"pointers": FieldCodec.compact({
             "bus": self.bus, "sources": [FieldCodec.encode(source) for source in self.sources],
-        }, ensure_ascii=True, separators=(",", ":"))}
+        }, ensure_ascii=True)}
 
 
 @dataclass(frozen=True, kw_only=True)
diff --git a/src/agent_comms/context_segments/optional_awareness.py b/src/agent_comms/context_segments/optional_awareness.py
index 14157ae34a..7748a149f2 100644
--- a/src/agent_comms/context_segments/optional_awareness.py
+++ b/src/agent_comms/context_segments/optional_awareness.py
@@ -179,7 +179,7 @@ def context(self) -> dict:
         }
 
     def require_resource_budget(self, max_text_bytes: int) -> None:
-        text = json.dumps(FieldCodec.encode(self.context()), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        text = FieldCodec.compact(FieldCodec.encode(self.context()), ensure_ascii=False, sort_keys=True)
         if len(text.encode("utf-8")) > max_text_bytes:
             raise ProjectionUnavailableError("binding awareness exceeds the byte budget")
 
diff --git a/src/agent_comms/context_segments/partner_activity.py b/src/agent_comms/context_segments/partner_activity.py
index bc0d81e966..9c85e26191 100644
--- a/src/agent_comms/context_segments/partner_activity.py
+++ b/src/agent_comms/context_segments/partner_activity.py
@@ -57,4 +57,4 @@ def capture(cls, owner, relations, goals):
         return (super().capture((OwnerProvenance(owner.incarnation, revision),), changes=changes),)
 
     def values(self):
-        return {"changes": json.dumps(FieldCodec.encode(self.changes), separators=(",", ":"))}
+        return {"changes": FieldCodec.compact(FieldCodec.encode(self.changes))}
diff --git a/src/agent_comms/context_segments/relationships.py b/src/agent_comms/context_segments/relationships.py
index 3071588783..96c8e1b4db 100644
--- a/src/agent_comms/context_segments/relationships.py
+++ b/src/agent_comms/context_segments/relationships.py
@@ -23,4 +23,4 @@ def capture(cls, owner, relations):
         return (super().capture((OwnerProvenance(owner.incarnation, revision),), records=records),)
 
     def values(self):
-        return {"records": json.dumps(FieldCodec.encode(self.records), separators=(",", ":"))}
+        return {"records": FieldCodec.compact(FieldCodec.encode(self.records))}
diff --git a/src/agent_comms/context_segments/wake.py b/src/agent_comms/context_segments/wake.py
index 8b4ccc28c2..a9240cdd7f 100644
--- a/src/agent_comms/context_segments/wake.py
+++ b/src/agent_comms/context_segments/wake.py
@@ -8,7 +8,6 @@
 
 from __future__ import annotations
 
-import json
 from dataclasses import dataclass
 
 from agent_comms.coordination_tables.assignments import WakeAssignment
@@ -94,16 +93,8 @@ def values(self):
                     "body": initial.message.body,
                 }
             )
-        selected_line = json.dumps(
-            selected,
-            ensure_ascii=True,
-            separators=(",", ":"),
-        )
-        work_context = json.dumps(
-            FieldCodec.encode(self.work_context),
-            ensure_ascii=True,
-            separators=(",", ":"),
-        )
+        selected_line = FieldCodec.compact(selected, ensure_ascii=True)
+        work_context = FieldCodec.compact(FieldCodec.encode(self.work_context), ensure_ascii=True)
         sent_values = []
         for outcome in self.sent_outcomes:
             message, reference = outcome.message, outcome.answer
@@ -117,11 +108,7 @@ def values(self):
                     FieldCodec.encode(reference) if reference is not None else None
                 ),
             })
-        sent_outcomes = json.dumps(
-            sent_values,
-            ensure_ascii=True,
-            separators=(",", ":"),
-        )
+        sent_outcomes = FieldCodec.compact(sent_values, ensure_ascii=True)
         return dict(
             count=len(selected),
             selected=selected_line,
@@ -148,15 +135,13 @@ def values(self):
         self.assignment.require_selected_source(self.source, self.owner)
         message = self.source.message
         return dict(
-            message=json.dumps(
+            message=FieldCodec.compact(
                 {
                     "source_seq": self.assignment.wire_seq,
                     "sender": message.sender,
                     "target": message.target,
                     "audience": self.assignment.audience.value,
                     "body": message.body,
-                },
-                ensure_ascii=True,
-                separators=(",", ":"),
+                }, ensure_ascii=True,
             )
         )
diff --git a/src/agent_comms/context_store.py b/src/agent_comms/context_store.py
index 4462a72683..4747261c1c 100644
--- a/src/agent_comms/context_store.py
+++ b/src/agent_comms/context_store.py
@@ -186,7 +186,7 @@ class ContextSegmentVersion(TypedTable):
 
     @classmethod
     def append(cls, db: sqlite3.Connection, segment: RecordedSegmentManifest) -> int:
-        encoded = json.dumps(FieldCodec.encode(segment), sort_keys=True, separators=(",", ":"))
+        encoded = FieldCodec.compact(FieldCodec.encode(segment), sort_keys=True)
         digest = hashlib.sha256(encoded.encode()).hexdigest()
         rows = cls.select(db, digest=digest)
         if rows:
diff --git a/src/agent_comms/exporting.py b/src/agent_comms/exporting.py
index a4960abea6..e63f80a52b 100644
--- a/src/agent_comms/exporting.py
+++ b/src/agent_comms/exporting.py
@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import hashlib
-import json
 import math
 import os
 from abc import abstractmethod
@@ -42,7 +41,7 @@ def row(self, message: Message) -> bytes:
 
     @staticmethod
     def json_record(record: Mapping[str, object]) -> str:
-        return json.dumps(record, ensure_ascii=False, separators=(",", ":"))
+        return FieldCodec.compact(record, ensure_ascii=False)
 
 
 class SelectableWireExportFormat(WireExportFormat):
diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py
index 7cc42ffe9f..081583e7dc 100644
--- a/src/agent_comms/field_codec.py
+++ b/src/agent_comms/field_codec.py
@@ -1352,6 +1352,21 @@ def encode(cls, value: object, annotation: object = None) -> Any: ...
     def encode(cls, value: object, annotation: object = None) -> Any:
         return compile_decoder(annotation).encode(value)
 
+    @staticmethod
+    def compact(
+        value: object, *, ensure_ascii: bool = True,
+        sort_keys: bool = False, allow_nan: bool = True,
+    ) -> str:
+        """Write an already encoded JSON value without layout whitespace.
+
+        Callers retain their boundary's Unicode, ordering and numeric contract;
+        declared values pass through ``encode`` before this text boundary.
+        """
+        return json.dumps(
+            value, ensure_ascii=ensure_ascii, sort_keys=sort_keys,
+            allow_nan=allow_nan, separators=(",", ":"),
+        )
+
     @classmethod
     def encode_document(cls, value: object, declared: object) -> object:
         """A document whose declaration names its leaf member stores that member untagged."""
diff --git a/src/agent_comms/fresh_private_session.py b/src/agent_comms/fresh_private_session.py
index 3f0d8ab332..c7e29b75a9 100644
--- a/src/agent_comms/fresh_private_session.py
+++ b/src/agent_comms/fresh_private_session.py
@@ -2,14 +2,13 @@
 
 from __future__ import annotations
 
-import json
 import os
 from datetime import UTC, datetime
 from pathlib import Path
 from uuid import uuid4
 
 from agent_comms.child_process import Platform
-from .field_codec import RecordedTimestamp
+from .field_codec import FieldCodec, RecordedTimestamp
 from .native_entries import SessionEntry
 from .native_failures import NativePiUnavailable
 from agent_comms.native_session_files import NativeSessionFiles
@@ -33,7 +32,7 @@ def create_fresh_private_session(
         id=session_id, version=3, cwd=str(worktree),
         timestamp=RecordedTimestamp.written(datetime.now(UTC)),
     )
-    encoded = json.dumps(header.to_wire(), separators=(",", ":"), allow_nan=False).encode() + b"\n"
+    encoded = FieldCodec.compact(header.to_wire(), allow_nan=False).encode() + b"\n"
     try:
         descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
         try:
diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py
index 9341f10c50..eb7624d175 100644
--- a/src/agent_comms/goal_attempts.py
+++ b/src/agent_comms/goal_attempts.py
@@ -31,7 +31,7 @@
 
 from .child_process import Platform
 from .declared_family import DeclaredFamily
-from .field_codec import MalformedRecord
+from .field_codec import FieldCodec, MalformedRecord
 from .goal_attempt_identity import FailureNotObserved, GoalAttemptIdentity
 from .goal_attempt_phase import (
     ClaimedAttempt,
@@ -598,7 +598,7 @@ def record_provider_usage(
         """Attribute one provider response to its claimed goal attempt exactly once."""
         if not response_id:
             raise ValueError("Provider response ID is required.")
-        raw = json.dumps(usage.to_wire(), sort_keys=True, separators=(",", ":"), allow_nan=False)
+        raw = FieldCodec.compact(usage.to_wire(), sort_keys=True, allow_nan=False)
         reservation = self.reservation
 
         def write(conn: sqlite3.Connection) -> str:
diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py
index 83266488b0..fc239e2c29 100644
--- a/src/agent_comms/goals.py
+++ b/src/agent_comms/goals.py
@@ -4,7 +4,6 @@
 
 from agent_comms.lifecycle_capability import Active
 
-import json
 import os
 import re
 import shlex
@@ -471,7 +470,7 @@ def stored(cls, repository: str, sha: str) -> tuple[CheckedCommit, ...]:
 
     def store(self, repository: str, sha: str) -> None:
         self.git(repository, *self.NOTES, "append", "-m",
-                 json.dumps(FieldCodec.encode(self), separators=(",", ":")), sha)
+                 FieldCodec.compact(FieldCodec.encode(self)), sha)
 
 
 class GoalOrder(DeclaredFamily, affix="GoalOrder"):
diff --git a/src/agent_comms/nk_foreground.py b/src/agent_comms/nk_foreground.py
index 3d2ea51478..16264ea4a9 100644
--- a/src/agent_comms/nk_foreground.py
+++ b/src/agent_comms/nk_foreground.py
@@ -13,7 +13,6 @@
 
 import argparse
 import asyncio
-import json
 import os
 import re
 import select
@@ -116,7 +115,7 @@ def reserve_foreground_owner(
 
 
 def _emit(payload: dict[str, object]) -> None:
-    json.dump(payload, sys.stdout, separators=(",", ":"))
+    sys.stdout.write(FieldCodec.compact(payload))
     sys.stdout.write("\n")
     sys.stdout.flush()
 
diff --git a/src/agent_comms/pi_helper.py b/src/agent_comms/pi_helper.py
index f0a41b53f8..0ef23ba345 100644
--- a/src/agent_comms/pi_helper.py
+++ b/src/agent_comms/pi_helper.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import json
 import shutil
 from collections.abc import Mapping
 from dataclasses import dataclass
@@ -95,7 +94,7 @@ def node_launch(cls, package: Path, environment: dict[str, str]):
     @classmethod
     async def run(cls, request, *, cwd: Path, env: Mapping[str, str] = INHERITED_ENVIRONMENT):
         command, environment = cls.launch(request, dict(env))
-        payload = json.dumps(FieldCodec.encode(request), separators=(",", ":"), allow_nan=False)
+        payload = FieldCodec.compact(FieldCodec.encode(request), allow_nan=False)
         outcome = await BoundedRun.run(
             command,
             input=payload.encode(),
diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py
index e8bd49d81d..a6a0f1bc91 100644
--- a/src/agent_comms/plan_phases.py
+++ b/src/agent_comms/plan_phases.py
@@ -644,7 +644,11 @@ class OwnedDispatchPhase(RemainingSlopPhase):
     evidence = AllEvidence(
         (
             PatternAbsent(r"class TurnInboxReceiver\b"),
-            PatternAbsent(r"separators=\(\",\", ?\":\"\)"),
+            PatternAbsent(
+                r"""separators\s*=\s*\(\s*["'],["']\s*,\s*["']:["']\s*\)""",
+                ("src", ":!src/agent_comms/field_codec.py"),
+            ),
+            PatternPresent(r"def compact\(", ("src/agent_comms/field_codec.py",)),
         )
     )
 
diff --git a/src/agent_comms/private_sidecar.py b/src/agent_comms/private_sidecar.py
index 5c7c195f60..4086e30043 100644
--- a/src/agent_comms/private_sidecar.py
+++ b/src/agent_comms/private_sidecar.py
@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import hashlib
-import json
 import os
 import secrets
 import sqlite3
@@ -29,6 +28,7 @@
 
 from agent_comms.locked_store import SQLiteFile
 from agent_comms.coordination_errors import IdentityConflict
+from agent_comms.field_codec import FieldCodec
 
 from .private_path import ObservedPath
 from .store_files import StoreOwner
@@ -218,7 +218,7 @@ def _schema(row_type: type[TypedTable]) -> dict[str, str]:
 
 
 def _digest(row_type: type[TypedTable]) -> str:
-    return hashlib.sha256(json.dumps(_schema(row_type), separators=(",", ":")).encode()).hexdigest()
+    return hashlib.sha256(FieldCodec.compact(_schema(row_type)).encode()).hexdigest()
 
 
 def _verify_schema(connection: sqlite3.Connection, row_type: type[TypedTable]) -> None:
@@ -367,7 +367,7 @@ def sidecar_connection(
 
 def encode_request(text: str, *, streaming_behavior: str | None = None) -> str:
     """JSON-encode like the pinned native ``_claimNativeInput`` request."""
-    return json.dumps(
+    return FieldCodec.compact(
         {
             "kind": "prompt",
             "text": text,
@@ -375,9 +375,7 @@ def encode_request(text: str, *, streaming_behavior: str | None = None) -> str:
             "streamingBehavior": streaming_behavior,
             "expandPromptTemplates": True,
             "source": "rpc",
-        },
-        separators=(",", ":"),
-        ensure_ascii=False,
+        }, ensure_ascii=False,
     )
 
 
diff --git a/src/agent_comms/response_acknowledgement.py b/src/agent_comms/response_acknowledgement.py
index f87729b996..5b554fdc9b 100644
--- a/src/agent_comms/response_acknowledgement.py
+++ b/src/agent_comms/response_acknowledgement.py
@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import hashlib
-import json
 import math
 from abc import abstractmethod
 from collections.abc import Mapping
@@ -14,12 +13,10 @@
     MAX_IDENTIFIER_CHARS, require_bounded, require_nonempty, validate_execution_id,
 )
 from agent_comms.errors import RelationViolationError
-from agent_comms.field_codec import MalformedRecord, NullMember, RecordMember, SubclassUnion
+from agent_comms.field_codec import FieldCodec, MalformedRecord, NullMember, RecordMember, SubclassUnion
 
 def _canonical(value: object) -> bytes:
-    return json.dumps(
-        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
-    ).encode("utf-8")
+    return FieldCodec.compact(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode("utf-8")
 
 
 def public_envelope_digest(record: Mapping[str, object]) -> str:
diff --git a/src/agent_comms/retained_task_facts.py b/src/agent_comms/retained_task_facts.py
index 9068bb2fbb..ba7d21d1b6 100644
--- a/src/agent_comms/retained_task_facts.py
+++ b/src/agent_comms/retained_task_facts.py
@@ -246,8 +246,7 @@ def original_input_facts(self, inputs: InputDispositions) -> tuple[ExactTaskFact
 
     @staticmethod
     def canonical_bytes(record: object) -> bytes:
-        return json.dumps(record, sort_keys=True, separators=(",", ":"),
-                          allow_nan=False).encode()
+        return FieldCodec.compact(record, sort_keys=True, allow_nan=False).encode()
 
     def original_text_source(self, reference: MessageReference) -> Message | StoredInput:
         originals = {key: source for fact in self.facts
diff --git a/src/agent_comms/selected_source_batch.py b/src/agent_comms/selected_source_batch.py
index c50065785c..d4aaaccce9 100644
--- a/src/agent_comms/selected_source_batch.py
+++ b/src/agent_comms/selected_source_batch.py
@@ -1,13 +1,13 @@
 """Captured selected sources share one input; they do not create another inbox."""
 
 import hashlib
-import json
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from .bus_publication import Delivered
 from .coordination_errors import IdentityConflict
 from .coordination_tables.assignments import WakeAssignment
+from .field_codec import FieldCodec
 from .messages import InfoKind, Message
 from .native_failures import NativePiUnavailable
 from .turn_context import ReplyRouteSegment, WireProvenance
@@ -85,7 +85,7 @@ def assignment_ids(self) -> tuple[str, ...]:
     @property
     def execution_id(self) -> str:
         return "wirev1" + hashlib.sha256(
-            json.dumps(self.assignment_ids, separators=(",", ":")).encode()
+            FieldCodec.compact(self.assignment_ids).encode()
         ).hexdigest()
 
     @property
diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py
index 229c35f005..aa8623b633 100644
--- a/src/agent_comms/selected_write_plan.py
+++ b/src/agent_comms/selected_write_plan.py
@@ -10,7 +10,6 @@
 
 import base64
 import hashlib
-import json
 import os
 from contextlib import suppress
 from dataclasses import dataclass, fields
@@ -220,7 +219,7 @@ def submit(
                 expected_attempt_ordinal=1, operation_id=operation_id, resource=str(path),
                 contents_b64=base64.b64encode(raw).decode("ascii"), digest=hashlib.sha256(raw).hexdigest(),
             )
-            data = json.dumps(FieldCodec.encode(record), sort_keys=True, separators=(",", ":")).encode()
+            data = FieldCodec.compact(FieldCodec.encode(record), sort_keys=True).encode()
             try:
                 fd = os.open(
                     record_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600
@@ -270,6 +269,4 @@ def applied(self, assignment: WakeAssignment, owner: Thread, operation_id: str)
         if accepted.operation_id != operation_id:
             raise IdentityConflict("Selected write intent is no longer unique")
         completed = accepted.applied()
-        AtomicFile(path).write_bytes(json.dumps(
-            FieldCodec.encode(completed), sort_keys=True, separators=(",", ":")
-        ).encode())
+        AtomicFile(path).write_bytes(FieldCodec.compact(FieldCodec.encode(completed), sort_keys=True).encode())
diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py
index 3466334e40..ede337ca43 100644
--- a/src/agent_comms/threads.py
+++ b/src/agent_comms/threads.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import json
 import os
 import time
 from abc import ABC, abstractmethod
@@ -333,7 +332,7 @@ def native_launch_arguments(
         catalog = ThreadToolCatalogResult(
             tuple(NativeRequestTools.catalog()), self.service_tier, launch.arguments(),
         )
-        AtomicFile(path).write_text(json.dumps(catalog.to_wire(), separators=(",", ":")))
+        AtomicFile(path).write_text(FieldCodec.compact(catalog.to_wire()))
         return (*arguments.argv, "--comms-catalog", str(path))
 
     def launch(self) -> ThreadLaunch:
diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py
index 9097705679..274a84b49d 100644
--- a/src/agent_comms/turn_context.py
+++ b/src/agent_comms/turn_context.py
@@ -818,7 +818,7 @@ def provider_value(self): ...
     def render_into(self, prompt_parts, provider): ...
 
     def text(self):
-        return json.dumps(self.provider_value(), ensure_ascii=False, separators=(",", ":"))
+        return FieldCodec.compact(self.provider_value(), ensure_ascii=False)
 
     def measured_manifest(self):
         # The SDK measures/hashes its exact original JSON representation.
@@ -1006,7 +1006,7 @@ def public_spans(self) -> tuple[ContextSpan, ...]:
         offset = 1
         spans = []
         for tool in self.tools:
-            text = json.dumps(tool, ensure_ascii=False, separators=(",", ":"))
+            text = FieldCodec.compact(tool, ensure_ascii=False)
             coordinates = ContributionCoordinates.capture(type(self), self.provenance, offset, text)
             spans.append(ContextSpan(self.sha256, coordinates))
             offset += coordinates.length + 1
@@ -1117,7 +1117,7 @@ def values(self) -> dict[str, object]:
             name=self.owner.name,
             parent=self.parent,
             project=self.project,
-            peers=json.dumps(FieldCodec.encode(self.peers), separators=(",", ":")),
+            peers=FieldCodec.compact(FieldCodec.encode(self.peers)),
         )
 
 
diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py
index 04c4777692..0b4ba43f24 100644
--- a/src/agent_comms/typed_table.py
+++ b/src/agent_comms/typed_table.py
@@ -272,7 +272,7 @@ def accepts(cls, annotation: object) -> bool:
 
     @classmethod
     def to_sql(cls, value: object) -> str | None:
-        return None if value is None else json.dumps(value, separators=(",", ":"), allow_nan=False)
+        return None if value is None else FieldCodec.compact(value, allow_nan=False)
 
     @classmethod
     def from_sql(cls, value: object) -> object:
diff --git a/src/agent_comms/working_memory_labels.py b/src/agent_comms/working_memory_labels.py
index acce235be9..57a0b9a01d 100644
--- a/src/agent_comms/working_memory_labels.py
+++ b/src/agent_comms/working_memory_labels.py
@@ -44,9 +44,11 @@ class QuestionVersion:
     def current(cls, question: type[SpanQuestion], owner_rules: tuple[str, ...] = ()):
         # A relation to authored rules is a different question when that original
         # rule set changes. Other questions discard that unrelated input.
-        raw = json.dumps({"definition": question.declaration(),
-                          "owner_rules": question.disclosed_owner_rules(owner_rules)},
-                         sort_keys=True, separators=(",", ":")).encode()
+        raw = FieldCodec.compact(
+            {"definition": question.declaration(),
+             "owner_rules": question.disclosed_owner_rules(owner_rules)},
+            sort_keys=True,
+        ).encode()
         return cls(question, hashlib.sha256(raw).hexdigest())
 
 

From c1a8e8fb174933d8e101d60231ad021c35c564bd Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Fri, 23 Oct 2026 04:55:50 -0400
Subject: [PATCH 3/3] Declare inbox types before use and select its single MRO
 operation directly

---
 src/agent_comms/agent_backend.py            | 24 ++++++++++-----------
 src/agent_comms/audience_manifest.py        |  4 +++-
 src/agent_comms/response_acknowledgement.py | 13 +++++++++--
 src/agent_comms/turn_inputs.py              | 17 ++++++---------
 4 files changed, 32 insertions(+), 26 deletions(-)

diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py
index 868bba037a..241873fa34 100644
--- a/src/agent_comms/agent_backend.py
+++ b/src/agent_comms/agent_backend.py
@@ -136,7 +136,18 @@ class InputContent:
     contributions: tuple[InputContributionCoordinates, ...] = ()
 
 
-class TurnInbox(asyncio.Queue["TurnInboxItem"]):
+class TurnInboxItem(ABC):
+    """An inbox item carries its input identities; the forwarding owner dispatches it."""
+
+    @property
+    def carried_inputs(self) -> tuple[InputId, ...]:
+        return ()
+
+    def refuse(self, inputs, turn):
+        inputs.refuse(turn, self.carried_inputs)
+
+
+class TurnInbox(asyncio.Queue[TurnInboxItem]):
     """The live inbox parks its head until the existing observer sees READY."""
     def __init__(self):
         super().__init__()
@@ -166,17 +177,6 @@ def hold(self, item: TurnInboxItem) -> None:
         self.begin(item)
 
 
-class TurnInboxItem(ABC):
-    """An inbox item carries its input identities; the forwarding owner dispatches it."""
-
-    @property
-    def carried_inputs(self) -> tuple[InputId, ...]:
-        return ()
-
-    def refuse(self, inputs, turn):
-        inputs.refuse(turn, self.carried_inputs)
-
-
 @dataclass(frozen=True, kw_only=True)
 class InputRequest(TurnInboxItem):
     """One queued user input handed to a backend: identity, content and busy policy.
diff --git a/src/agent_comms/audience_manifest.py b/src/agent_comms/audience_manifest.py
index b3cb48f283..a9d35212f7 100644
--- a/src/agent_comms/audience_manifest.py
+++ b/src/agent_comms/audience_manifest.py
@@ -42,7 +42,9 @@ def _bounded_text(value: str, field: str, *, thread_name: bool = False) -> None:
 
 def _canonical_json(value: object, *, limit: int) -> bytes:
     try:
-        encoded = FieldCodec.compact(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode("utf-8")
+        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
     if len(encoded) > limit:
diff --git a/src/agent_comms/response_acknowledgement.py b/src/agent_comms/response_acknowledgement.py
index 5b554fdc9b..6388f981ce 100644
--- a/src/agent_comms/response_acknowledgement.py
+++ b/src/agent_comms/response_acknowledgement.py
@@ -13,10 +13,19 @@
     MAX_IDENTIFIER_CHARS, require_bounded, require_nonempty, validate_execution_id,
 )
 from agent_comms.errors import RelationViolationError
-from agent_comms.field_codec import FieldCodec, MalformedRecord, NullMember, RecordMember, SubclassUnion
+from agent_comms.field_codec import (
+    FieldCodec,
+    MalformedRecord,
+    NullMember,
+    RecordMember,
+    SubclassUnion,
+)
+
 
 def _canonical(value: object) -> bytes:
-    return FieldCodec.compact(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode("utf-8")
+    return FieldCodec.compact(
+        value, sort_keys=True, ensure_ascii=False, allow_nan=False,
+    ).encode("utf-8")
 
 
 def public_envelope_digest(record: Mapping[str, object]) -> str:
diff --git a/src/agent_comms/turn_inputs.py b/src/agent_comms/turn_inputs.py
index 9e5c248e90..cd74008bab 100644
--- a/src/agent_comms/turn_inputs.py
+++ b/src/agent_comms/turn_inputs.py
@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable, Iterable
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
@@ -93,19 +92,14 @@ async def forward(self, session: NativeSession) -> None:
         while True:
             self.current = await self.queue.get()
             try:
-                if not await self.dispatch(self.current, session):
+                # An inbox item performs one operation. PreparedSession must not
+                # also run its UnwrittenPrompt ancestor and send a prompt.
+                handler = next(self.handlers_for(self.current))
+                if not await handler(self.current, session):
                     return
             finally:
                 self.current = None
 
-    async def consume_handlers(
-        self, item: TurnInboxItem,
-        handlers: Iterable[Callable[..., Awaitable[bool]]], session: NativeSession,
-    ) -> bool:
-        # One inbox item performs one operation. PreparedSession must not also
-        # execute its UnwrittenPrompt ancestor and accidentally send a prompt.
-        return await next(iter(handlers))(item, session)
-
     @handles(TurnInboxItem)
     async def unhandled(self, item: TurnInboxItem, session: NativeSession) -> bool:
         raise TypeError(f"No input handler declared for {type(item).__name__}")
@@ -115,7 +109,8 @@ async def original_prompt(self, item: UnwrittenPrompt, session: TurnSession) ->
         def write():
             session.grant_prompt()
             session.native.send(session.command)
-        async with MaintenanceBarrier(RegistryFile.path_in(session.startup.root)).ingress_async() as status:
+        barrier = MaintenanceBarrier(RegistryFile.path_in(session.startup.root))
+        async with barrier.ingress_async() as status:
             with status.phase.send_new(
                 session.send_boundary, None, session.command.input_id, session.command.message,
             ) as outcome: