Narrow decoded records from the shape check, and restore Pi field paths

src/agent_comms/field_codec.py

@@ -25,6 +25,7 @@
     Literal,
     Protocol,
     Self,
+    TypeGuard,
     TypeVar,
     Union,
     cast,
@@ -85,6 +86,25 @@ def __exit__(
                 raise self.error_cls(f"{self.subject}: {exc}") from exc
             return False
 
+    class _Located:
+        """Record a refused field under its key so sibling refusals are reported together."""
+
+        __slots__ = ("fix", "name", "problems")
+
+        def __init__(self, problems: dict[str, str], name: str, fix: str) -> None:
+            self.problems = problems
+            self.name = name
+            self.fix = fix
+
+        def __enter__(self) -> None:
+            return None
+
+        def __exit__(self, _exc_type: object, exc: object, _traceback: object) -> bool:
+            if isinstance(exc, MalformedRecord):
+                self.problems.update(exc.located(self.name, self.fix))
+                return True
+            return False
+
     @classmethod
     def refusing(cls, subject: str) -> MalformedRecord._Refusal:
         """Any check that refuses decoded data with ValueError refuses a malformed record.
@@ -466,6 +486,10 @@ def name(self) -> str:
     def key(self) -> str:
         return WireName.read(self.annotation, self.name)
 
+    def locating(self, problems: dict[str, str]) -> MalformedRecord._Located:
+        """A refused value joins its record's problems under this field's key."""
+        return MalformedRecord._Located(problems, self.key, f"supply {self.annotation}")
+
     @cached_property
     def required(self) -> bool:
         return WireRequired.read(self.annotation, False)
@@ -2272,7 +2296,7 @@ def encode(cls, value: object) -> object:
 
 
 def _declared_member[Family: DeclaredFamily](
-    family: type[Family], name: str,
+    family: type[Family], name: object,
 ) -> type[Family]:
     """A wire name the family does not declare is malformed data."""
     try:
@@ -2696,9 +2720,13 @@ class _FamilyRecordDecoder[Member: DeclaredFamily](_ContextDecoder[Member]):
 
     target: type[Member]
 
-    def fits_wire(self, data: object) -> bool:
+    def tagged(self, data: object) -> TypeGuard[dict[str, object]]:
+        """The family form: an object naming its member under the family tag."""
         return isinstance(data, dict) and isinstance(data.get(self.target.family_discriminator), str)
 
+    def fits_wire(self, data: object) -> bool:
+        return self.tagged(data)
+
     @property
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return tuple(_member_decoder(member) for member in self.target.members_with(self.target))
@@ -2732,12 +2760,11 @@ def shape_refusal(self, data: object) -> MalformedRecord:
         return MalformedRecord("Expected a family object.")
 
     def member(self, data: object) -> tuple[type[Member], dict]:
-        if not _FamilyRecordDecoder.fits_wire(self, data):
+        if not self.tagged(data):
             raise _FamilyRecordDecoder.shape_refusal(self, data)
         tag = self.target.family_discriminator
-        payload = cast(dict, data)
-        return _declared_member(self.target, payload[tag]), {
-            key: value for key, value in payload.items() if key != tag
+        return _declared_member(self.target, data[tag]), {
+            key: value for key, value in data.items() if key != tag
         }
 
     def decode_bound(self, data: object, context: Mapping[str, object]) -> Member:
@@ -2754,7 +2781,7 @@ def leaf(self) -> ValueDecoder[object]:
         return _member_decoder(self.target)
 
     def fits_wire(self, data: object) -> bool:
-        return super().fits_wire(data) or self.leaf.fits_wire(data)
+        return self.tagged(data) or self.leaf.fits_wire(data)
 
     def shape_refusal(self, data: object) -> MalformedRecord:
         # The direct leaf form is this document's declared shape.
@@ -2779,7 +2806,7 @@ class _RecordDecoder(_ContextDecoder[Any]):
 
     target: type
 
-    def fits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> TypeGuard[dict[str, object]]:
         if not isinstance(data, dict):
             return False
         plan = FieldCodec._record_plan(self.target)
@@ -2829,10 +2856,10 @@ def shape_refusal(self, data: object) -> MalformedRecord:
             return RecordProblems(self.problems(data))
         return MalformedRecord("Expected a record object; fix: supply a JSON object.")
 
-    def checked(self, data: object) -> dict:
+    def checked(self, data: object) -> dict[str, object]:
         if not self.fits_wire(data):
             raise self.shape_refusal(data)
-        return cast(dict, data)
+        return data
 
     @property
     def field_names(self) -> tuple[str, ...]:
@@ -2852,7 +2879,7 @@ def decode_selected(self, data, selections, *, fields: tuple[str, ...], context=
             if entry.name not in fields:
                 continue
             key = (target, entry.name)
-            try:
+            with entry.locating(problems):
                 if entry.name in field_types:
                     narrowed = compile_decoder(field_types[entry.name])
                     values.update(entry.decoded_using(
@@ -2861,8 +2888,6 @@ def decode_selected(self, data, selections, *, fields: tuple[str, ...], context=
                 else:
                     values.update(entry.decoded_selected(data, selections[key], selections)
                                   if key in selections else entry.decoded_bound(data, context))
-            except MalformedRecord as error:
-                problems.update(error.located(entry.key, f"supply {entry.annotation}"))
         if problems:
             raise RecordProblems(problems)
         return FieldCodec.construct(target, values)

src/agent_comms/pi_payloads.py

@@ -36,6 +36,7 @@
     MalformedRecord,
     NullMember,
     PresentMember,
+    RecordProblems,
     SingleFieldValue,
     SingleExternalMember,
     TextRepresentation,
@@ -209,16 +210,19 @@ def decode_fields(
         cls, value: dict[str, JsonValue], decoded: dict[str, object],
     ) -> dict[str, object]:
         hints = FieldCodec._types(cls)
+        problems: dict[str, str] = {}
         for f, key in FieldCodec.record_fields(cls):
             if f.name in decoded:
                 decoded[f.name] = FieldCodec.decode_resolved(hints[f.name], decoded[f.name])
-            else:
-                plan = FieldCodec.field_plan(cls, f.name)
+                continue
+            plan = FieldCodec.field_plan(cls, f.name)
+            with plan.locating(problems):
                 captured = plan.decoded_external(value)
-                if f.name in captured:
-                    decoded.update(captured)
-                elif plan.constructor_required or plan.required:
+                if f.name not in captured and (plan.constructor_required or plan.required):
                     raise MalformedRecord(f"Missing required field {key}")
+                decoded.update(captured)
+        if problems:
+            raise RecordProblems(problems)
         return decoded
 
     @classmethod

tests/test_native_transcript.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from agent_comms.field_codec import FieldCodec
+from agent_comms.field_codec import FieldCodec, MalformedRecord
 from agent_comms.native_entries import MessageEntry, NativeEntry, TranscriptProjection, UnknownEntry
 from agent_comms.native_transcript import NativeTranscript
 from agent_comms.pi_payloads import UserMessage
@@ -78,9 +78,9 @@ def test_unknown_and_partial_entries_keep_byte_boundaries(tmp_path):
 
 @pytest.mark.parametrize("raw, error, match", [
     (encoded({"type": "message", "timestamp": "2026-10-01T00:00:00.000Z", "message": {"role": "assistant", "content": 42}}),
-     ValueError, r"^message\.content: Expected a JSON array\.; fix: supply tuple\[agent_comms\.pi_payloads\.PiContent, \.\.\.\]$"),
-    (b'{"type":"message"\n', json.JSONDecodeError,
-     r"^Expecting ',' delimiter: line 2 column 1 \(char 18\)$"),
+     ValueError, r"^message\.content: Expected a JSON array\.; fix: supply .*\btuple\[agent_comms\.pi_payloads\.PiContent, \.\.\.\]"),
+    (b'{"type":"message"\n', MalformedRecord,
+     r"^Invalid JSON: Expecting ',' delimiter: line 2 column 1 \(char 18\)$"),
 ])
 @pytest.mark.parametrize("read", [
     lambda reader, size: reader.forward(0, size),
@@ -90,7 +90,7 @@ def test_unknown_and_partial_entries_keep_byte_boundaries(tmp_path):
 def test_corrupt_complete_entries_raise_in_every_read_direction(tmp_path, raw, error, match, read):
     path = tmp_path / "corrupt.jsonl"
     path.write_bytes(raw)
-    with pytest.raises(error):
+    with pytest.raises(error, match=match):
         tuple(read(NativeTranscript(path), path.stat().st_size))