Put sidebar state and listing selection in their existing owners

src/agent_comms/acp.py

@@ -52,7 +52,6 @@
     encode_updates,
 )
 from .agent_event_updates import AcpEventConsumer
-from .bus_publication import stable_thread_lookup
 from .image_inputs import prompt_images
 from .cohort_foreground import _accept_visible_deliveries
 from .comms import Comms, wire
@@ -420,7 +419,7 @@ def read_admission():
             ),
         )
         await _accept_visible_deliveries(
-            bus, wire_root_id, store_path, stable_thread_lookup(owner.created_at), 0,
+            bus, wire_root_id, store_path, owner.incarnation.lookup, 0,
             owner_name=owner.name, acquire_package=execution.validate,
         )
         if self.inputs.owns_execution(session_id, owner):
@@ -433,7 +432,7 @@ def read_admission():
             return 0
         def select(store):
             with store.session.read():
-                participant = store.participants.get(stable_thread_lookup(owner.created_at))
+                participant = store.participants.get(owner.incarnation.lookup)
                 candidate = next_sealed_assignment(
                     store, participant.lookup, owner.name, after_seq=admission_after_seq,
                 )

src/agent_comms/activity.py

@@ -399,14 +399,22 @@ def _observed(self, thread: str, events: Mapping[str, Activity], *,
             return Activity(thread=thread, state=ActivityState.IDLE, timestamp=latest.timestamp)
         return latest
 
-    def all_current(self, *, active: frozenset[str] = frozenset(),
-                    threads: Iterable[str] = ()) -> dict[str, Activity]:
-        """Latest activity per thread (idle included for known threads)."""
-        result = self._read_snapshot().latest
+    def all_current(self, *, active: frozenset[str] = frozenset()) -> dict[str, Activity]:
+        """Latest activity for every observed thread."""
+        latest = self._read_snapshot().latest
+        return self._current_for(latest, latest, active)
+
+    def current_for(self, threads: Iterable[str], *,
+                    active: frozenset[str] = frozenset()) -> dict[str, Activity]:
+        """Latest activity for the selected thread names, including absent ones."""
+        return self._current_for(threads, self._read_snapshot().latest, active)
+
+    def _current_for(self, threads: Iterable[str], events: Mapping[str, Activity],
+                     active: frozenset[str]) -> dict[str, Activity]:
         now = time.time()
         return {
-            thread: self._observed(thread, result, active=thread in active, now=now)
-            for thread in dict.fromkeys((*result, *threads))
+            thread: self._observed(thread, events, active=thread in active, now=now)
+            for thread in dict.fromkeys(threads)
         }
 
     def _read_snapshot(self) -> ActivitySnapshot:

src/agent_comms/agent_activity.py

@@ -9,11 +9,13 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from .registration import Registration
 from .native_input_owner import RegistryOwner
 
 if TYPE_CHECKING:
     from .agent_events import RuntimeInfoObservation
+    from .registration import Registration
+    from .registry_document import RegistrySnapshot
+    from .bus_publication import CommittedDelivery
     from .notification_assignment import AssignmentProgress
     from .presentation import MessageNotification
 from .activity import (
@@ -22,7 +24,6 @@
 from .declared_family import DeclaredFamily
 from .audience_manifest import FrozenRecipient
 from .errors import RelationViolationError
-from .registry_document import RegistrySnapshot
 from .routing import TurnRouting
 from .runtime_info import AgentRuntimeInfo, RuntimeInfoStore
 from .store_files import _store_lock
@@ -45,7 +46,7 @@ def after_inbox_read(self, notification, source, reads, document, snapshot):
 
     def displayed_recipient(self, notification, source, reads, document, snapshot):
         return reads.displayed_recipient(
-            source.message, notification.recipient_identity, snapshot, document=document
+            source, notification.recipient_identity, snapshot, document=document
         )
 
     def selected_response(
@@ -127,30 +128,31 @@ def _observed(self, participant: Thread, owner: OwnerIdentity,
         activity = ObservedActivity.acquire(event, owner)
         return participant.execution.project_activity(participant, activity)
 
-    def all_activity(self, *, snapshot: RegistrySnapshot | None = None) -> Mapping[str, ObservedActivity]:
-        snapshot = snapshot or self.registry.snapshot()
-        events = self.activity.all_current(
-            active=frozenset(name for name, thread in snapshot.threads.items() if thread.executing),
-            threads=snapshot.threads,
+    def all_activity(self, threads: Iterable[Thread], *, snapshot: RegistrySnapshot) -> Mapping[str, ObservedActivity]:
+        requested = {thread.name: thread for thread in threads}
+        if not requested:
+            return {}
+        events = self.activity.current_for(
+            requested,
+            active=frozenset(name for name, thread in requested.items() if thread.executing),
         )
         return {
             name: self._observed(thread, snapshot.owner_identity(name), events[name])
-            for name, thread in snapshot.threads.items()
+            for name, thread in requested.items()
         }
 
-    def observe_recipients(self, recipients: Iterable[FrozenRecipient], *, snapshot: RegistrySnapshot) -> Mapping[str, RecipientActivity]:
-        """Acquire one bounded window's recipients from the original source snapshot."""
-        from .bus_publication import stable_thread_lookup
-
-        snapshot.require_unambiguous_ownership()
-        observations: dict[str, RecipientActivity] = {
-            recipient.recipient_lookup: UnavailableRecipientActivity() for recipient in recipients
-        }
-        for thread in snapshot.threads.values():
-            lookup = stable_thread_lookup(thread.created_at)
-            if lookup not in observations:
+    def observe_recipients(
+        self, recipients: Iterable[tuple[CommittedDelivery, FrozenRecipient]], *,
+        snapshot: RegistrySnapshot,
+    ) -> Mapping[str, RecipientActivity]:
+        """Acquire one bounded window's recipients from the original sends."""
+        observations: dict[str, RecipientActivity] = {}
+        for original, recipient in recipients:
+            lookup = recipient.recipient_lookup
+            if lookup in observations:
                 continue
-            observations[lookup] = thread.execution.observe_recipient(self, snapshot, thread)
+            selected = snapshot.recipient(recipient, original)
+            observations[lookup] = selected.activity(self, snapshot)
         return observations
 
     def _emit_activity(self, activity: Activity) -> None:

src/agent_comms/attempt_recovery.py

@@ -224,15 +224,14 @@ def settle_stopped_owner(cls, root: Path, owner) -> None:
     @classmethod
     def recover_owner(cls, root: Path, owner) -> None:
         """Resolve an exited predecessor once, before this worker begins draining."""
-        from .bus_publication import stable_thread_lookup
         from .coordination_tables.executions import CurrentExecutions
 
         path = root / "coordination.sqlite3"
         if not path.exists():
             return
         with Coordination(str(path)) as store:
             pointer = CurrentExecutions.one(
-                store.session._connection, owner_lookup=stable_thread_lookup(owner.created_at)
+                store.session._connection, owner_lookup=owner.incarnation.lookup
             )
             if pointer is not None and pointer.unresolved:
                 cls.recover_released_native_attempt(store, pointer.execution_id)

src/agent_comms/bus_activity_index.py

@@ -109,9 +109,7 @@ def for_views(cls, channels: Mapping[str, Channel],
         """Combine target-owned clocks using each declaration's history membership."""
         result = {}
         for name, channel in channels.items():
-            targets = channel.history_targets
-            clocks = (tuple(by_target.values()) if targets is None
-                      else tuple(by_target.get(target, cls()) for target in targets))
+            clocks = tuple(channel.history_targets.matching_values(by_target))
             result[name] = cls(max((item.last_message for item in clocks), default=0),
                                max((item.last_user_input for item in clocks), default=0))
         return result

src/agent_comms/bus_display_index.py

@@ -52,7 +52,7 @@ class DisplayCheckpoint(AppendCheckpoint):
     semantics: DisplayMetricScope
     activity: dict[str, tuple[float, float]]
     counts: dict[str, int]
-    schema: Literal[3] = field(default=3, kw_only=True)
+    schema: Literal[4] = field(default=4, kw_only=True)
 
     def __post_init__(self) -> None:
         super().__post_init__()

src/agent_comms/bus_page_index.py

@@ -14,6 +14,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, BinaryIO, Literal
 
+from .channel_targets import HistoryTargets
 from .coordination_database import CoordinationStore
 from .bus_projection import AppendCheckpoint
 from .private_path import FileRevision
@@ -224,7 +225,7 @@ def offsets(
         lower: int | None,
         upper: int | None,
         descending: bool,
-        targets: frozenset[str] | None,
+        targets: HistoryTargets,
         before_offset: int | None = None,
     ) -> Generator[BusPageRow, None, None]:
         clauses: list[str] = []
@@ -238,11 +239,7 @@ def offsets(
         if before_offset is not None:
             clauses.append("offset < ?")
             params.append(before_offset)
-        if targets is not None:
-            if not targets:
-                return
-            clauses.append("target IN (" + ",".join("?" for _ in targets) + ")")
-            params.extend(sorted(targets))
+        targets.constrain_offsets(clauses, params)
         where = " WHERE " + " AND ".join(clauses) if clauses else ""
         direction = "DESC" if descending else "ASC"
         query = (

src/agent_comms/bus_publication.py

@@ -87,21 +87,6 @@ def reject_private_wire_fields(record: Mapping[str, object]) -> None:
         raise ValueError("Bus-private wire fields cannot be exported.")
 
 
-def stable_thread_lookup(created_at: float) -> str:
-    """Immutable creation identity; duplicate timestamps fail at capture.
-
-    The registry preserves Thread.created_at across renames and restarts. Its
-    canonical name, aliases, mutable tags, PID, and session attachment are NOT
-    lookup identities. An unknown creation time (zero) cannot authorize delivery.
-    """
-    if isinstance(created_at, bool) or not isinstance(created_at, (float, int)):
-        raise ValueError("Thread creation identity is invalid.")
-    if not math.isfinite(created_at) or created_at <= 0:
-        raise ValueError("Thread creation identity is unavailable.")
-    identity = f"agent-comms:thread-creation:v1\0{float(created_at)!r}".encode()
-    return hashlib.sha256(identity).hexdigest()[:32]
-
-
 def _decision_wire(decision: WakeDecision | NoWakeDecision) -> dict[str, str]:
     from .wake import NoWakeDecision, WakeDecision
 

src/agent_comms/bus_route_counts.py

@@ -29,7 +29,7 @@
 from .typed_table import Column, ForeignKey, Index, SQLiteJournalMode, TypedRow, TypedTable
 
 if TYPE_CHECKING:
-    from .registry_document import RegistrySnapshot
+    from .registry_provenance import RegistryProvenance
     from .wire_log import OpenedWireSnapshot
 
 
@@ -452,7 +452,7 @@ class OpenedRouteCounts(BusRouteCounts):
 
     def __init__(
         self, bus_path: Path, source: OpenedWireSnapshot,
-        registry: RegistrySnapshot, deliveries: Mapping[str, DeliveryScope],
+        registry: RegistryProvenance, deliveries: Mapping[str, DeliveryScope],
     ):
         super().__init__(bus_path)
         self.source = source
@@ -582,7 +582,7 @@ def _request_bindings(self, actors):
             ))
         return {
             "targets": json.dumps({
-                actor: sorted(self.deliveries[actor].index_targets) for actor in actors
+                actor: sorted(self.deliveries[actor].index_targets.names) for actor in actors
             }),
             "senders": json.dumps(senders), "unrecorded": json.dumps(unrecorded),
             "through": self.source.through,

src/agent_comms/catalog_document.py

@@ -6,10 +6,10 @@
 from collections.abc import Mapping
 from dataclasses import dataclass, field, replace
 
-from .channel_targets import BuiltinChannel, Tag
+from .channel_targets import BuiltinChannel, HistoryTargets, Tag
 from .channels import Channel, SavedView, ViewPredicate
 from .display_order import ChannelSort, CreatedThreadSort, NameChannelSort, ThreadSort
-from .thread_provenance import ThreadProvenance
+from .thread_identity import ThreadProvenance
 
 
 @dataclass(frozen=True, slots=True)
@@ -101,7 +101,7 @@ def targets_for(self, tags: frozenset[str]) -> frozenset[str]:
             f"#{tag}" for tag in tags
         )
 
-    def history_targets(self, target: str) -> frozenset[str] | None:
+    def history_targets(self, target: str) -> HistoryTargets:
         return self.resolve(target).history_targets
 
     def is_view_target(self, target: str) -> bool:

src/agent_comms/channel_targets.py

@@ -4,10 +4,63 @@
 
 from dataclasses import dataclass
 from abc import ABC, abstractmethod
+from collections.abc import Iterator, Mapping
+
+from .declared_family import DeclaredFamily
 from typing import ClassVar
 from enum import StrEnum
 
 
+class HistoryTargets(DeclaredFamily, affix="HistoryTargets"):
+    """Captured target membership, including an explicit unrestricted member."""
+
+    @abstractmethod
+    def includes(self, target: str) -> bool: ...
+
+    @abstractmethod
+    def matching_values[T](self, values: Mapping[str, T]) -> Iterator[T]: ...
+
+    @abstractmethod
+    def constrain_offsets(self, clauses: list[str], params: list[object]) -> None: ...
+
+    @abstractmethod
+    def require_names(self) -> frozenset[str]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class AllHistoryTargets(HistoryTargets):
+    def includes(self, target):
+        return True
+
+    def matching_values(self, values):
+        return iter(values.values())
+
+    def constrain_offsets(self, clauses, params):
+        pass
+
+    def require_names(self):
+        from .errors import RelationViolationError
+        raise RelationViolationError("An aggregate is not an exportable conversation.")
+
+
+@dataclass(frozen=True, slots=True)
+class NamedHistoryTargets(HistoryTargets):
+    names: frozenset[str]
+
+    def includes(self, target):
+        return target in self.names
+
+    def matching_values(self, values):
+        return (values[target] for target in self.names if target in values)
+
+    def constrain_offsets(self, clauses, params):
+        clauses.append("target IN (" + ",".join("?" for _ in self.names) + ")")
+        params.extend(sorted(self.names))
+
+    def require_names(self):
+        return self.names
+
+
 class BuiltinChannel(StrEnum):
     ANY = "#any"
     NONE = "#none"
@@ -38,8 +91,8 @@ def matches(self, tags: frozenset[str]) -> bool:
         return self is not self.NONE or not tags
 
     @property
-    def history_targets(self) -> frozenset[str] | None:
-        return None if self.aggregate else frozenset((self.value,))
+    def history_targets(self) -> HistoryTargets:
+        return AllHistoryTargets() if self.aggregate else NamedHistoryTargets(frozenset((self.value,)))
 
 
 class CommandTarget(ABC):

src/agent_comms/channels.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass, field
 from enum import StrEnum
 
-from .channel_targets import _TAG_CHARS, BuiltinChannel, Tag
+from .channel_targets import _TAG_CHARS, BuiltinChannel, HistoryTargets, NamedHistoryTargets, Tag
 from .declared_family import DeclaredFamily
 from .display_order import CreatedThreadSort, ThreadSort
 from .errors import RelationViolationError
@@ -75,6 +75,13 @@ def matches(self, tags: frozenset[str]) -> bool:
         builtin = self.builtin
         return builtin.matches(tags) if builtin is not None else bool(self.tags & tags)
 
+    @property
+    def tag_description(self) -> str:
+        builtin = self.builtin
+        if builtin is not None:
+            return builtin.value
+        return ', '.join(sorted(self.tags))
+
     def can_set_archived(self, archived: bool) -> bool:
         """Only nonbuiltin channels admit a changed archive preference."""
         return self.builtin is None and self.archived != archived
@@ -93,11 +100,11 @@ def exact(self) -> bool:
         return self.view is None and self.builtin is None
 
     @property
-    def history_targets(self) -> frozenset[str] | None:
+    def history_targets(self) -> HistoryTargets:
         if self.view is not None:
             return self.view.history_targets
         builtin = self.builtin
-        return builtin.history_targets if builtin is not None else frozenset({self.name})
+        return builtin.history_targets if builtin is not None else NamedHistoryTargets(frozenset({self.name}))
 
     def to_wire(self) -> dict[str, object]:
         return FieldCodec.encode(self)
@@ -157,10 +164,10 @@ class SavedView:
     original_targets: frozenset[str] = frozenset()
 
     @property
-    def history_targets(self) -> frozenset[str]:
+    def history_targets(self) -> NamedHistoryTargets:
         # Original addressed messages remain attached to their immutable target;
         # this is a read projection, never a rewritten message or delivery scope.
-        return self.original_targets | frozenset(f"#{tag}" for tag in self.predicate.tags)
+        return NamedHistoryTargets(self.original_targets | frozenset(f"#{tag}" for tag in self.predicate.tags))
 
     def __post_init__(self) -> None:
         if (