Include declared value and spelling classes in producer digests

src/agent_comms/field_codec.py

@@ -584,7 +584,7 @@ def rewrite_wire(
 
     def schema_fields(self):
         annotation, declared = self.annotation, self.declared
-        base = FieldCodec._representation(annotation)[0]
+        base = FieldCodec.representation(annotation)[0]
         if self.reject_null and get_origin(base) in (Union, types.UnionType):
             members = tuple(item for item in get_args(base) if item is not type(None))
             annotation = members[0] if len(members) == 1 else Union[members]  # noqa: UP007 - runtime tuple
@@ -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)[1] or FieldRepresentation
         return super().claims(declared, annotation) or representation.field_plan is cls
 
 
@@ -747,7 +747,7 @@ class InlineFieldPlan(DeclaredFieldPlan):
 
     @cached_property
     def inlined(self) -> type:
-        declared, representation = FieldCodec._representation(self.annotation)
+        declared, representation = FieldCodec.representation(self.annotation)
         match representation:
             case type() as inline if issubclass(inline, InlineRecord):
                 return inline.wire_record()
@@ -784,7 +784,7 @@ class InlineFamilyPlan(InlineFieldPlan):
 
     @classmethod
     def claims(cls, declared: Field, annotation: object) -> bool:
-        value_type = FieldCodec._representation(annotation)[0]
+        value_type = FieldCodec.representation(annotation)[0]
         return (
             super().claims(declared, annotation)
             and isinstance(value_type, type)
@@ -1883,8 +1883,9 @@ def external_decoders(cls, target: type) -> Iterator[ExternalDecoder]:
     def external_decoder(cls, target: type[ExternalField]) -> ExternalDecoder:
         return target.decode_external
 
-    @staticmethod
-    def _representation(annotation: object) -> tuple[object, type[FieldRepresentation] | None]:
+    @classmethod
+    def representation(cls, annotation: object) -> tuple[object, type[FieldRepresentation] | None]:
+        """The declared value type and the codec owning its wire representation."""
         if annotation is None:
             return None, None
         if annotation is object:
@@ -1913,10 +1914,10 @@ def _representation(annotation: object) -> tuple[object, type[FieldRepresentatio
     @classmethod
     def _value_representation(cls, value: object, annotation: object):
         if annotation is None:
-            return cls._representation(type(value))[1]
-        _, representation = cls._representation(annotation)
+            return cls.representation(type(value))[1]
+        _, representation = cls.representation(annotation)
         if representation is None:
-            _, representation = cls._representation(type(value))
+            _, representation = cls.representation(type(value))
         return representation
 
     @classmethod
@@ -2016,7 +2017,7 @@ def decode_record(
         """Decode declared fields, with nominal narrowing supplied by their contextual owner."""
         decoder = _RecordDecoder(target)
         for name, narrowed in field_types.items():
-            annotation, _ = cls._representation(cls.field_plan(target, name).annotation)
+            annotation, _ = cls.representation(cls.field_plan(target, name).annotation)
             if not issubclass(narrowed, annotation):
                 raise TypeError(f"{target.__name__}.{name} cannot widen its declared type")
         return cast(T, decoder.decode_selected(
@@ -2143,7 +2144,7 @@ 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)
+        annotation, representation = cls.representation(annotation)
         if representation is not None:
             schema = representation.schema()
             if type(None) in get_args(annotation):
@@ -2197,7 +2198,7 @@ def project(cls, value: object, view: str, annotation: object = None) -> JsonVal
         if is_dataclass(value) and not isinstance(value, type):
             return cls._project_record(value, view)
         if isinstance(value, (tuple, list)):
-            arguments = get_args(cls._representation(annotation)[0])
+            arguments = get_args(cls.representation(annotation)[0])
             return [cls.project(item, view, arguments[0] if arguments else None) for item in value]
         return cls.encode(value)
 
@@ -3171,7 +3172,7 @@ 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)
+    declared, representation = FieldCodec.representation(annotation)
     if representation is None:
         return _structural_decoder(declared)
     try:

src/agent_comms/message_reference.py

@@ -7,7 +7,7 @@
 from abc import abstractmethod
 from dataclasses import dataclass
 
-from .field_codec import NullMember, RecordMember, SingleFieldValue, SubclassUnion
+from .field_codec import MessageId, NullMember, RecordMember, SingleFieldValue, SubclassUnion
 from .page_traversal import LiveCursor, NewestCursor, PageCursor, PageTraversal
 
 
@@ -20,7 +20,7 @@ def writes(cls) -> tuple[type[ProducedRow], ...]:
         return (MessageIdentitySource, AnsweredSource)
 
     seq: int
-    message_id: str
+    message_id: MessageId
 
 
 class InboxStart(SubclassUnion, affix="Start"):

src/agent_comms/typed_table.py

@@ -17,7 +17,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence
 from contextlib import AbstractContextManager, ExitStack, closing, contextmanager
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, is_dataclass
 from enum import Enum, IntEnum, IntFlag
 from functools import cache
 from importlib import import_module
@@ -45,8 +45,11 @@
     FieldCodec,
     FieldDirective,
     FieldPlan,
+    FieldRepresentation,
     MalformedRecord,
+    NominalText,
     NullMember,
+    TextRepresentation,
     ValueDecoder,
     WireName,
     WireValue,
@@ -180,7 +183,10 @@ class TextStorage(SqlStorage):
 
     @classmethod
     def accepts(cls, annotation: object) -> bool:
-        return annotation is str
+        if annotation is str:
+            return True
+        _, representation = FieldCodec.representation(annotation)
+        return representation is not None and issubclass(representation, TextRepresentation)
 
 
 class IntegerStorage(SqlStorage):
@@ -2084,7 +2090,7 @@ def _producer_digest(cls) -> str:
 
     @classmethod
     def producer_owners(cls) -> tuple[type, ...]:
-        """Only this table's writers and their inherited implementations.
+        """This table's writers, column values and their inherited implementations.
 
         An ancestor declaring other outputs owns those outputs alone. Inherited
         producer capabilities follow the writer; unrelated mechanisms do not.
@@ -2102,6 +2108,17 @@ 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 []
+        while annotations:
+            value, representation = FieldCodec.representation(annotations.pop())
+            annotations.extend(get_args(value))
+            for owner in (value, representation):
+                if not isinstance(owner, type):
+                    continue
+                owners.update(
+                    parent for parent in owner.__mro__
+                    if issubclass(parent, (FieldRepresentation, NominalText))
+                )
         return tuple(sorted(owners, key=lambda owner: (owner.__module__, owner.__qualname__)))
 
 

tests/test_bus.py

@@ -39,6 +39,7 @@
     RelationViolationError,
     UnclaimedError,
 )
+from agent_comms.field_codec import MessageId
 from agent_comms.message_reference import MessageReference
 from agent_comms.read_basis import RepliesDisplayScope
 from agent_comms.response_acknowledgement import (
@@ -131,6 +132,40 @@ def test_projection_digest_ignores_infrastructure_source(
     assert projection_source_digest(projection_source) == before
 
 
+def test_projection_digest_tracks_declared_column_spelling(projection_source: Path) -> None:
+    before = json.loads(projection_source_digest(projection_source))["message_identity_source"]
+    codec = projection_source / "field_codec.py"
+    original = codec.read_text()
+    related = "digits: ClassVar[int] = 12"
+    assert original.count(related) == 1
+    codec.write_text(original.replace(related, "digits: ClassVar[int] = 13"))
+    assert (
+        json.loads(projection_source_digest(projection_source))["message_identity_source"] != before
+    )
+    unrelated = 'spelling = re.compile("[0-9a-f]{7,40}")'
+    assert original.count(unrelated) == 1
+    codec.write_text(original.replace(unrelated, 'spelling = re.compile("[0-9a-f]{8,40}")'))
+    assert (
+        json.loads(projection_source_digest(projection_source))["message_identity_source"] == before
+    )
+
+
+def test_nominal_message_id_column_stores_plain_text() -> None:
+    identifier = MessageId.mint("nominal column storage")
+    row = MessageIdentitySource(seq=1, message_id=identifier, sender="sender")
+    with sqlite3.connect(":memory:") as db:
+        MessageIdentitySource.create(db)
+        row.insert(db)
+        value, storage = db.execute(
+            f'SELECT message_id, typeof(message_id) FROM "{MessageIdentitySource.declared_name}"',
+        ).fetchone()
+        assert storage == "text"
+        assert value == str(identifier)
+        restored, = MessageIdentitySource.select(db)
+        assert type(restored.message_id) is MessageId
+        assert restored == row
+
+
 @pytest.mark.parametrize(
     ("module", "old", "new"),
     (
@@ -519,7 +554,8 @@ def test_unanswered_reference_selection_requires_the_recorded_original_and_exact
     with pytest.raises(RelationViolationError):
         bus.append(lambda seq, db: PublicRecord.from_message(
             Message("lead", "#comms", "Wrong original reference", InfoKind, seq=seq,
-                    answers=(MessageReference(original.seq + offset, "wrong-hash"),))))
+                    answers=(MessageReference(
+                        original.seq + offset, MessageId.mint("wrong-hash")),))))
     assert bus.revision() == original.seq
     bus.append(lambda seq, db: PublicRecord.from_message(
         Message("another-unit", "#comms", "Someone else's answer", InfoKind, seq=seq,

tests/test_bus_history_seq.py

@@ -15,7 +15,7 @@
 from agent_comms.messages import BusRow, InfoKind, Message
 from agent_comms.comms import wire
 from agent_comms.errors import HumanSendRefusedError
-from agent_comms.field_codec import FieldCodec, MalformedRecord
+from agent_comms.field_codec import FieldCodec, MalformedRecord, MessageId
 from agent_comms.historical_views import HistoryArchive
 from agent_comms.locked_store import RegistryFile
 from agent_comms.message_bus import MessageBus
@@ -104,7 +104,7 @@ def test_message_identity_lookup_preserves_missing_and_ambiguous_refusal(tmp_pat
     original = messages.bus.lookup(1)
     assert messages.bus.lookup_message(original.message_id).reference == original.reference
     with pytest.raises(RowNotFoundError):
-        messages.bus.lookup_message("missing")
+        messages.bus.lookup_message(MessageId.mint("missing"))
     messages.bus.append(lambda seq, db: replace(original, seq=seq))
     with pytest.raises(MalformedRecord, match="Expected one"):
         messages.bus.lookup_message(original.message_id)

tests/test_current_input_origin.py

@@ -14,7 +14,7 @@
 
 from agent_comms.acp_extension import QueuePromptRequest, decode_request, encode_request
 from agent_comms.errors import RelationViolationError
-from agent_comms.field_codec import FieldCodec, TurnId
+from agent_comms.field_codec import FieldCodec, MessageId, TurnId
 from agent_comms.goal_states import ActiveGoal, CompletedGoal, PausedGoal
 from agent_comms.goals import AbsentGoalCheckpoint, Goal, GoalRevision, NoGoal, PresentGoalCheckpoint
 from agent_comms.input_attempt import ReservedInput
@@ -133,7 +133,8 @@ async def test_foreign_or_changed_origin_refused_before_any_reservation(live_own
     invalid = (replace(origin, root_id="f" * 32),
                replace(origin, author=replace(origin.author, created_at=origin.author.created_at + 1)),
                replace(origin, project="/another/project"),
-               WireInputOrigin(origin.root_id, MessageReference(1, "forged-reference")))
+               WireInputOrigin(origin.root_id, MessageReference(
+                   1, MessageId.mint("forged-reference"))))
     for witness in invalid:
         with comms.registry.store.locked(), pytest.raises(RelationViolationError):
             with ExitStack() as ownership:
@@ -166,7 +167,8 @@ async def test_historical_unknown_author_stays_neutral_and_human_scope_invalidat
     assert fact.for_owner(owner, comms.registry.snapshot()) == CurrentHumanInputTaskFact(authored)
     changed = replace(owner, worktree="/another/project")
     assert fact.for_owner(changed, comms.registry.snapshot()) == HumanInputTaskFact(authored)
-    routed = replace(restored, origin=WireInputOrigin(origin.root_id, MessageReference(1, "original-id")))
+    routed = replace(restored, origin=WireInputOrigin(
+        origin.root_id, MessageReference(1, MessageId.mint("original-id"))))
     assert routed.origin.retained_fact(routed) == InputTaskFact(routed)
     assert FieldCodec.decode(ReservedInput, FieldCodec.encode(routed)) == routed
 
@@ -272,8 +274,9 @@ def captured(name):
         comms.messaging.pin_input_constraint('beta', replace(subject, origin=replace(origin, root_id='f'*32)),
                                              worktree=origin.project)
     with pytest.raises(RelationViolationError):
-        comms.messaging.pin_input_constraint('beta', WireProvenance(MessageReference(1, 'not-input')),
-                                             worktree=origin.project)
+        comms.messaging.pin_input_constraint(
+            'beta', WireProvenance(MessageReference(1, MessageId.mint('not-input'))),
+            worktree=origin.project)
     assert comms.bus.bus.revision() == before
     assert inputs.path.read_bytes() == original_inputs
 

tests/test_historical_notifications.py

@@ -19,7 +19,7 @@
 )
 from agent_comms.comms import wire
 from agent_comms.coordination_errors import DMScopeChanged, SchemaVersionError
-from agent_comms.field_codec import TurnId
+from agent_comms.field_codec import MessageId, TurnId
 from agent_comms.read_basis import HistoricalDisplay, ReadCursor
 from agent_comms.read_ledger import ReadLedger
 from agent_comms.thread_identity import AgentThreadRole, UserThreadRole
@@ -192,9 +192,10 @@ def test_sent_recipient_progresses_through_durable_read_and_answer(archived_noti
         activity=read_activity,
         updated_at_ms=1_000,
     )
+    reply_id = MessageId.mint("durable-reply")
     answered_activity = replace(
         read_activity,
-        response=AnsweredResponse(MessageReference(99, "durable-reply"), 3_000),
+        response=AnsweredResponse(MessageReference(99, reply_id), 3_000),
     )
     answered = CompletedAssignment.build(
         FullWake(), "durable-execution", "sender"
@@ -219,7 +220,7 @@ def test_sent_recipient_progresses_through_durable_read_and_answer(archived_noti
         observation=RecordedRecipientActivity(),
         activity=replace(
             sent_activity,
-            response=AnsweredResponse(MessageReference(99, "durable-reply"), 3_000),
+            response=AnsweredResponse(MessageReference(99, reply_id), 3_000),
         ),
         updated_at_ms=2_000,
     )
@@ -230,7 +231,7 @@ def test_sent_recipient_progresses_through_durable_read_and_answer(archived_noti
     assert answered.label.startswith("Answered ") and answered.answered_at_ms == 3_000
     assert delivered.label.startswith("Read ") and delivered.state == ShownToModelProgress(2_000)
     assert delivered_answer.label.startswith("Answered ")
-    assert delivered_answer.answer == MessageReference(99, "durable-reply")
+    assert delivered_answer.answer == MessageReference(99, reply_id)
     assert sent.recipient_identity == read.recipient_identity == answered.recipient_identity
 
 

tests/test_holder_standing_rules.py

@@ -33,7 +33,7 @@
 from agent_comms.service_passes import ServicePassStore
 from agent_comms.coordinator import Coordination
 from agent_comms.errors import CommittedMessageError, HumanSendRefusedError, RelationViolationError
-from agent_comms.field_codec import FieldCodec, TurnId
+from agent_comms.field_codec import FieldCodec, MessageId, TurnId
 from agent_comms.goal_actions import (
     CreateProblemGoalAction,
     ReissuedOrderGoalAction,
@@ -383,7 +383,7 @@ def test_latest_answer_uses_the_complete_original_reference_and_time_cut(held_bu
     publish(held_bus, "lead", "worker", since + 3)
     with pytest.raises(RelationViolationError):
         publish(held_bus, "lead", "worker", since + 4,
-                answers=(MessageReference(question.seq, "wrong-original"),))
+                answers=(MessageReference(question.seq, MessageId.mint("wrong-original")),))
     expired = publish(held_bus, "worker", "lead", since - 1)
     publish(held_bus, "lead", "worker", since + 5, answers=(expired.reference,))
     scope = RepliesDisplayScope(frozenset({"lead"}), frozenset({"worker"}))
@@ -406,7 +406,8 @@ def test_unanswered_pages_match_the_complete_answer_reference(held_bus):
                       for offset in range(105))
     with pytest.raises(RelationViolationError):
         publish(held_bus, "lead", "worker", now,
-                answers=(MessageReference(questions[0].seq, "not-the-original-hash"),))
+                answers=(MessageReference(
+                    questions[0].seq, MessageId.mint("not-the-original-hash")),))
     answered = (questions[1].reference, questions[-1].reference)
     held_bus.messaging.send("lead", "outsider", "Answers retain their original references",
                             answers=answered)

tests/test_input_disposition.py

@@ -7,7 +7,7 @@
 import pytest
 
 from agent_comms.errors import RelationViolationError
-from agent_comms.field_codec import TurnId
+from agent_comms.field_codec import MessageId, TurnId
 from agent_comms.input_attempt import (
     BoundUnknownInput,
     ClearedNativeInput,
@@ -178,7 +178,7 @@ def test_admitted_snapshot_retains_cut_without_blocking_writer(tmp_path):
 
 
 ROOT, NATIVE, SENT = "c" * 32, "a" * 32, "sent text"
-REFERENCE = MessageReference(7, "original-id")
+REFERENCE = MessageReference(7, MessageId.mint("original-id"))
 RESERVED = ReservedInput("bus:7", 7, "owner", 1, "owner", "ask",
                          origin=WireInputOrigin(ROOT, REFERENCE))
 BOUND = RESERVED._transition(

tests/test_observer_death.py

@@ -10,6 +10,7 @@
 from agent_comms.active_route import LocalRoute
 from agent_comms.coordination_errors import StaleRevision
 from agent_comms.errors import RelationViolationError, UnregisteredThreadError
+from agent_comms.field_codec import MessageId
 from agent_comms.thread_identity import RecordedThreadIncarnation
 from agent_comms.owner_connection import OwnerIdentityChangedError
 from agent_comms.message_reference import MessageReference
@@ -63,7 +64,7 @@ async def test_an_invalid_notification_reference_does_not_kill_the_observer(comm
     with ObservationProcess.awaited(comms.route) as process:
         with pytest.raises(RelationViolationError):
             await process.read(ReadNotifications(
-                str(comms.root), (MessageReference(1, "not-the-original"),)
+                str(comms.root), (MessageReference(1, MessageId.mint("not-the-original")),)
             ))
         assert await process.read(ReadNotifications(str(comms.root), ())) == {}