Rebuild bus projections from one joined cursor with carried claim holders

src/agent_comms/bus_tables.py

@@ -3,16 +3,19 @@
 from __future__ import annotations
 
 import sqlite3
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
+from abc import ABC, abstractmethod
 from contextlib import closing
-from dataclasses import Field, dataclass, fields, make_dataclass
+from dataclasses import Field, dataclass, field, fields, make_dataclass
 from functools import cache
 from inspect import get_annotations
+from itertools import accumulate
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, get_origin, get_type_hints
 
 from agent_comms import messages
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.envelope_claim_transitions import (
+    ClaimOwner,
     ClaimProjection,
     ClaimTransition,
     ClaimTransitionError,
@@ -54,9 +57,9 @@ class BusProjectionTables(ProducedRow, SharedFileTables, DerivedTables):
     def create_schema(cls, db: sqlite3.Connection) -> None:
         """Create/rebuild rows from history using the append projection owner."""
         super().create_schema(db)
-        with closing(BusRow.stream(db, order_by=("seq",))) as headers:
-            for header in headers:
-                header.kind.read_from(db, header.seq).project_into(db)
+        holders = ReplayedHolders()
+        for record in BusCapability.history(db):
+            record.project_into(db, holders)
 
 
 class BusSchema(ComposedSchemas, BusTables, ProjectionProducerTables, BusProjectionTables):
@@ -357,17 +360,20 @@ def _project(
         })
 
     @classmethod
-    def project(cls, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+    def project(
+        cls, db: sqlite3.Connection, transition: ClaimTransition, holders: ClaimHolders,
+    ) -> ClaimProjection:
         """Validate before the original is inserted; the caller retains the writer."""
         try:
             transition.require_new_operation(lambda operation: ClaimOperations.used(db, operation))
-            return transition.project(cls.claims_for(db, transition.resources, transition.seq - 1))
+            return transition.project(holders.before(db, transition))
         except (ClaimTransitionError, IdentityConflict) as error:
             raise RelationViolationError("Committed claim transition is inconsistent.") from error
 
     @classmethod
     def apply(
         cls, db: sqlite3.Connection, transition: ClaimTransition, projected: ClaimProjection,
+        holders: ClaimHolders,
     ) -> None:
         removed = transition.resources + transition.expired
         marks = ",".join("?" for _ in removed)
@@ -376,6 +382,47 @@ def apply(
             cls(owner.resource, owner.seq).insert(db)
         for operation_id in transition.operation_ids:
             ClaimOperations(operation_id, transition.seq).insert(db)
+        holders.advance(removed, projected)
+
+
+class ClaimHolders(ABC):
+    """The current holders a claim transition projects from, before it is applied."""
+
+    @abstractmethod
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection: ...
+
+    def advance(self, removed: tuple[str, ...], projected: ClaimProjection) -> None:
+        """Held rows already record the applied transition."""
+
+
+class StoredHolders(ClaimHolders):
+    """One append reads the holders of its resources from their table."""
+
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+        return HeldResources.claims_for(db, transition.resources, transition.seq - 1)
+
+
+@dataclass
+class ReplayedHolders(ClaimHolders):
+    """An in-order rebuild carries the holders it already projected; it never rereads them.
+
+    Each owner was validated in the projection that produced it; only the subset a
+    transition projects from is validated again, as one ``ClaimProjection``.
+    """
+
+    last_seq: int = 0
+    owners: dict[str, ClaimOwner] = field(default_factory=dict)
+
+    def before(self, db: sqlite3.Connection, transition: ClaimTransition) -> ClaimProjection:
+        return ClaimProjection(self.last_seq, {
+            resource: self.owners[resource]
+            for resource in transition.resources if resource in self.owners})
+
+    def advance(self, removed: tuple[str, ...], projected: ClaimProjection) -> None:
+        for resource in removed:
+            self.owners.pop(resource, None)
+        self.owners.update(projected)
+        self.last_seq = projected.last_seq
 
 
 @dataclass(frozen=True)
@@ -438,18 +485,42 @@ def rows(self) -> tuple[CapabilityRow, ...]:
     @classmethod
     def read_from(cls, db: sqlite3.Connection, seq: int) -> Self:
         """Reconstruct this declared value from its own capability tables."""
-        values = {}
+        rows = []
         for table in cls.tables():
-            rows = table.select(db, seq=seq)
-            if not rows:
+            stored = table.select(db, seq=seq)
+            if not stored:
                 raise RelationViolationError(f"Bus row {seq} lacks {table.declared_name}")
-            values.update({item.name: getattr(rows[0], item.name) for item in fields(table)})
-        return FieldCodec.construct(cls, values)
+            rows.append(stored[0])
+        return cls.assembled(rows)
+
+    @classmethod
+    def assembled(cls, rows: Iterable[CapabilityRow]) -> Self:
+        """This declared value from one stored row of each of its capability tables."""
+        return FieldCodec.construct(cls, {
+            item.name: getattr(row, item.name) for row in rows for item in fields(row)})
+
+    @staticmethod
+    def history(db: sqlite3.Connection) -> Iterator[BusCapability]:
+        """Every record in sequence order from one joined cursor, each stored row decoded once."""
+        tables = TypedTable.members_with(CapabilityRow)
+        offsets = dict(zip(tables, accumulate(
+            (len(BusRow.columns()), *(len(table.columns()) for table in tables)))))
+        header = BusRow.qualified("seq")
+        with closing(db.execute(
+            f"SELECT {', '.join(row.qualified(*row.column_projection()) for row in (BusRow, *tables))} "
+            f'FROM "{BusRow.declared_name}" '
+            + " ".join(f'LEFT JOIN "{table.declared_name}" ON {table.qualified("seq")} = {header}'
+                       for table in tables)
+            + f" ORDER BY {header}"
+        )) as cursor:
+            for stored in cursor:
+                kind = BusRow.decode_at(stored, 0).kind
+                yield kind.assembled(table.decode_at(stored, offsets[table]) for table in kind.tables())
 
     def projections(self) -> Iterator[TypedTable]:
         return iter(())
 
-    def project_into(self, db: sqlite3.Connection) -> None:
+    def project_into(self, db: sqlite3.Connection, holders: ClaimHolders) -> None:
         for row in self.projections():
             row.insert(db)
 

src/agent_comms/messages.py

@@ -14,7 +14,14 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, Type, cast, dataclass_transform
 
-from agent_comms.bus_tables import Addressed, BusCapability, BusRow, HeldResources
+from agent_comms.bus_tables import (
+    Addressed,
+    BusCapability,
+    BusRow,
+    ClaimHolders,
+    HeldResources,
+    StoredHolders,
+)
 from agent_comms.response_policy import InformationalPolicy, ResponseEligibility, ResponsePolicy
 from agent_comms.typed_table import Column
 
@@ -576,7 +583,7 @@ def append_to(self, db: sqlite3.Connection, seq: int) -> int:
         stored = replace(self, seq=seq)
         for row in stored.rows():
             row.insert(db)
-        stored.project_into(db)
+        stored.project_into(db, StoredHolders())
         return stored.seq
 
     @property
@@ -614,14 +621,11 @@ class Claimed(BusCapability):
     def require_claim_transition(self) -> ClaimTransition:
         return self.claim_transition
 
-    def replay_claim(self, db: sqlite3.Connection) -> None:
+    def project_into(self, db: sqlite3.Connection, holders: ClaimHolders) -> None:
         """Append and the shared derived rebuild replay this same original transition."""
-        projected = HeldResources.project(db, self.claim_transition)
-        HeldResources.apply(db, self.claim_transition, projected)
-
-    def project_into(self, db: sqlite3.Connection) -> None:
-        super().project_into(db)
-        self.replay_claim(db)
+        super().project_into(db, holders)
+        projected = HeldResources.project(db, self.claim_transition, holders)
+        HeldResources.apply(db, self.claim_transition, projected, holders)
 
 
 @dataclass(frozen=True, kw_only=True)

src/agent_comms/typed_table.py

@@ -1753,6 +1753,18 @@ def stream(
         """Rows decoded as the caller consumes them; close it to release the query."""
         return cls._query(db, where, parameters, order_by, descending, key, row_type, limit)
 
+    @classmethod
+    def qualified(cls, *names: str) -> str:
+        """Columns named through this table, for one select joining tables that share names."""
+        table = _identifier(cls.declared_name)
+        return ", ".join(f"{table}.{_identifier(item.wire_name)}" for item in cls._fields_named(names))
+
+    @classmethod
+    def decode_at(cls, row: tuple[object, ...], offset: int) -> Self:
+        """Decode this table's columns, selected by ``qualified`` in declaration order at ``offset``."""
+        return cls._decode_row(row, {name: offset + index
+                                     for index, name in enumerate(cls.column_projection())})
+
     @classmethod
     def one(cls, db: sqlite3.Connection, **key: object) -> Self | None:
         rows = cls._keyed(db, key)

tests/test_bus_resource_claims.py

@@ -11,7 +11,14 @@
 
 from agent_comms.bus import Bus, ClaimRecord, KeyedPrivateRecord, PublicRecord, RecordBuilder
 from agent_comms.bus_publication import HumanOrigin
-from agent_comms.bus_tables import BusProjectionTables, BusRow, BusTables, ClaimOperations, HeldResources
+from agent_comms.bus_tables import (
+    BusProjectionTables,
+    BusRow,
+    BusTables,
+    ClaimOperations,
+    HeldResources,
+    ReplayedHolders,
+)
 from agent_comms.child_process import NoProcess, ProcessIdentity
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.envelope_claim_transitions import (
@@ -184,8 +191,9 @@ def test_expiry_and_rebuild_use_claimed_replay_not_a_checkpoint(claims: ClaimBus
     with claims.bus.writing() as db:
         HeldResources.delete(db)
         ClaimOperations.delete(db)
+        holders = ReplayedHolders()
         for original in Claimed.table.select(db, order_by=("seq",)):
-            Claimed(claim_transition=original.claim_transition).replay_claim(db)
+            Claimed(claim_transition=original.claim_transition).project_into(db, holders)
     assert claims.bus.claim_projection() == before
     assert claims.counts() == counts