diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index fa211330fe..d1b1968402 100644 --- a/src/agent_comms/bus_tables.py +++ b/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,8 +378,7 @@ 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) @@ -381,11 +386,11 @@ def declared_fields(cls, owner: type) -> tuple[Field, ...]: @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)] diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index 196a10fc21..f4f4e43813 100644 --- a/src/agent_comms/messages.py +++ b/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.""" diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 99b27d4bfa..3a6911cdc8 100644 --- a/src/agent_comms/typed_table.py +++ b/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, diff --git a/tests/test_bus_history_seq.py b/tests/test_bus_history_seq.py index 326af9df10..39327ab112 100644 --- a/tests/test_bus_history_seq.py +++ b/tests/test_bus_history_seq.py @@ -11,6 +11,7 @@ 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 @@ -18,7 +19,6 @@ 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