Let structural field representations own their behavior

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]

src/agent_comms/field_codec.py

@@ -659,7 +659,7 @@ class DeclaredFieldPlan(FieldPlan):
 
     @classmethod
     def claims(cls, declared: Field, annotation: object) -> bool:
-        representation = FieldCodec.representation(annotation)[1] or FieldRepresentation
+        _, representation = FieldCodec.representation(annotation)
         return super().claims(declared, annotation) or representation.field_plan is cls
 
 
@@ -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(
@@ -1169,6 +1179,30 @@ 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_view(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]:
+        owner = ValueDecoder.represented_class(target)
+        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."""
@@ -1208,6 +1242,78 @@ 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_view(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]:
+        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]):
     """Null is this representation's input, not the structural target's."""
 
@@ -1884,10 +1990,12 @@ 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, type[FieldRepresentation]]:
+        """The value type and its declared representation, structural when unencoded."""
         if annotation is None:
-            return None, None
+            return None, StructuralRepresentation
         if annotation is object:
             return object, _JsonValueRepresentation
         if get_origin(annotation) is Annotated:
@@ -1902,23 +2010,22 @@ def representation(cls, annotation: object) -> tuple[object, type[FieldRepresent
             raise TypeError("A field must have one representation")
         if isinstance(target, TypeAliasType):
             if target is JsonValue:
-                return object, representations[0] if representations else _JsonValueRepresentation
+                if representations:
+                    return object, representations[0]
+                return object, _JsonValueRepresentation
         if representations:
             return target, representations[0]
         if target is object:
             return object, _JsonValueRepresentation
         if isinstance(target, type) and issubclass(target, (NominalText, WireValue)):
             return target, target.representation()
-        return target, None
+        return target, StructuralRepresentation
 
     @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
+    def _value_representation(
+        cls, value: object, annotation: object,
+    ) -> type[FieldRepresentation]:
+        return cls.representation(annotation)[1].for_value(value)
 
     @classmethod
     def record_fields(cls, record: type) -> tuple[tuple[Field[object], str], ...]:
@@ -2064,11 +2171,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)
@@ -2145,44 +2247,7 @@ def _record_schema(
     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:
-            schema = representation.schema()
-            if type(None) in get_args(annotation):
-                return cls.alternatives_schema("anyOf", [schema, {"type": "null"}])
-            return schema
-        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}")
+        return representation.schema_for(annotation)
 
     @classmethod
     def project(cls, value: object, view: str, annotation: object = None) -> JsonValue:
@@ -2193,8 +2258,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) is not None:
-            return RepresentedProjection.of(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:
         if is_dataclass(value) and not isinstance(value, type):
             return cls._project_record(value, view)
         if isinstance(value, (tuple, list)):
@@ -3173,13 +3240,7 @@ def compile_decoder(annotation: object) -> ValueDecoder[object]: ...
 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)
+    return representation.decoder_for(annotation, declared)
 
 
 def _structural_decoder(annotation: object) -> ValueDecoder[object]:

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,

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:

src/agent_comms/typed_table.py

@@ -186,7 +186,7 @@ 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)
+        return issubclass(representation, TextRepresentation)
 
 
 class IntegerStorage(SqlStorage):
@@ -2108,17 +2108,18 @@ 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())
             annotations.extend(get_args(value))
-            for owner in (value, representation):
-                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__)))