Keep bus append and capability queries with their declared owners

src/agent_comms/bus_tables.py

@@ -3,9 +3,9 @@
 from __future__ import annotations
 
 import sqlite3
-import time
+from collections.abc import Iterator
 from contextlib import closing
-from dataclasses import Field, dataclass, fields, make_dataclass, replace
+from dataclasses import Field, dataclass, fields, make_dataclass
 from functools import cache
 from inspect import get_annotations
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, get_origin, get_type_hints
@@ -285,7 +285,7 @@ def references(cls) -> tuple[ForeignKey, ...]:
     def owner_for(cls, db: sqlite3.Connection, resource: str) -> ResourceClaim:
         originals = messages.Claimed.table.select(
             db, where=f'{{seq}}=(SELECT claim_seq FROM "{cls.declared_name}" WHERE resource=?)',
-            parameters=(resource,),
+            parameters=(resource,), row_type=messages.Claimed,
         )
         return next((row.claim_transition.claimed(resource) for row in originals), UnclaimedResource())
 
@@ -312,8 +312,8 @@ def _project(
         ), (through,)).fetchone()[0])
         sequences = tuple({row.claim_seq for row in current})
         marks = ",".join("?" for _ in sequences)
-        transitions = {row.seq: row.claim_transition for row in claim_table.select(
-            db, where=f'{{seq}} IN ({marks})', parameters=sequences,
+        transitions = {row.claim_transition.seq: row.claim_transition for row in claim_table.select(
+            db, where=f'{{seq}} IN ({marks})', parameters=sequences, row_type=messages.Claimed,
         )}
         return ClaimProjection(last_seq, {
             row.resource: transitions[row.claim_seq].claimed(row.resource) for row in current
@@ -341,22 +341,28 @@ def apply(
             ClaimOperations(operation_id, transition.seq).insert(db)
 
 
+@dataclass(frozen=True)
+class CapabilityRow(TypedTable, BusTables, declared_name=None):
+    """Every generated capability row belongs to one original bus sequence."""
+
+    seq: Annotated[int, Column(primary_key=True, references=(BusRow, "seq"))]
+
+
 class _CapabilityTable:
     """A class's table is derived once, after its dataclass decorator has run."""
 
-    def __get__(self, instance, owner) -> type[TypedTable]:
+    def __get__(self, instance: BusCapability | None, owner: type[BusCapability]) -> type[CapabilityRow]:
         return self.for_class(owner)
 
     @cache
-    def for_class(self, owner) -> type[TypedTable]:
+    def for_class(self, owner: type[BusCapability]) -> type[CapabilityRow]:
         hints = get_type_hints(owner, include_extras=True)
         return make_dataclass(
             BusTables.__name__.removesuffix("Tables") + owner.__name__ + "Row",
-            [("seq", Annotated[int, Column(primary_key=True, references=(BusRow, "seq"))]),
-             *((item.name, hints[item.name], item) for item in BusCapability.declared_fields(owner)),
+            [*((item.name, hints[item.name], item) for item in BusCapability.declared_fields(owner)),
              *((name, hints[name], vars(owner)[name]) for name in get_annotations(owner)
                if get_origin(hints[name]) is ClassVar and name in vars(owner))],
-            bases=(TypedTable, BusTables),
+            bases=(CapabilityRow,),
             namespace={"__module__": owner.__module__},
             frozen=True,
         )
@@ -372,20 +378,19 @@ def deliveries(self) -> tuple[Delivered, ...]:
         return ()
 
     @classmethod
-    @cache
-    def declared_fields(cls, owner: type) -> tuple[Field, ...]:
+    def declared_fields(cls, owner: type[BusCapability]) -> tuple[Field, ...]:
         own = get_annotations(owner)
         header_fields = {item.name for item in fields(BusRow)}
         return tuple(item for item in fields(owner)
                      if item.name in own and item.name not in header_fields)
 
     @classmethod
     @cache
-    def tables(cls) -> tuple[type[TypedTable], ...]:
+    def tables(cls) -> tuple[type[CapabilityRow], ...]:
         return tuple(base.table for base in cls.__mro__
                      if issubclass(base, BusCapability) and BusCapability.declared_fields(base))
 
-    def rows(self) -> tuple[TypedTable, ...]:
+    def rows(self) -> tuple[CapabilityRow, ...]:
         """Project declared Python fields without flattening their wire representations."""
         return tuple(
             FieldCodec.construct(table, {
@@ -404,23 +409,13 @@ def read_from(cls, db: sqlite3.Connection, seq: int) -> Self:
             values.update({item.name: getattr(rows[0], item.name) for item in fields(table)})
         return FieldCodec.construct(cls, values)
 
-    def projections(self):
+    def projections(self) -> Iterator[TypedTable]:
         return iter(())
 
     def project_into(self, db: sqlite3.Connection) -> None:
         for row in self.projections():
             row.insert(db)
 
-    def append_to(self, db: sqlite3.Connection, seq: int) -> int:
-        """The caller's transaction owns the sequence, including preserved history."""
-        BusRow.insert_values(db, seq=seq, kind=type(self), at=time.time())
-        stored = replace(self, seq=seq)
-        for row in stored.rows():
-            row.insert(db)
-        stored.project_into(db)
-        return stored.seq
-
-
 @dataclass(frozen=True)
 class Addressed(BusCapability):
     sender: Annotated[str, WireName("from"), WireOrder(2)]

src/agent_comms/messages.py

@@ -14,7 +14,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, Type, cast, dataclass_transform
 
-from agent_comms.bus_tables import Addressed, BusCapability, HeldResources
+from agent_comms.bus_tables import Addressed, BusCapability, BusRow, HeldResources
 from agent_comms.response_policy import InformationalPolicy, ResponseEligibility, ResponsePolicy
 from agent_comms.typed_table import Column
 
@@ -572,6 +572,15 @@ def reply_target(self) -> str | None:
 class BusRecord(Message, DeclaredFamily, affix="Record"):
     """Every record declaration owns its frozen value and derived tables."""
 
+    def append_to(self, db: sqlite3.Connection, seq: int) -> int:
+        """The caller's transaction owns the sequence, including preserved history."""
+        BusRow.insert_values(db, seq=seq, kind=type(self), at=time.time())
+        stored = replace(self, seq=seq)
+        for row in stored.rows():
+            row.insert(db)
+        stored.project_into(db)
+        return stored.seq
+
     @property
     def publication_snapshot(self) -> Message:
         """Delivery capabilities are not message content in a publication intent."""

src/agent_comms/typed_table.py

@@ -15,7 +15,7 @@
 import sqlite3
 import types
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Generator, Iterable, Iterator, Mapping
+from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence
 from contextlib import AbstractContextManager, ExitStack, closing, contextmanager
 from dataclasses import dataclass, field
 from enum import Enum, IntEnum, IntFlag
@@ -33,6 +33,7 @@
     cast,
     get_args,
     get_origin,
+    overload,
 )
 
 from .declared_family import DeclaredFamily, declared_words
@@ -50,8 +51,15 @@
 )
 from .lifecycle import LifecycleState
 from .locked_store import (
-    DurableFile, FileStore, PublishedSQLiteFile, RegistryFile, RootFile, RuntimeFile,
-    SQLiteFile, StoreBusy, StoreCommittedError,
+    DurableFile,
+    FileStore,
+    PublishedSQLiteFile,
+    RegistryFile,
+    RootFile,
+    RuntimeFile,
+    SQLiteFile,
+    StoreBusy,
+    StoreCommittedError,
 )
 from .maintenance_barrier import HOLDER_NONCE, MaintenanceBarrier
 from .store_files import _store_lock
@@ -1649,6 +1657,22 @@ def _query(
             parameters + ((limit,) if limit is not None else ()),
         ))
 
+    @overload
+    @classmethod
+    def select(
+        cls, db: sqlite3.Connection, *, row_type: type[RelatedRow], where: str = "1",
+        parameters: tuple = (), order_by: tuple[str, ...] = (), descending: bool = False,
+        limit: int | None = None, **key: object,
+    ) -> list[RelatedRow]: ...
+
+    @overload
+    @classmethod
+    def select(
+        cls, db: sqlite3.Connection, *, row_type: None = None, where: str = "1",
+        parameters: tuple = (), order_by: tuple[str, ...] = (), descending: bool = False,
+        limit: int | None = None, **key: object,
+    ) -> list[Self]: ...
+
     @classmethod
     def select(
         cls,
@@ -1660,9 +1684,25 @@ def select(
         descending: bool = False,
         row_type: type[RelatedRow] | None = None, limit: int | None = None,
         **key: object,
-    ) -> list[Self | RelatedRow]:
+    ) -> Sequence[Self | RelatedRow]:
         return list(cls._query(db, where, parameters, order_by, descending, key, row_type, limit))
 
+    @overload
+    @classmethod
+    def stream(
+        cls, db: sqlite3.Connection, *, row_type: type[RelatedRow], where: str = "1",
+        parameters: tuple = (), order_by: tuple[str, ...] = (), descending: bool = False,
+        limit: int | None = None, **key: object,
+    ) -> Generator[RelatedRow, None, None]: ...
+
+    @overload
+    @classmethod
+    def stream(
+        cls, db: sqlite3.Connection, *, row_type: None = None, where: str = "1",
+        parameters: tuple = (), order_by: tuple[str, ...] = (), descending: bool = False,
+        limit: int | None = None, **key: object,
+    ) -> Generator[Self, None, None]: ...
+
     @classmethod
     def stream(
         cls,

tests/test_bus_history_seq.py

@@ -11,14 +11,14 @@
 from agent_comms.bus import Bus, BusSnapshot, PublicRecord
 from agent_comms.bus_publication import AgentOrigin, HumanOrigin
 from agent_comms.bus_tables import BusRow, MessageIdentitySource
+from agent_comms.comms import Comms
 from agent_comms.errors import HumanSendRefusedError
 from agent_comms.field_codec import FieldCodec, MalformedRecord
 from agent_comms.historical_views import HistoryArchive
 from agent_comms.locked_store import RegistryFile
 from agent_comms.message_bus import MessageBus
 from agent_comms.message_page import MessagePageRequest, PageBounds
 from agent_comms.messages import InfoKind, Message
-from agent_comms.messaging import Messaging
 from agent_comms.native_transcript import NativeTranscript
 from agent_comms.page_traversal import LiveCursor, OlderTraversal
 from agent_comms.presentation import WireRevision
@@ -131,8 +131,7 @@ def test_origins_check_message_identity_inside_actual_writer_transaction(tmp_pat
 def test_user_identity_is_created_once_under_the_registry_owner(tmp_path):
     messages = channel_bus(tmp_path / "source")
     root = messages.bus.path.parent
-    clients = tuple(Messaging(root, Registration(RegistryFile.path_in(root)), messages)
-                    for _ in range(2))
+    clients = tuple(Comms(root).messaging for _ in range(2))
     with ThreadPoolExecutor(max_workers=2) as executor:
         users = tuple(executor.map(lambda client: client.user_identity(str(root)), clients))
     assert users[0].incarnation == users[1].incarnation