diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index e71f1caac1..29dce0497c 100644 --- a/src/agent_comms/bus_tables.py +++ b/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) diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index b7722c205d..9aff478d3a 100644 --- a/src/agent_comms/messages.py +++ b/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) diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index b08232fbf5..10454db5f1 100644 --- a/src/agent_comms/typed_table.py +++ b/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) diff --git a/tests/test_bus_resource_claims.py b/tests/test_bus_resource_claims.py index 0d4c13bf10..17589592ee 100644 --- a/tests/test_bus_resource_claims.py +++ b/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