Acquire display identities and relationship inputs through their declared owners

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
@@ -49,6 +49,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 +131,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 +180,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 +196,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 +263,24 @@
 
     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,
+        )
 
     def message_history(self) -> Iterator[Message]:
         """Public bulk export, paged within one committed SQLite snapshot."""
@@ -288,22 +313,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 +494,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 +524,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 +689,7 @@
     @contextmanager
     def reading(self) -> Iterator[BusSnapshot]:
         yield self
+
+    @contextmanager
+    def captured_reading(self) -> Iterator[BusSnapshot]:
+        yield self

src/agent_comms/channel_targets.py

@@ -16,6 +16,7 @@
 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
 
@@ -419,9 +420,7 @@ def bindings[C: CliCommand](
     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 +482,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:

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:

src/agent_comms/display_references.py

@@ -158,8 +158,8 @@ class DisplayReferences(RootedReferences):
     @staticmethod
     def recorded(root, source) -> DisplayReferences:
         registry = source.provenance
-        targets = {name: HistoricalDisplayReference(str(root), thread.incarnation, source.key)
-                   for name, thread in registry.named_members()}
+        targets = {name: HistoricalDisplayReference(str(root), incarnation, source.key)
+                   for name, incarnation in registry.named_incarnations()}
         return DisplayReferences(str(root), targets)
 
     def for_source(self, source) -> DisplayReferences:
@@ -209,7 +209,7 @@ class CurrentDisplayReferences(SnapshotRestored, DisplayReferences):
 
     @classmethod
     def from_registry(cls, root: str, registry: RegistrySnapshot, channels: tuple[str, ...] = ()) -> Self:
-        return cls(root, {name: thread.incarnation for name, thread in registry.named_members()}, channels)
+        return cls(root, dict(registry.named_incarnations()), channels)
 
     def __post_init__(self) -> None:
         """The target and lookup views of the one stored member map."""