Let existing owners dispatch inbox inputs and encode compact JSON

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()
 
 

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,23 +136,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()
 
@@ -165,30 +167,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.
@@ -204,9 +192,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)
@@ -218,8 +203,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)
@@ -233,8 +216,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)
@@ -243,8 +224,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/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/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})}"

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)

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")
 

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))}

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))}

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,
             )
         )

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:

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):

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."""

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: