Hash the capability classes that produce bus projections

src/agent_comms/envelope_claim_transitions.py

@@ -437,11 +437,9 @@ class UnclaimedResource(ResourceClaim):
 
 @dataclass(frozen=True)
 class ClaimOwner(AcquiringClaim, ResourceClaim):
+    # Built only by Acquisition.claimed from a validated acquisition and one of its own claims.
     resource: Annotated[str, ResourcePathText]
 
-    def __post_init__(self) -> None:
-        FieldCodec.require_fields(self)
-
     def acquire(
         self, scope: RecordedScope, registry: RegistrySnapshot, publish: Callable[[], ClaimOwner]
     ) -> ClaimOwner:
@@ -465,7 +463,7 @@ class ClaimProjection(Mapping[str, ClaimOwner]):
     """Immutable sorted claim table plus the last processed envelope sequence."""
 
     last_seq: Annotated[int, ProjectionSequence] = 0
-    # Built from a dict, which the codec checks; held afterwards as a read-only view.
+    # Built from a dict by the transitions and the holder tables; held as a read-only view.
     _claims: dict[str, ClaimOwner] = field(default_factory=dict)
 
     def __post_init__(self) -> None:

src/agent_comms/messages.py

@@ -71,11 +71,16 @@ class BusTables(TypedTable, AppendOnlyTables, SharedFileTables, DurableTables, d
 class BusProjectionTables(ProducedTable, SharedFileTables, DerivedTables, declared_name=None):
     """Reverse lookups are derived from the record, in its append transaction."""
 
+    @classmethod
+    def producer_capabilities(cls) -> tuple[tuple[type[DeclaredFamily], type], ...]:
+        return ((BusRecord, BusCapability), (ClaimTransition, ClaimTransition))
+
     @classmethod
     def producer_classes(cls) -> tuple[type[DeclaredFamily], ...]:
-        return (
-            *BusRecord.discover_members(agent_comms, BusCapability),
-            *ClaimTransition.discover_members(agent_comms, ClaimTransition),
+        return tuple(
+            member
+            for family, capability in cls.producer_capabilities()
+            for member in family.discover_members(agent_comms, capability)
         )
 
     @classmethod

src/agent_comms/typed_table.py

@@ -1426,6 +1426,11 @@ class ProjectionProducerTables(SharedFileTables, DerivedTables):
 class ProducedRow(ABC):
     """Rows rebuilt when their declared producing modules change."""
 
+    @classmethod
+    @abstractmethod
+    def producer_capabilities(cls) -> tuple[tuple[type[DeclaredFamily], type], ...]:
+        """Each record family and the capability whose declarations produce these rows."""
+
     @classmethod
     @abstractmethod
     def producer_classes(cls) -> tuple[type[DeclaredFamily], ...]:
@@ -1977,18 +1982,20 @@ def producer_digest(cls) -> str:
     @cache
     def _producer_digest(cls) -> str:
         producers = cls.producer_classes()
-        owners: set[type[DeclaredFamily]] = {
+        owners: set[type] = {
             owner
             for row in (cls, *cls.members_with(cls))
             for owner in row.__mro__
             if issubclass(owner, cls)
         }
+        # The capability's classes outside the record family (Message, Claimed) hold the
+        # production code itself, so they belong to the digest beside the family.
+        capabilities = tuple(capability for _, capability in cls.producer_capabilities())
         for producer in producers:
-            root = producer.family_root()
             owners.update(
                 owner
                 for owner in producer.__mro__
-                if issubclass(owner, cls) or issubclass(owner, root)
+                if issubclass(owner, (cls, producer.family_root(), *capabilities))
             )
         source = "\0".join(
             f"{owner.__module__}.{owner.__qualname__}\0{getsource(owner)}"

tests/test_bus.py

@@ -145,6 +145,16 @@ def test_projection_digest_ignores_infrastructure_source(
             "for record in BusCapability.history(db):",
             "for record in reversed(tuple(BusCapability.history(db))):",
         ),
+        (
+            "messages.py",
+            "yield MessageIdentitySource(self.seq, self.message_id, self.sender)",
+            "yield MessageIdentitySource(self.seq, self.message_id, self.target)",
+        ),
+        (
+            "messages.py",
+            "HeldResources.apply(db, self.claim_transition, projected, holders)",
+            "HeldResources.apply(db, self.claim_transition, projected, holders)\n        del projected",
+        ),
     ),
 )
 def test_projection_digest_tracks_producing_source(

tests/test_envelope_claim_transitions.py

@@ -187,8 +187,6 @@ def test_exact_incarnation_and_generation_fence_releases(tmp_path):
     assert new[a].owner == "peer" and new[a].seq == 3
     with pytest.raises(ClaimTransitionError):
         ClaimProjection(0, {a: new[a]})
-    with pytest.raises(MalformedRecord):
-        ClaimProjection(1, [])
     with pytest.raises(ClaimTransitionError):
         transition(4, releases=(ClaimRelease(a, G1),)).project(new)