Integration batch 2026-10-30: #3697 #3737 #3616 #3595 #3723 #3607

pyproject.toml

@@ -64,6 +64,7 @@ agent-comms-commander = "agent_comms.commander.cli:main"
 agent-comms-integrate = "agent_comms.integration:main"
 agent-comms-commander-skill = "agent_comms_hooks.skill:main"
 agent-comms-commander-reread = "agent_comms_hooks.reread:main"
+agent-comms-prime = "agent_comms_hooks.prime:main"
 agent-comms-commander-deferral-guard = "agent_comms_hooks.deferral:main"
 agent-comms-commander-bounded-waits = "agent_comms_hooks.bounded_waits:main"
 agent-comms-read-gate = "agent_comms_hooks.read_gate:main"

src/agent_comms/attempt_recovery.py

@@ -191,7 +191,7 @@ def require_session_exit(self, session_files: tuple[Path, ...]) -> None:
         if not self.issued_for(self._store):
             raise RecoveryBlocked("native exit observation requires acquired owner loss")
         session_dir = (
-            self._store.session.path.parent / "native-sessions" / self.release.incarnation().lookup
+            self._store.session.path.parent / "native-sessions" / self.release.incarnation.lookup
         )
         proc = Path("/proc")
         if not proc.is_dir():

src/agent_comms/bus.py

@@ -1,3 +1,3 @@
 """Append-only database history and capability-declared record queries."""
 
 from __future__ import annotations
@@ -38,6 +38,7 @@
     Claimed,
     ClaimMessage,
     Message,
+    StoredWireHistory,
     MessageKind,
     PublicRecord,
 )
@@ -49,6 +50,7 @@
     ResourceClaim,
 )
 from agent_comms.errors import CommittedMessageError, RelationViolationError, UnclaimedError
+from agent_comms.coordination_errors import CoordinationReadUnavailable
 from agent_comms.field_codec import ClaimId, FieldCodec, MessageId, ThreadLookup
 from agent_comms.locked_store import BusFile, FileStore, SQLiteReader, StoreCommittedError
 from agent_comms.message_reference import MessageReference
@@ -130,13 +132,18 @@
     def advanced(self, source: BusReader, selection: DisplayUnreadSelection) -> DisplayUnread:
         queries = selection.queries()
         through = source.revision()
-        if through == self.through and tuple(self.counted) == queries:
-            return self
         earlier = {query.counting: query for query in self.counted}
-        kept = {query: query for query in queries if query in self.counted}
-        moved = {query: (earlier[query.counting], parts) for query in queries
-                 if query not in kept and query.counting in earlier
-                 and (parts := query.moved(earlier[query.counting]))}
+        reusable = {query: earlier[query.counting] for query in queries
+                    if query.counting in earlier
+                    and query.same_membership(earlier[query.counting])}
+        kept = {query: previous for query, previous in reusable.items()
+                if query == previous}
+        if through == self.through and len(kept) == len(queries):
+            if tuple(self.counted) == queries:
+                return self
+            return DisplayUnread(through, {query: self.counted[kept[query]] for query in queries})
+        moved = {query: (previous, parts) for query, previous in reusable.items()
+                 if query not in kept and (parts := query.moved(previous))}
         progressed = (*kept, *(previous for previous, _ in moved.values()))
         appended = source.count_unread(
             progressed if through != self.through else (), through=through, after=self.through)
@@ -174,6 +181,10 @@
         with self.observing() as db:
             yield BusSnapshot.from_connection(db)
 
+    @abstractmethod
+    def captured_reading(self) -> AbstractContextManager[BusSnapshot]:
+        """Retain a cut independently of ordinary reads on the acquiring thread."""
+
     def revision(self) -> int:
         with self.observing() as db:
             return BusRow.last_seq(db)
@@ -186,10 +197,10 @@
         with self.observing() as db:
             return BusActivityIndex.aggregate(db, predicate, parameters)
 
-    def participant_activity(self, names: frozenset[str]) -> ChannelActivity:
-        marks = ','.join('?' for _ in names)
+    def participant_activity(self, predicate: str,
+                             parameters: tuple[object, ...]) -> ChannelActivity:
         with self.observing() as db:
-            return BusDisplayIndex.aggregate(db, f'participant IN ({marks})', tuple(names))
+            return BusDisplayIndex.aggregate(db, predicate, parameters)
 
     def claim_projection(self) -> ClaimProjection:
         with self.observing() as db:
@@ -253,9 +264,36 @@
 
     def page(self, before: int, limit: int) -> tuple[BusRecord, ...]:
         with self.observing() as db:
-            return BusCapability.at_sequences(db, tuple(row.seq for row in BusRow.select(
-                db, where="{seq} < ?", parameters=(before,),
-                order_by=("seq",), descending=True, limit=limit)))
+            return BusCapability.at_sequences(
+                db, tuple(row.seq for row in self._page_rows(db, before, limit)),
+            )
+
+    def input_page(self, before: int, limit: int) -> tuple[MessageInput, ...]:
+        """The same global sequence window, without attachment/report/answer acquisition."""
+        with self.observing() as db:
+            return tuple(
+                MessageInput.read_from(db, row.seq)
+                for row in self._page_rows(db, before, limit)
+            )
+
+    @staticmethod
+    def _page_rows(db: sqlite3.Connection, before: int, limit: int) -> list[BusRow]:
+        return BusRow.select(
+            db, where="{seq} < ?", parameters=(before,),
+            order_by=("seq",), descending=True, limit=limit,
+        )
+
+    @contextmanager
+    def stored_message_history(
+        self, predicate: str = "1", parameters: tuple[object, ...] = (),
+        *, through: int,
+    ) -> Iterator[StoredWireHistory]:
+        """Stream declared public wire fields from one read snapshot, without Messages."""
+        with self.observing() as db:
+            query = Message.stored_wire_query(db, MessageIdentitySource)
+            query = (f'SELECT seq, timestamp, wire FROM ({query}) '
+                     f'WHERE seq <= ? AND ({predicate})')
+            yield StoredWireHistory(db, query, (through, *parameters))
 
     def message_history(self) -> Iterator[Message]:
         """Public bulk export, paged within one committed SQLite snapshot."""
@@ -288,22 +326,22 @@
         upper: int | None = None, descending: bool = False, limit: int = 100,
         kinds: tuple[type[MessageKind], ...] = (),
     ) -> tuple[Message, ...]:
-        """A captured display scope narrows the addressed table before decoding."""
+        """A captured display scope reads public message capabilities, not delivery payloads."""
         with self.observing() as db:
             upper = BusRow.last_seq(db) + 1 if upper is None else upper
             order = "DESC" if descending else "ASC"
             members = ""
             if kinds:
                 members = MessageInput.render(' AND {type} IN (') + ','.join('?' for _ in kinds) + ')'
-            rows = db.execute(
+            return tuple(Message.selected(
+                db,
                 f'SELECT seq FROM "{Addressed.table.declared_name}" '
                 f'JOIN "{MessageInput.table.declared_name}" USING(seq) '
                 f'WHERE ({predicate}) AND seq > ? AND seq < ?{members} '
                 f'ORDER BY seq {order} LIMIT ?',
                 (*parameters, lower, upper, *(kind.declared_name for kind in kinds), limit),
-            )
-            headers = (BusRow.require(db, seq=FieldCodec.decode(int, row[0])) for row in rows)
-            return BusCapability.at_sequences(db, tuple(header.seq for header in headers))
+                descending=descending,
+            ))
 
     def latest_conversations(
         self, predicate: str, parameters: tuple[object, ...], through: int,
@@ -469,7 +507,8 @@
                 counts[query] = FieldCodec.decode(int, db.execute(
                     f'WITH {ReadCursor.sql("?")}, requests AS ('
                     f'SELECT \'viewer\' AS actor,* FROM "{Addressed.table.declared_name}" '
-                    f'JOIN "{MessageInput.table.declared_name}" USING(seq) WHERE seq > ? AND seq <= ? '
+                    f'JOIN "{MessageInput.table.declared_name}" USING(seq) '
+                    f'WHERE seq > ? AND seq <= ? '
                     f'AND ({query.predicate}) AND sender NOT IN (SELECT value FROM json_each(?))), '
                     f'{ReadCursor.joined_sql} SELECT count(*) FROM cursors WHERE {ReadCursor.unread_sql}',
                     (CursorRow.binding(query.cursors), max(after, query.lower_bound), through,
@@ -498,6 +537,14 @@
         with self.reader_snapshot() as db:
             yield db
 
+    @contextmanager
+    def captured_reading(self) -> Iterator[BusSnapshot]:
+        # Unlike a nested ordinary read, a presentation outlives this worker's
+        # next request. Its transaction must never enter the thread-local pool.
+        with CoordinationReadUnavailable.checking():
+            with closing(self.read_snapshot(self.path, check_same_thread=False)) as db:
+                yield BusSnapshot.from_connection(db)
+
 
 class Bus(FileStore, StoredBusReader, ReconciledFile, WalStore, CarriedWriter, BusFile, Watched):
     """One SQLite transaction owns the sequence and all of a record's fields."""
@@ -655,3 +702,7 @@
     @contextmanager
     def reading(self) -> Iterator[BusSnapshot]:
         yield self
+
+    @contextmanager
+    def captured_reading(self) -> Iterator[BusSnapshot]:
+        yield self

src/agent_comms/bus_tables.py

@@ -132,6 +132,17 @@ def require_matched(cls, matches: Iterable[ReferenceMatch], refusal: Exception)
         if not all(match.matched for match in matches):
             raise refusal
 
+    @classmethod
+    def reference_predicate(
+        cls, references: tuple[MessageReference, ...],
+    ) -> tuple[str, tuple[object, ...]]:
+        plans = FieldCodec.field_plans(MessageReference)
+        columns = cls._column_list(tuple(plan.name for plan in plans))
+        selected = ', '.join('json_extract(value, ?)' for plan in plans)
+        return (f'({columns}) IN (SELECT {selected} FROM json_each(?))',
+                (*(f'$.{plan.key}' for plan in plans),
+                 FieldCodec.compact([FieldCodec.encode(reference) for reference in references])))
+
     @classmethod
     def require_sequences(cls, db: sqlite3.Connection, sequences: tuple[int, ...]) -> None:
         matches = ReferenceMatch.read(db.execute(

src/agent_comms/channel_targets.py

@@ -16,15 +16,15 @@
 from .repo import Checkout
 from .response_policy import DirectPolicy, MentionedPolicy
 from .thread_identity import ThreadIncarnation
+from .thread_name import ThreadName
 from .typed_table import sql_literal
 from .wake_policy import BoundedTriageWake, FullWake, MessageAudience
 
 if TYPE_CHECKING:
     from .audience_manifest import AudiencedMessage, FrozenRecipient
     from .catalog_document import CatalogAccess, CatalogDocument
     from .channels import Channel
-    from .cli_command import CliCommand
-    from .cli_result import CliResult
+    from .cli_command import CliCommand, CommandBinding
     from .comms import Comms
     from .goal_management import Goals
     from .historical_views import ChannelHistory, DMHistory, HistoryView
@@ -166,7 +166,7 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
     def bindings[C: CliCommand](
         self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved,
         *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]: ...
+    ) -> tuple[CommandBinding, ...]: ...
 
     def history_view(self, viewer: DisplayViewer, require_channel: Callable[[str], Channel],
                      channel: type[ChannelHistory], direct: type[DMHistory]) -> HistoryView:
@@ -177,7 +177,7 @@ def history_result(self, require_channel: Callable[[str], Channel], messages: tu
         raise ValueError("History requires a #channel or DM peer.")
 
     def contexts(
-        self, declaration: type[CliCommand[CliResult]], channel: ChannelSelection,
+        self, declaration: type[CliCommand], channel: ChannelSelection,
     ) -> tuple[SurroundingChannel, ...]:
         return (NoChannel(),)
 
@@ -393,7 +393,7 @@ def history_result(self, require_channel: Callable[[str], Channel], messages: tu
         return self.history_result_type.for_target(self.name, messages, has_older)
 
     def contexts(
-        self, declaration: type[CliCommand[CliResult]], channel: ChannelSelection,
+        self, declaration: type[CliCommand], channel: ChannelSelection,
     ) -> tuple[SurroundingChannel, ...]:
         return declaration.selection_channels(channel, self.name)
 
@@ -413,15 +413,13 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
     def bindings[C: CliCommand](
         self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved: ResolvedThread,
         *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
+    ) -> tuple[CommandBinding, ...]:
         return declaration.thread_bindings(comms, resolved.thread, resolved.status, channel, catalog=catalog)
 
     def validate(self, message: Message) -> None:
         if message.sender == self.name and message.type.conversation_message:
             raise RelationViolationError(f"Thread {message.sender!r} cannot message itself.")
-        allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_")
-        if not set(self.name) <= allowed:
-            raise ValueError(f"Message target {self.name!r} is not a thread name or #channel.")
+        ThreadName.require(self.name)
 
     def wake_policy(self, recipient: FrozenRecipient, mentions: frozenset[str]) -> FullWake:
         return FullWake()
@@ -483,10 +481,7 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "name", self.incarnation.name)
 
     def selected_name(self, snapshot: RegistryProvenance) -> str:
-        if not self.incarnation.current(snapshot):
-            raise UnregisteredThreadError(
-                f"Original DM peer {self.name!r} is no longer registered.")
-        return super().selected_name(snapshot)
+        return self.incarnation.require_current(snapshot).name
 
     def require_publication(self, message: Message, snapshot: RegistryProvenance,
                             require_channel: Callable[[str], Channel]) -> None:
@@ -562,7 +557,7 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
     def bindings[C: CliCommand](
         self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved: ResolvedChannel,
         *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
+    ) -> tuple[CommandBinding, ...]:
         return declaration.channel_bindings(comms, resolved, project=project, snapshot=snapshot, catalog=catalog)
 
     def validate(self, sender: Message | None = None):
@@ -654,7 +649,7 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
     def bindings[C: CliCommand](
         self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved,
         *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
+    ) -> tuple[CommandBinding, ...]:
         return declaration.project_bindings(comms, resolved)
 
 

src/agent_comms/check_root.py

@@ -24,7 +24,7 @@
 from .launch_runtime import ThreadLaunch
 from .native_entries import ImportedSessionWriter, NativeEntry
 from .native_session_files import CheckSession
-from .read_basis import MemberActivityDisplayScope
+from .read_basis import NamedActivityDisplayScope
 from .registry_document import RosterVisibility
 from .session_file import RegistrySessionLocation
 from .thread_storage import StorageRoots
@@ -94,11 +94,11 @@ def _populate(self, copied: Comms) -> None:
         targets = frozenset(
             target for name in names for target in catalog.targets_for(snapshot.require(name).tags)
         )
-        scope = MemberActivityDisplayScope("#check", targets, participant_names=selected | aliases)
+        scope = NamedActivityDisplayScope("#check", targets, participants=selected | aliases)
         with self.source.bus.bus.reading() as source:
-            messages = scope.history(source, limit=self.max_records)
+            history = scope.history(source, limit=self.max_records)
         copied.registry.store.prepare_native_root()
-        Bus.install(Bus.path_in(copied.root), history=messages)
+        Bus.install(Bus.path_in(copied.root), history=history)
         copied.channels.catalog.replace(catalog)
         declarations = {}
         for name in names:
@@ -131,7 +131,7 @@ def _populate(self, copied: Comms) -> None:
                     edit.document.advance_claim(name)
                     edit.document.owners.advance(name)
             edit.commit()
-        if messages:
+        if history:
             # Imported history is displayable, not fresh input for this root.
             # Position acceptance itself; human read markers own a different fact.
             privacy = copied.registry.store.privacy()
@@ -141,7 +141,7 @@ def _populate(self, copied: Comms) -> None:
                         AcceptanceFrontier(
                             wire_root_id=privacy.namespace,
                             recipient_lookup=thread.incarnation.require_recorded().lookup,
-                            through_seq=messages[-1].seq,
+                            through_seq=history[-1].seq,
                         ).advance(db)
 
     def _session(self, original: Thread, root: Path, slot: str, project: Path) -> str: