From 7f04146993882f5a4c7d165807dc2e12f310b2f5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:24:11 -0400 Subject: [PATCH 1/6] Keep representation queries total and carry nominal message IDs to their callers --- src/agent_comms/envelope_claim_transitions.py | 3 +- src/agent_comms/field_codec.py | 72 +++++++++---------- src/agent_comms/selected_write_plan.py | 6 +- src/agent_comms/typed_table.py | 12 ++-- tests/test_carry_wake_claim_turn_ids.py | 4 +- tests/test_envelope_claim_transitions.py | 4 +- tests/test_selected_tool_broker.py | 4 +- 7 files changed, 51 insertions(+), 54 deletions(-) diff --git a/src/agent_comms/envelope_claim_transitions.py b/src/agent_comms/envelope_claim_transitions.py index f7133dfc84..0b497bf023 100644 --- a/src/agent_comms/envelope_claim_transitions.py +++ b/src/agent_comms/envelope_claim_transitions.py @@ -19,6 +19,7 @@ EnvelopeSequence, FieldCodec, Id128Text, + MessageId, TurnId, NullMember, ProjectionSequence, @@ -208,7 +209,7 @@ class WakeClaim(RecordMember, RecordedScope): wire_root_id: Annotated[str, Id128Text] source_seq: int - source_message_id: Annotated[str, BoundedText] + source_message_id: MessageId wake_assignment_id: Annotated[str, BoundedText, WireName("wake_claim_id")] wake_revision: int recipient_lookup: Annotated[str, Id128Text] diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index e8f4d37a83..918aed68b1 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -659,8 +659,10 @@ class DeclaredFieldPlan(FieldPlan): @classmethod def claims(cls, declared: Field, annotation: object) -> bool: - representation = FieldCodec.representation(annotation)[1] or FieldRepresentation - return super().claims(declared, annotation) or representation.field_plan is cls + _, representations = FieldCodec.representation(annotation) + return super().claims(declared, annotation) or any( + owner.field_plan is cls for owner in (representations or (FieldRepresentation,)) + ) class TransientFieldPlan(DeclaredFieldPlan): @@ -747,10 +749,10 @@ class InlineFieldPlan(DeclaredFieldPlan): @cached_property def inlined(self) -> type: - declared, representation = FieldCodec.representation(self.annotation) - match representation: - case type() as inline if issubclass(inline, InlineRecord): - return inline.wire_record() + declared, representations = FieldCodec.representation(self.annotation) + for representation in representations: + if issubclass(representation, InlineRecord): + return representation.wire_record() return cast(type, declared) @cached_property @@ -1884,41 +1886,38 @@ def external_decoder(cls, target: type[ExternalField]) -> ExternalDecoder: return target.decode_external @classmethod - def representation(cls, annotation: object) -> tuple[object, type[FieldRepresentation] | None]: - """The declared value type and the codec owning its wire representation.""" + def representation( + cls, annotation: object, + ) -> tuple[object, tuple[type[FieldRepresentation], ...]]: + """The value type and its zero or one declared wire representation classes.""" if annotation is None: - return None, None + return None, () if annotation is object: - return object, _JsonValueRepresentation + return object, (_JsonValueRepresentation,) if get_origin(annotation) is Annotated: target, *metadata = get_args(annotation) else: target, metadata = annotation, () - representations = [ + representations = tuple( item for item in metadata if isinstance(item, type) and issubclass(item, FieldRepresentation) - ] + ) if len(representations) > 1: raise TypeError("A field must have one representation") if isinstance(target, TypeAliasType): if target is JsonValue: - return object, representations[0] if representations else _JsonValueRepresentation + return object, representations or (_JsonValueRepresentation,) if representations: - return target, representations[0] + return target, representations if target is object: - return object, _JsonValueRepresentation + return object, (_JsonValueRepresentation,) if isinstance(target, type) and issubclass(target, (NominalText, WireValue)): - return target, target.representation() - return target, None + return target, (target.representation(),) + return target, () @classmethod def _value_representation(cls, value: object, annotation: object): - if annotation is None: - return cls.representation(type(value))[1] - _, representation = cls.representation(annotation) - if representation is None: - _, representation = cls.representation(type(value)) - return representation + return cls.representation(annotation)[1] or cls.representation(type(value))[1] @classmethod def record_fields(cls, record: type) -> tuple[tuple[Field[object], str], ...]: @@ -2064,11 +2063,6 @@ def record_member[Member: DeclaredFamily]( """Read a record's declared selector without decoding its payload.""" return _FamilyRecordDecoder(family).member(data) - @classmethod - def record_data(cls, record: type, data: object) -> dict: - """Validate the envelope independently of reading its declared payloads.""" - return _RecordDecoder(record).checked(data) - @classmethod def field_plan(cls, record: type, name: str) -> FieldPlan: return next(item for item in cls._record_plan(record).bindings if item.name == name) @@ -2144,8 +2138,8 @@ def _record_schema( @classmethod def value_schema(cls, annotation: object) -> dict[str, JsonValue]: """Derive external scalar/array choices from the same decode declarations.""" - annotation, representation = cls.representation(annotation) - if representation is not None: + annotation, representations = cls.representation(annotation) + for representation in representations: schema = representation.schema() if type(None) in get_args(annotation): return cls.alternatives_schema("anyOf", [schema, {"type": "null"}]) @@ -2193,7 +2187,7 @@ def project(cls, value: object, view: str, annotation: object = None) -> JsonVal Projection is deliberately one way. Excluded secrets cannot be rebuilt from a read-only view, and a view never serves as a persistence record. """ - if cls._value_representation(value, annotation) is not None: + if cls._value_representation(value, annotation): return RepresentedProjection.of(value, view, annotation) if is_dataclass(value) and not isinstance(value, type): return cls._project_record(value, view) @@ -3172,14 +3166,14 @@ def compile_decoder(annotation: object) -> ValueDecoder[object]: ... @lru_cache(maxsize=None) def compile_decoder(annotation: object) -> ValueDecoder[object]: """Compile each immutable declaration once; value paths never parse it again.""" - declared, representation = FieldCodec.representation(annotation) - if representation is None: - return _structural_decoder(declared) - try: - owner = ValueDecoder.annotation_class(declared) - except KeyError: - owner = _ScalarDecoder - return owner.with_representation(annotation, declared, get_args(declared), representation) + declared, representations = FieldCodec.representation(annotation) + for representation in representations: + try: + owner = ValueDecoder.annotation_class(declared) + except KeyError: + owner = _ScalarDecoder + return owner.with_representation(annotation, declared, get_args(declared), representation) + return _structural_decoder(declared) def _structural_decoder(annotation: object) -> ValueDecoder[object]: diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py index 4af2e424db..e6fc1bb6fb 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -33,7 +33,7 @@ from .coordination_cohort import sealed_cohort_assignments from .declared_family import DeclaredFamily from .envelope_claim_transitions import ExistingFileClaim -from .field_codec import ClaimId, FieldCodec, Id128Text, MalformedRecord +from .field_codec import ClaimId, FieldCodec, Id128Text, MalformedRecord, MessageId from .message_reference import MessageReference from .private_path import PrivateDirectoryRole, PrivateFileRole from .thread_identity import ClaimedIdentity, ClaimScoped, RecordedThreadIncarnation @@ -63,7 +63,7 @@ class SelectedWriteIntent(DeclaredFamily, affix="WriteIntent"): schema: Literal[1] root_id: str source_seq: int - source_message_id: str + source_message_id: MessageId claim_id: str owner: RecordedThreadIncarnation claim_epoch: int @@ -172,7 +172,7 @@ def submit( if type(source_seq) is not int or source_seq <= 0: raise IdentityConflict("Selected write requires a positive source sequence") try: - FieldCodec.decode(str, source_message_id) + source_message_id = FieldCodec.decode(MessageId, source_message_id) FieldCodec.decode(str, resource) FieldCodec.decode(str, contents) except MalformedRecord as error: diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index ef52785821..bf0e39cbbc 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -185,8 +185,8 @@ class TextStorage(SqlStorage): def accepts(cls, annotation: object) -> bool: if annotation is str: return True - _, representation = FieldCodec.representation(annotation) - return representation is not None and issubclass(representation, TextRepresentation) + _, representations = FieldCodec.representation(annotation) + return any(issubclass(owner, TextRepresentation) for owner in representations) class IntegerStorage(SqlStorage): @@ -2108,11 +2108,13 @@ def producer_owners(cls) -> tuple[type, ...]: if issubclass(parent, ProjectionSource) and ("writes" not in vars(parent) or parent in writers) ) - annotations = [column.annotation for column in cls._columns] if is_dataclass(cls) else [] + annotations: list[object] = ( + [column.annotation for column in cls._columns] if is_dataclass(cls) else [] + ) while annotations: - value, representation = FieldCodec.representation(annotations.pop()) + value, representations = FieldCodec.representation(annotations.pop()) annotations.extend(get_args(value)) - for owner in (value, representation): + for owner in (value, *representations): if not isinstance(owner, type): continue owners.update( diff --git a/tests/test_carry_wake_claim_turn_ids.py b/tests/test_carry_wake_claim_turn_ids.py index 02b762c2e1..4247c62dee 100644 --- a/tests/test_carry_wake_claim_turn_ids.py +++ b/tests/test_carry_wake_claim_turn_ids.py @@ -10,7 +10,7 @@ from agent_comms.bus import Bus from agent_comms.envelope_claim_transitions import ClaimTransition, WakeClaim, WakeClaims -from agent_comms.field_codec import FieldCodec, MalformedRecord, TurnId +from agent_comms.field_codec import FieldCodec, MalformedRecord, MessageId, TurnId from agent_comms.messages import BusRow, Claimed @@ -23,7 +23,7 @@ def old_claim_bus(root: Path, *, turn_id: str = "a" * 32) -> Path: generation="b" * 32, resource_scope=WakeClaim( owner_claimed_generation=1, operation_id="c" * 32, turn_id=TurnId("a" * 32), - wire_root_id="d" * 32, source_seq=1, source_message_id="source-1", + wire_root_id="d" * 32, source_seq=1, source_message_id=MessageId.mint("source-1"), wake_assignment_id="cohort-v1:" + "e" * 64, wake_revision=1, recipient_lookup="f" * 32, execution_id="execution-1", participant_generation=1, attempt_ordinal=1, diff --git a/tests/test_envelope_claim_transitions.py b/tests/test_envelope_claim_transitions.py index fa14fc4403..b7ee6c8089 100644 --- a/tests/test_envelope_claim_transitions.py +++ b/tests/test_envelope_claim_transitions.py @@ -12,7 +12,7 @@ ClaimTransitionError, ExistingFileClaim, NativeClaim, ReleasedClaims, UnscopedClaims, WakeClaim, WakeClaims, ) -from agent_comms.field_codec import FieldCodec, MalformedRecord, TurnId +from agent_comms.field_codec import FieldCodec, MalformedRecord, MessageId, TurnId from agent_comms.messages import ClaimMessage, ClaimNoticeKind, Message G1 = "a" * 32 @@ -71,7 +71,7 @@ def test_selected_wake_binding_survives_claim_projection(tmp_path): claim_scope = WakeClaim( wire_root_id="c" * 32, source_seq=7, - source_message_id="source-7", + source_message_id=MessageId.mint("source-7"), wake_assignment_id="cohort-v1:" + "d" * 64, wake_revision=3, recipient_lookup="e" * 32, diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py index 8990846b83..ae1c25996b 100644 --- a/tests/test_selected_tool_broker.py +++ b/tests/test_selected_tool_broker.py @@ -18,7 +18,7 @@ from agent_comms.coordinator import Coordination from agent_comms.coordination_database import CoordinationStore from agent_comms.envelope_claim_transitions import WakeClaim -from agent_comms.field_codec import TurnId +from agent_comms.field_codec import MessageId, TurnId from agent_comms.native_tool_call import ( DuplicateToolCallIdentity, OwnerToolConnection, @@ -102,7 +102,7 @@ def test_owner_mode_denies_unreserved_input_before_consuming_ledger(private_comm claim_scope = WakeClaim( wire_root_id="a" * 32, source_seq=1, - source_message_id="source", + source_message_id=MessageId.mint("source"), wake_assignment_id="cohort-v1:" + "b" * 64, wake_revision=1, recipient_lookup="c" * 32, From bfa24f66899df588427f7fe273b370f24681284a Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:33:45 -0400 Subject: [PATCH 2/6] Give structural field representation its own behavior and decode whole wake claims --- src/agent_comms/field_codec.py | 118 ++++++++++++++------- src/agent_comms/selected_turn.py | 3 +- src/agent_comms/typed_table.py | 13 ++- tools/cutover/carry_wake_claim_turn_ids.py | 42 ++++---- 4 files changed, 111 insertions(+), 65 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 918aed68b1..17129799ec 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -659,10 +659,8 @@ class DeclaredFieldPlan(FieldPlan): @classmethod def claims(cls, declared: Field, annotation: object) -> bool: - _, representations = FieldCodec.representation(annotation) - return super().claims(declared, annotation) or any( - owner.field_plan is cls for owner in (representations or (FieldRepresentation,)) - ) + _, representation = FieldCodec.representation(annotation) + return super().claims(declared, annotation) or representation.field_plan is cls class TransientFieldPlan(DeclaredFieldPlan): @@ -749,10 +747,10 @@ class InlineFieldPlan(DeclaredFieldPlan): @cached_property def inlined(self) -> type: - declared, representations = FieldCodec.representation(self.annotation) - for representation in representations: - if issubclass(representation, InlineRecord): - return representation.wire_record() + declared, representation = FieldCodec.representation(self.annotation) + match representation: + case type() as inline if issubclass(inline, InlineRecord): + return inline.wire_record() return cast(type, declared) @cached_property @@ -1171,6 +1169,33 @@ class FieldRepresentation[T](MroDispatch, ABC): field_plan: ClassVar[type[FieldPlan]] = DeclaredFieldPlan decoder: ClassVar[type[_RepresentedDecoder]] = _RepresentedDecoder + @classmethod + def for_value(cls, value: object) -> type[FieldRepresentation]: + return cls + + @classmethod + def project(cls, value: object, view: str, annotation: object) -> JsonValue: + return RepresentedProjection.of(value, view, annotation) + + @classmethod + def producer_owners(cls) -> tuple[type[FieldRepresentation], ...]: + return tuple(parent for parent in cls.__mro__ if issubclass(parent, FieldRepresentation)) + + @classmethod + def decoder_for(cls, annotation: object, target: object) -> ValueDecoder[object]: + try: + owner = ValueDecoder.annotation_class(target) + except KeyError: + owner = _ScalarDecoder + return owner.with_representation(annotation, target, get_args(target), cls) + + @classmethod + def schema_for(cls, annotation: object) -> dict[str, JsonValue]: + schema = cls.schema() + if type(None) in get_args(annotation): + return FieldCodec.alternatives_schema("anyOf", [schema, {"type": "null"}]) + return schema + @classmethod def wire_structure(cls, structural: DecoderPlan) -> DecoderPlan: """A custom scalar form stays opaque; declared structural forms supply their plan.""" @@ -1210,6 +1235,30 @@ def argument_options(cls, argument: Argument) -> dict[str, object]: return {} +class StructuralRepresentation(FieldRepresentation): + """The value has no representation codec, so FieldCodec handles its structure.""" + + @classmethod + def for_value(cls, value: object) -> type[FieldRepresentation]: + return FieldCodec.representation(type(value))[1] + + @classmethod + def project(cls, value: object, view: str, annotation: object) -> JsonValue: + return FieldCodec._project_structure(value, view, annotation) + + @classmethod + def producer_owners(cls) -> tuple[type[FieldRepresentation], ...]: + return () + + @classmethod + def decoder_for(cls, annotation: object, target: object) -> ValueDecoder[object]: + return _structural_decoder(target) + + @classmethod + def schema_for(cls, annotation: object) -> dict[str, JsonValue]: + return FieldCodec._structure_schema(annotation) + + class AcceptsNull[T](FieldRepresentation[T]): """Null is this representation's input, not the structural target's.""" @@ -1888,36 +1937,38 @@ def external_decoder(cls, target: type[ExternalField]) -> ExternalDecoder: @classmethod def representation( cls, annotation: object, - ) -> tuple[object, tuple[type[FieldRepresentation], ...]]: - """The value type and its zero or one declared wire representation classes.""" + ) -> tuple[object, type[FieldRepresentation]]: + """The value type and its declared representation, structural when unencoded.""" if annotation is None: - return None, () + return None, StructuralRepresentation if annotation is object: - return object, (_JsonValueRepresentation,) + return object, _JsonValueRepresentation if get_origin(annotation) is Annotated: target, *metadata = get_args(annotation) else: target, metadata = annotation, () - representations = tuple( + representations = [ item for item in metadata if isinstance(item, type) and issubclass(item, FieldRepresentation) - ) + ] if len(representations) > 1: raise TypeError("A field must have one representation") if isinstance(target, TypeAliasType): if target is JsonValue: - return object, representations or (_JsonValueRepresentation,) + if representations: + return object, representations[0] + return object, _JsonValueRepresentation if representations: - return target, representations + return target, representations[0] if target is object: - return object, (_JsonValueRepresentation,) + return object, _JsonValueRepresentation if isinstance(target, type) and issubclass(target, (NominalText, WireValue)): - return target, (target.representation(),) - return target, () + return target, target.representation() + return target, StructuralRepresentation @classmethod def _value_representation(cls, value: object, annotation: object): - return cls.representation(annotation)[1] or cls.representation(type(value))[1] + return cls.representation(annotation)[1].for_value(value) @classmethod def record_fields(cls, record: type) -> tuple[tuple[Field[object], str], ...]: @@ -2138,12 +2189,11 @@ def _record_schema( @classmethod def value_schema(cls, annotation: object) -> dict[str, JsonValue]: """Derive external scalar/array choices from the same decode declarations.""" - annotation, representations = cls.representation(annotation) - for representation in representations: - schema = representation.schema() - if type(None) in get_args(annotation): - return cls.alternatives_schema("anyOf", [schema, {"type": "null"}]) - return schema + annotation, representation = cls.representation(annotation) + return representation.schema_for(annotation) + + @classmethod + def _structure_schema(cls, annotation: object) -> dict[str, JsonValue]: origin, args = get_origin(annotation), get_args(annotation) if origin in (Union, types.UnionType): members = list(args) @@ -2187,8 +2237,10 @@ def project(cls, value: object, view: str, annotation: object = None) -> JsonVal Projection is deliberately one way. Excluded secrets cannot be rebuilt from a read-only view, and a view never serves as a persistence record. """ - if cls._value_representation(value, annotation): - return RepresentedProjection.of(value, view, annotation) + return cls._value_representation(value, annotation).project(value, view, annotation) + + @classmethod + def _project_structure(cls, value: object, view: str, annotation: object) -> JsonValue: if is_dataclass(value) and not isinstance(value, type): return cls._project_record(value, view) if isinstance(value, (tuple, list)): @@ -3166,14 +3218,8 @@ def compile_decoder(annotation: object) -> ValueDecoder[object]: ... @lru_cache(maxsize=None) def compile_decoder(annotation: object) -> ValueDecoder[object]: """Compile each immutable declaration once; value paths never parse it again.""" - declared, representations = FieldCodec.representation(annotation) - for representation in representations: - try: - owner = ValueDecoder.annotation_class(declared) - except KeyError: - owner = _ScalarDecoder - return owner.with_representation(annotation, declared, get_args(declared), representation) - return _structural_decoder(declared) + declared, representation = FieldCodec.representation(annotation) + return representation.decoder_for(annotation, declared) def _structural_decoder(annotation: object) -> ValueDecoder[object]: diff --git a/src/agent_comms/selected_turn.py b/src/agent_comms/selected_turn.py index 20993af9ab..140e14d961 100644 --- a/src/agent_comms/selected_turn.py +++ b/src/agent_comms/selected_turn.py @@ -16,6 +16,7 @@ from .coordination_tables.responses import ResponseObligation from .durable_turn import DurableTurn from .envelope_claim_transitions import WakeClaim +from .field_codec import FieldCodec, MessageId from .optional_awareness_projection import OptionalAwarenessProjection from .native_failures import NativePiUnavailable from .owner_fence import prepare_fence_token @@ -151,7 +152,7 @@ def tool_owner(self, session, input_id, action) -> SelectedCodingToolOwner: claim_scope = WakeClaim( wire_root_id=participant.root_id, source_seq=assignment.wire_seq, - source_message_id=assignment.message_id, + source_message_id=FieldCodec.decode(MessageId, assignment.message_id), wake_assignment_id=assignment.assignment_id, wake_revision=assignment.revision, recipient_lookup=participant.lookup, diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index bf0e39cbbc..725c731be5 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -185,8 +185,8 @@ class TextStorage(SqlStorage): def accepts(cls, annotation: object) -> bool: if annotation is str: return True - _, representations = FieldCodec.representation(annotation) - return any(issubclass(owner, TextRepresentation) for owner in representations) + _, representation = FieldCodec.representation(annotation) + return issubclass(representation, TextRepresentation) class IntegerStorage(SqlStorage): @@ -2112,15 +2112,14 @@ def producer_owners(cls) -> tuple[type, ...]: [column.annotation for column in cls._columns] if is_dataclass(cls) else [] ) while annotations: - value, representations = FieldCodec.representation(annotations.pop()) + value, representation = FieldCodec.representation(annotations.pop()) annotations.extend(get_args(value)) - for owner in (value, *representations): - if not isinstance(owner, type): - continue + if isinstance(value, type): owners.update( - parent for parent in owner.__mro__ + parent for parent in value.__mro__ if issubclass(parent, (FieldRepresentation, NominalText)) ) + owners.update(representation.producer_owners()) return tuple(sorted(owners, key=lambda owner: (owner.__module__, owner.__qualname__))) diff --git a/tools/cutover/carry_wake_claim_turn_ids.py b/tools/cutover/carry_wake_claim_turn_ids.py index 8fa61792f9..f0ed7f1374 100644 --- a/tools/cutover/carry_wake_claim_turn_ids.py +++ b/tools/cutover/carry_wake_claim_turn_ids.py @@ -1,6 +1,5 @@ """Validate version-one wake turn identities before stamping their narrowed format.""" -import sys from contextlib import closing from functools import partial from pathlib import Path @@ -8,7 +7,7 @@ from agent_comms.bus import Bus from agent_comms.envelope_claim_transitions import WakeClaim -from agent_comms.field_codec import FieldCodec, TurnId +from agent_comms.field_codec import FieldCodec, MalformedRecord from agent_comms.historical_views import HistoryArchive from agent_comms.installer.probes.carry import ArchivedInputs, Carried, Carry from agent_comms.locked_store import RootFile @@ -27,22 +26,17 @@ def roots(self) -> tuple[Path, ...]: @staticmethod def rewrite(path: Path, record: type, payload: dict[str, object]) -> dict[str, object]: - if record is not WakeClaim or payload["version"] != 1: - return payload - value = payload["turn_id"] try: - turn_id = FieldCodec.decode(TurnId, value) - finally: - # Keep the codec's refusal and traceback; add the old file/value without catching it. - if (error := sys.exception()) is not None: - error.add_note(f"{path}: WakeClaim.turn_id={value!r}") - return { - **payload, "turn_id": FieldCodec.encode(turn_id, TurnId), "version": WakeClaim.version, - } + return FieldCodec.encode( + FieldCodec.decode(WakeClaim, {**payload, "version": WakeClaim.version}) + ) + except MalformedRecord as error: + error.add_note(f"{path}: {payload!r}") + raise def carry(self, *, apply: bool) -> Carried: table = Claimed.table - column, = table._fields_named(("claim_transition",)) + (column,) = table._fields_named(("claim_transition",)) changed = 0 rewritten: list[Path] = [] for path in self.input_files(): @@ -50,20 +44,26 @@ def carry(self, *, apply: bool) -> Carried: with closing(Bus.read_snapshot(path)) as db: if not table.present(db): continue - for seq, stored in db.execute(table._sql( - f'SELECT {{seq}}, {{claim_transition}} FROM "{table.declared_name}"' - )): + for seq, stored in db.execute( + table._sql(f'SELECT {{seq}}, {{claim_transition}} FROM "{table.declared_name}"') + ): original = column.storage.from_sql(stored) current = column.value.rewrite_document( - original, partial(self.rewrite, path), lambda record: record is WakeClaim, + original, + partial(self.rewrite, path), + lambda record: record is WakeClaim, ) if current != original: updates.append((column.storage.to_sql(current), seq)) changed += len(updates) if apply and updates: with closing(Bus.writer(path)) as db, Bus.write_transaction(db): - db.executemany(table._sql( - f'UPDATE "{table.declared_name}" SET {{claim_transition}}=? WHERE {{seq}}=?' - ), updates) + db.executemany( + table._sql( + f'UPDATE "{table.declared_name}" SET ' + "{claim_transition}=? WHERE {seq}=?" + ), + updates, + ) rewritten.append(path.relative_to(self.root)) return Carried(changed, rewritten=tuple(rewritten)) From 386d6259681e0f1b83cba8fb906f7aadf326d54d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:40:50 -0400 Subject: [PATCH 3/6] Type the selected representation before dispatching its projection --- src/agent_comms/field_codec.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 17129799ec..d016c0d7e4 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -1967,7 +1967,9 @@ def representation( return target, StructuralRepresentation @classmethod - def _value_representation(cls, value: object, annotation: object): + def _value_representation( + cls, value: object, annotation: object, + ) -> type[FieldRepresentation]: return cls.representation(annotation)[1].for_value(value) @classmethod From 8d2a9adebca141d2f9b1036534b2f2f88e2b832b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:41:57 -0400 Subject: [PATCH 4/6] Resolve represented decoder candidates without swallowed lookup errors --- src/agent_comms/field_codec.py | 41 +++++++++++++--------- tools/cutover/carry_wake_claim_turn_ids.py | 8 +++-- 2 files changed, 29 insertions(+), 20 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index d016c0d7e4..511a0834e1 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -963,18 +963,28 @@ def from_annotation(cls, target: object, args: tuple[object, ...]) -> ValueDecod return cls(target) @classmethod - def annotation_class(cls, annotation: object) -> type[ValueDecoder[object]]: - try: - return cls._by_annotation[annotation] - except KeyError: - pass + def _registered(cls, annotation: object) -> tuple[type[ValueDecoder[object]], ...]: + matches: list[type[ValueDecoder[object]]] = [] + if annotation in cls._by_annotation: + matches.append(cls._by_annotation[annotation]) declarations = annotation.__mro__ if isinstance(annotation, type) else (annotation,) for declaration in declarations: - try: - return cls._by_origin[get_origin(declaration) or declaration] - except KeyError: - continue - raise KeyError(annotation) + origin = get_origin(declaration) or declaration + if origin in cls._by_origin: + matches.append(cls._by_origin[origin]) + return tuple(matches) + + @classmethod + def annotation_class(cls, annotation: object) -> type[ValueDecoder[object]]: + matches = cls._registered(annotation) + if not matches: + raise KeyError(annotation) + return matches[0] + + @classmethod + def represented_class(cls, annotation: object) -> type[ValueDecoder[object]]: + """A represented value with no registered decoder is a scalar of its representation.""" + return next(iter(cls._registered(annotation)), _ScalarDecoder) @classmethod def with_representation( @@ -1174,7 +1184,7 @@ def for_value(cls, value: object) -> type[FieldRepresentation]: return cls @classmethod - def project(cls, value: object, view: str, annotation: object) -> JsonValue: + def project_view(cls, value: object, view: str, annotation: object) -> JsonValue: return RepresentedProjection.of(value, view, annotation) @classmethod @@ -1183,10 +1193,7 @@ def producer_owners(cls) -> tuple[type[FieldRepresentation], ...]: @classmethod def decoder_for(cls, annotation: object, target: object) -> ValueDecoder[object]: - try: - owner = ValueDecoder.annotation_class(target) - except KeyError: - owner = _ScalarDecoder + owner = ValueDecoder.represented_class(target) return owner.with_representation(annotation, target, get_args(target), cls) @classmethod @@ -1243,7 +1250,7 @@ def for_value(cls, value: object) -> type[FieldRepresentation]: return FieldCodec.representation(type(value))[1] @classmethod - def project(cls, value: object, view: str, annotation: object) -> JsonValue: + def project_view(cls, value: object, view: str, annotation: object) -> JsonValue: return FieldCodec._project_structure(value, view, annotation) @classmethod @@ -2239,7 +2246,7 @@ def project(cls, value: object, view: str, annotation: object = None) -> JsonVal Projection is deliberately one way. Excluded secrets cannot be rebuilt from a read-only view, and a view never serves as a persistence record. """ - return cls._value_representation(value, annotation).project(value, view, annotation) + return cls._value_representation(value, annotation).project_view(value, view, annotation) @classmethod def _project_structure(cls, value: object, view: str, annotation: object) -> JsonValue: diff --git a/tools/cutover/carry_wake_claim_turn_ids.py b/tools/cutover/carry_wake_claim_turn_ids.py index f0ed7f1374..e00f514326 100644 --- a/tools/cutover/carry_wake_claim_turn_ids.py +++ b/tools/cutover/carry_wake_claim_turn_ids.py @@ -27,9 +27,11 @@ def roots(self) -> tuple[Path, ...]: @staticmethod def rewrite(path: Path, record: type, payload: dict[str, object]) -> dict[str, object]: try: - return FieldCodec.encode( - FieldCodec.decode(WakeClaim, {**payload, "version": WakeClaim.version}) - ) + return { + **FieldCodec.encode( + FieldCodec.decode(WakeClaim, {**payload, "version": WakeClaim.version}) + ) + } except MalformedRecord as error: error.add_note(f"{path}: {payload!r}") raise From 5c41d99b8f816f0bdc18d342e9b810deb3fd2136 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:45:29 -0400 Subject: [PATCH 5/6] Keep structural schema generation on the structural representation --- src/agent_comms/field_codec.py | 86 +++++++++++++++++++--------------- 1 file changed, 49 insertions(+), 37 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 511a0834e1..f19ef43a2f 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -1263,7 +1263,55 @@ def decoder_for(cls, annotation: object, target: object) -> ValueDecoder[object] @classmethod def schema_for(cls, annotation: object) -> dict[str, JsonValue]: - return FieldCodec._structure_schema(annotation) + origin, args = get_origin(annotation), get_args(annotation) + if origin in (Union, types.UnionType): + members = list(args) + if len(members) == 1: + return FieldCodec.value_schema(members[0]) + return FieldCodec.alternatives_schema( + "anyOf", [FieldCodec.value_schema(item) for item in members], + ) + if origin is Literal: + return { + **FieldCodec.value_schema(type(args[0])), + "enum": [value for _, value in compile_decoder(annotation).editor_choices()], + } + if origin in (list, tuple, frozenset): + return {"type": "array", "items": FieldCodec.value_schema(args[0])} + if origin is dict: + if args[0] is str: + return {"type": "object", "additionalProperties": FieldCodec.value_schema(args[1])} + return { + "type": "array", + "items": { + "type": "array", + "prefixItems": [FieldCodec.value_schema(arg) for arg in args], + "minItems": 2, + "maxItems": 2, + }, + } + if origin is type and args and issubclass(args[0], DeclaredFamily): + return { + "type": "string", + "enum": [value for _, value in compile_decoder(annotation).editor_choices()], + } + if isinstance(annotation, type) and issubclass(annotation, Enum): + values = [value for _, value in compile_decoder(annotation).editor_choices()] + return {**FieldCodec.value_schema(type(values[0])), "enum": values} + primitive = { + str: "string", + bool: "boolean", + int: "integer", + float: "number", + type(None): "null", + } + if annotation in primitive: + return {"type": primitive[annotation]} + if isinstance(annotation, type) and issubclass(annotation, DeclaredFamily): + return FieldCodec._family_schema(annotation) + if isinstance(annotation, type) and is_dataclass(annotation): + return FieldCodec.record_schema(annotation) + raise TypeError(f"No declared JSON schema for {annotation}") class AcceptsNull[T](FieldRepresentation[T]): @@ -2201,42 +2249,6 @@ def value_schema(cls, annotation: object) -> dict[str, JsonValue]: annotation, representation = cls.representation(annotation) return representation.schema_for(annotation) - @classmethod - def _structure_schema(cls, annotation: object) -> dict[str, JsonValue]: - origin, args = get_origin(annotation), get_args(annotation) - if origin in (Union, types.UnionType): - members = list(args) - if len(members) == 1: - return cls.value_schema(members[0]) - return cls.alternatives_schema("anyOf", [cls.value_schema(item) for item in members]) - if origin is Literal: - return {**cls.value_schema(type(args[0])), "enum": [value for _, value in compile_decoder(annotation).editor_choices()]} - if origin in (list, tuple, frozenset): - return {"type": "array", "items": cls.value_schema(args[0])} - if origin is dict: - if args[0] is str: - return {"type": "object", "additionalProperties": cls.value_schema(args[1])} - return {"type": "array", "items": {"type": "array", "prefixItems": [cls.value_schema(arg) for arg in args], "minItems": 2, "maxItems": 2}} - if origin is type and args and issubclass(cast(type, args[0]), DeclaredFamily): - return {"type": "string", "enum": [value for _, value in compile_decoder(annotation).editor_choices()]} - if isinstance(annotation, type) and issubclass(annotation, Enum): - values = [value for _, value in compile_decoder(annotation).editor_choices()] - return {**cls.value_schema(type(values[0])), "enum": values} - primitive = { - str: "string", - bool: "boolean", - int: "integer", - float: "number", - type(None): "null", - } - if annotation in primitive: - return {"type": primitive[annotation]} - if isinstance(annotation, type) and issubclass(annotation, DeclaredFamily): - return cls._family_schema(annotation) - if isinstance(annotation, type) and is_dataclass(annotation): - return cls.record_schema(annotation) - raise TypeError(f"No declared JSON schema for {annotation}") - @classmethod def project(cls, value: object, view: str, annotation: object = None) -> JsonValue: """Encode a redacted view using field exclusions and owned properties. From d0a24e2f7de1330de0f2a157c665ea611b71536f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 21:49:23 -0400 Subject: [PATCH 6/6] Mint the selected-tool owner fixture input ID at its declared kind --- tests/test_selected_tool_broker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py index ae1c25996b..9dc194992c 100644 --- a/tests/test_selected_tool_broker.py +++ b/tests/test_selected_tool_broker.py @@ -18,7 +18,7 @@ from agent_comms.coordinator import Coordination from agent_comms.coordination_database import CoordinationStore from agent_comms.envelope_claim_transitions import WakeClaim -from agent_comms.field_codec import MessageId, TurnId +from agent_comms.field_codec import InputId, MessageId, TurnId from agent_comms.native_tool_call import ( DuplicateToolCallIdentity, OwnerToolConnection, @@ -114,7 +114,7 @@ def test_owner_mode_denies_unreserved_input_before_consuming_ledger(private_comm attempt_ordinal=1, ) with Coordination(str(CoordinationStore.path_in(comms.root))) as store: - input_id = secrets.token_hex(16) + input_id = InputId.mint() mode = broker.selected_tool_mode_for_owner( comms, store, claim_scope, "owner", input_id )