Hash every class that produces projection rows into the projection digest

src/agent_comms/activity_clocks.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING, Self
 
 from .field_codec import FieldCodec
+from .projection_source import ProjectionSource
 from .typed_table import TypedRow
 
 if TYPE_CHECKING:
@@ -14,7 +15,7 @@
 
 
 @dataclass(frozen=True, slots=True)
-class ChannelActivity(TypedRow):
+class ChannelActivity(TypedRow, ProjectionSource):
     last_message: float = 0
     last_user_input: float = 0
 

src/agent_comms/audience_manifest.py

@@ -20,6 +20,7 @@
 from .channel_targets import exact_stored_target
 from .field_codec import FieldCodec, MalformedRecord, Sha256Text
 from .messages import Message
+from .projection_source import ProjectionSource
 
 AUDIENCE_VERSION: Final = 1
 MAX_RECIPIENTS: Final = 4096
@@ -57,7 +58,7 @@ def _digest(domain: bytes, payload: bytes) -> str:
 
 
 @dataclass(frozen=True, slots=True)
-class FrozenRecipient:
+class FrozenRecipient(ProjectionSource):
     """Stable coordination lookup paired with its send-time canonical name."""
 
     recipient_lookup: str
@@ -69,7 +70,7 @@ def __post_init__(self) -> None:
 
 
 @dataclass(frozen=True, slots=True)
-class FrozenAudience:
+class FrozenAudience(ProjectionSource):
     """Immutable precommit snapshot of the full delivered audience, not a acknowledgement.
 
     Neither the digest nor direct construction attests a bus append. At recovery,

src/agent_comms/channel_targets.py

@@ -11,6 +11,7 @@
 from .errors import HumanSendRefusedError, RelationViolationError, UnregisteredThreadError
 from .field_codec import NullMember, SingleFieldValue, SubclassUnion
 from .history_result import DmHistoryPageResult, HistoryPageResult
+from .projection_source import ProjectionSource
 from .response_policy import DirectPolicy, MentionedOnlyPolicy
 from .wake_policy import BoundedTriageWake, FullWake, MessageAudience, UnmentionedWake
 
@@ -103,7 +104,7 @@ def exact_stored_target(name: str) -> bool:
             if name in BuiltinChannelView.addresses() else True)
 
 
-class CommandTarget(ABC):
+class CommandTarget(ABC, ProjectionSource):
     """A command destination owns its binding and catalog participation."""
     needs_channel_state: ClassVar[bool] = False
 
@@ -573,7 +574,7 @@ def bindings[C: CliCommand](
 
 
 @dataclass(frozen=True, slots=True)
-class Tag:
+class Tag(ProjectionSource):
     name: str
 
     def __post_init__(self) -> None:

src/agent_comms/channels.py

@@ -10,6 +10,7 @@
 from typing import TYPE_CHECKING, ClassVar, Annotated
 
 
+from .projection_source import ProjectionSource
 from .channel_targets import ChannelTarget, Tag
 from .cli_result import CliResult
 from .declared_family import DeclaredFamily
@@ -47,7 +48,7 @@ def __missing__(self, tag: str) -> tuple[Thread, ...]:
 
 
 @dataclass(frozen=True, slots=True)
-class Channel(DeclaredFamily, CliResult, ToolResult, affix="Channel"):
+class Channel(DeclaredFamily, CliResult, ToolResult, ProjectionSource, affix="Channel"):
     """A captured channel presentation with behavior supplied by its declaration."""
 
     name: str

src/agent_comms/declared_family.py

@@ -217,12 +217,12 @@ def _member_declarations(cls) -> tuple[type[DeclaredFamily], ...]:
     @classmethod
     def discover_members(cls, package: ModuleType, capability: type) -> tuple[type[Self], ...]:
         """Acquire a producer package before deriving its declared capability members."""
-        DeclaredFamily._discover_package(package)
+        DeclaredFamily.discover_package(package)
         return cls.members_with(capability)
 
     @staticmethod
     @cache
-    def _discover_package(package: ModuleType) -> None:
+    def discover_package(package: ModuleType) -> None:
         """Import each module of the package once, so its members register themselves."""
         for module in pkgutil.iter_modules(package.__path__, package.__name__ + "."):
             if not module.ispkg:

src/agent_comms/envelope_claim_transitions.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING, Annotated, ClassVar, Literal
 
 
+from .projection_source import ProjectionSource
 from .coordination_errors import IdentityConflict
 from .field_codec import (
     BoundedText,
@@ -115,7 +116,7 @@ def validate_leaf(self, path: Path) -> None:
 
 
 @dataclass(frozen=True, slots=True)
-class AssignmentBinding:
+class AssignmentBinding(ProjectionSource):
     """The selected claim's revision and source, used for permission checks and its SQL row."""
 
     assignment_id: Annotated[str, BoundedText]
@@ -135,7 +136,7 @@ def __post_init__(self) -> None:
 
 
 @dataclass(frozen=True)
-class ClaimScope(SubclassUnion, ABC, affix=""):
+class ClaimScope(SubclassUnion, ABC, ProjectionSource, affix=""):
     """What one acquisition belongs to; its member answers reuse, liveness and operation use."""
 
     def reuses(self, other: RecordedScope) -> bool:
@@ -253,7 +254,7 @@ def __post_init__(self) -> None:
 
 
 @dataclass(frozen=True, slots=True)
-class ClaimRelease:
+class ClaimRelease(ProjectionSource):
     resource: Annotated[str, ResourcePathText]
     generation: Annotated[str, Id128Text]
 
@@ -276,7 +277,7 @@ class AcquiringClaim(ClaimIdentity):
 
 
 @dataclass(frozen=True)
-class ClaimTransition(ClaimIdentity, SubclassUnion, ABC, affix=""):
+class ClaimTransition(ClaimIdentity, SubclassUnion, ABC, ProjectionSource, affix=""):
     """One original resource transition; its member declares what it changes."""
 
     max_resources: ClassVar[int] = 32
@@ -459,7 +460,7 @@ def live(self, registry: RegistrySnapshot) -> bool:
 
 
 @dataclass(frozen=True, slots=True)
-class ClaimProjection(Mapping[str, ClaimOwner]):
+class ClaimProjection(Mapping[str, ClaimOwner], ProjectionSource):
     """Immutable sorted claim table plus the last processed envelope sequence."""
 
     last_seq: Annotated[int, ProjectionSequence] = 0

src/agent_comms/messages.py

@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import agent_comms
+from .projection_source import ProjectionSource
 from .turn_origin import TurnOrigin
 import builtins
 import hashlib
 import re
 import sqlite3
 import time
+from types import ModuleType
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from contextlib import closing
@@ -72,16 +74,8 @@ class BusProjectionTables(ProducedTable, SharedFileTables, DerivedTables, declar
     """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 tuple(
-            member
-            for family, capability in cls.producer_capabilities()
-            for member in family.discover_members(agent_comms, capability)
-        )
+    def producer_package(cls) -> ModuleType:
+        return agent_comms
 
     @classmethod
     def create_schema(cls, db: sqlite3.Connection) -> None:
@@ -204,7 +198,7 @@ def apply(
         holders.advance(removed, projected)
 
 
-class ProjectionPass(ABC):
+class ProjectionPass(ABC, ProjectionSource):
     """How one projection pass writes the rows its records derive: an append writes its
     few rows at once."""
 
@@ -274,7 +268,7 @@ class CapabilityRow(BusTables, declared_name=None):
     seq: Annotated[int, Column(primary_key=True, references=TableReference(BusRow, ("seq",)))]
 
 
-class _CapabilityTable:
+class _CapabilityTable(ProjectionSource):
     """A class's table is derived once, after its dataclass decorator has run."""
 
     def __get__(self, instance: BusCapability | None, owner: type[BusCapability]) -> type[CapabilityRow]:
@@ -295,7 +289,7 @@ def for_class(self, owner: type[BusCapability]) -> type[CapabilityRow]:
 
 
 @dataclass(frozen=True)
-class BusCapability(TypedRow):
+class BusCapability(TypedRow, ProjectionSource):
     """The hierarchy and each dataclass declaration own their table union."""
 
     table = _CapabilityTable()
@@ -374,7 +368,7 @@ class Addressed(BusCapability):
     indexes: ClassVar[tuple[Index, ...]] = (Index(("sender", "target", "seq")),)
 
 
-class MessageKind(DeclaredFamily, affix="Kind"):
+class MessageKind(DeclaredFamily, ProjectionSource, affix="Kind"):
     """A declared message kind; each member owns the validation it requires.
 
     Members are the kind themselves, never instances: the wire carries the
@@ -488,7 +482,7 @@ def require_publishable(cls, message: Message, worktree: Path) -> None:
         message.report.require_references_exist(worktree)
 
 
-class MembershipTransition(DeclaredFamily, affix="Membership"):
+class MembershipTransition(DeclaredFamily, ProjectionSource, affix="Membership"):
     @classmethod
     def conversation_role(cls) -> MembershipConversationRole:
         return MembershipConversationRole(cls)
@@ -515,13 +509,13 @@ def applies(previous: bool, current: bool) -> bool:
 
 
 @dataclass(frozen=True, slots=True)
-class NoMembershipChange(NullMember):
+class NoMembershipChange(NullMember, ProjectionSource):
     def conversation_role(self):
         return ConversationRole()
 
 
 @dataclass(frozen=True, slots=True)
-class MessageRoleWire:
+class MessageRoleWire(ProjectionSource):
     """The unchanged external fields, decoded once into a message role."""
 
     change: Annotated[
@@ -536,7 +530,7 @@ def role(self):
         return role.as_notice() if self.notice else role
 
 
-class MessageRole(InlineRecord):
+class MessageRole(InlineRecord, ProjectionSource):
     category: ClassVar[type[MessageCategory]]
 
     @abstractmethod
@@ -576,7 +570,7 @@ class SendableRole(MessageRole, DeclaredFamily, affix="Role"):
     pass
 
 
-class InformationalMessage:
+class InformationalMessage(ProjectionSource):
     def response_policy(self, message):
         return InformationalPolicy.instance()
 

src/agent_comms/projection_source.py

@@ -0,0 +1,25 @@
+"""The declared capability of code that produces derived projection rows."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+
+class ProjectionSource:
+    """A class whose code decides projection row content.
+
+    Its source, and every subclass's source, belongs to the digest that tells a
+    stored projection it is still fresh. A class reached while rows are produced
+    that does not inherit this capability is a hole in that digest.
+    """
+
+    __slots__ = ()
+    _sources: ClassVar[list[type[ProjectionSource]]] = []
+
+    def __init_subclass__(cls, **kwargs: object) -> None:
+        super().__init_subclass__(**kwargs)
+        ProjectionSource._sources.append(cls)
+
+    @staticmethod
+    def sources() -> tuple[type[ProjectionSource], ...]:
+        return tuple(ProjectionSource._sources)

src/agent_comms/read_basis.py

@@ -28,6 +28,7 @@
 from .mentions import ThreadMention
 from .message_reference import MessageReference
 from .private_path import FileObservation, MissingFile
+from .projection_source import ProjectionSource
 from .recorded_bus import RecordedBus, RecordedBusIdentity
 from .store_files import SharedLock
 from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation
@@ -817,7 +818,7 @@ def validate_identity(
         self.displayed.validate(self.viewer, snapshot, bus_identity)
 
 
-class MessageDisplayScope(DeclaredFamily, ABC, affix="DisplayScope"):
+class MessageDisplayScope(DeclaredFamily, ABC, ProjectionSource, affix="DisplayScope"):
     """Captured inclusion and its exact predicate on the bus capability tables."""
 
     def read_key(self, sender: str, target: str, snapshot: RegistryProvenance) -> Conversation:

src/agent_comms/response_policy.py

@@ -9,14 +9,15 @@
 from typing import TYPE_CHECKING
 
 from .declared_family import DeclaredFamily
+from .projection_source import ProjectionSource
 
 if TYPE_CHECKING:
     from .messages import Message
     from .threads import Thread
 
 
 @dataclass(frozen=True)
-class ResponsePolicy(DeclaredFamily, affix="Policy"):
+class ResponsePolicy(DeclaredFamily, ProjectionSource, affix="Policy"):
     starts_turn = False
     separate_turn = False
 
@@ -52,7 +53,7 @@ def disposition_key(self, message: Message, owner: Thread) -> str:
         return f"bus:{message.seq}:owner:{float(owner.created_at).hex()}"
 
 
-class NoChannelRecipients:
+class NoChannelRecipients(ProjectionSource):
     def recipients(
         self, message: Message, audience: Sequence[str], *, aliases: Mapping[str, str] | None = None
     ) -> tuple[str, ...]:
@@ -119,7 +120,7 @@ def guidance(self, message: Message, *, aliases: Mapping[str, str] | None = None
 
 
 @dataclass(frozen=True, slots=True)
-class ResponseEligibility:
+class ResponseEligibility(ProjectionSource):
     policy: ResponsePolicy
     recipients: tuple[str, ...] = ()
 

src/agent_comms/task_sources.py

@@ -13,6 +13,7 @@
 from .goals import GoalDeclaration, GoalRevision
 from .message_reference import MessageReference
 from .owner_requests import OwnerRequestId, RequestDirective
+from .projection_source import ProjectionSource
 from .thread_identity import (
     RecordedThreadIncarnation,
     ThreadIncarnation,
@@ -185,7 +186,7 @@ def root_source(self, message: Message, originals: dict[MessageReference, Messag
             raise RelationViolationError("Retained correction lacks its original source") from error
 
 
-class TaskAttachment(DeclaredFamily, affix="TaskAttachment"):
+class TaskAttachment(DeclaredFamily, ProjectionSource, affix="TaskAttachment"):
     """Absent attachments have no publication, fact or author effects."""
 
     retains_authored_task = False

src/agent_comms/thread_identity.py

@@ -16,6 +16,7 @@
 
 
 from agent_comms.participant_identity import ParticipantIdentity
+from .projection_source import ProjectionSource
 from .cli_options import JsonFlagRecord
 from .declared_family import DeclaredFamily, UniqueExternalSelector, UnknownFamilyMember, declared_words
 from .errors import RelationViolationError
@@ -421,7 +422,7 @@ def decode(cls, name: str, birth: float) -> ThreadIncarnation:
 
 
 @dataclass(frozen=True, slots=True)
-class TurnId:
+class TurnId(ProjectionSource):
     value: str
 
     @classmethod
@@ -525,7 +526,7 @@ def __eq__(self, other: object) -> bool:
 
 
 @dataclass(slots=True)
-class GenerationCounter:
+class GenerationCounter(ProjectionSource):
     """One monotonic allocation domain, including deleted-name tombstones."""
 
     counter: int = 0
@@ -558,7 +559,7 @@ def rename(self, old: str, new: str) -> None:
         self.generations[new] = self.generations.pop(old)
 
 
-class ThreadRole(DeclaredFamily, affix="ThreadRole"):
+class ThreadRole(DeclaredFamily, ProjectionSource, affix="ThreadRole"):
     """Whether a thread executes turns is the member's type; its wire name is its spelling."""
 
     origin_label: ClassVar[str]