Bound recipient and activity projections to requested threads

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/acp_extension.py

@@ -136,7 +136,7 @@ def logical_key(self) -> tuple[str, ...]: ...
 
     @property
     def owner_created_at(self):
-        return self.admission.incarnation.created_at
+        return self.admission.incarnation.require_recorded().created_at
 
     @property
     def admission_generation(self):
@@ -190,11 +190,11 @@ class CursorScope(AttachmentScope):
 
     def __post_init__(self):
         WireRootIdText.decode(self.wire_root_id)
+        self.admission.incarnation.require_recorded()
         if (
             not self.session_id
             or self.owner_pid <= 0
             or self.admission.admission_generation <= 0
-            or self.admission.incarnation.created_at <= 0
         ):
             raise ValueError("Invalid native cursor scope")
 
@@ -467,14 +467,6 @@ class CoordinationChangedUpdate(AgentCommsUpdate):
     title: str
     context_usage: ContextUsage | None
 
-    @property
-    def source_identity(self):
-        from .transcript_receipts import AssignedSourceIdentity
-
-        if self.thread.has_recorded_birth:
-            return AssignedSourceIdentity(self.wire_root, self.thread)
-        return None
-
     @property
     def persistence(self) -> str:
         return "shared on-disk wire"

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,21 +224,19 @@ 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_response import PublicationAppendDispatches
         from .coordination_tables.executions import CurrentExecutions
 
         path = root / "coordination.sqlite3"
         if not path.exists():
             return
         with Coordination(str(path)) as store:
-            lookup = stable_thread_lookup(owner.created_at)
+            lookup = owner.incarnation.lookup
             pointer = CurrentExecutions.one(store.session._connection, owner_lookup=lookup)
             if pointer is None or not pointer.unresolved:
                 return
-            # A frozen response may already exist on the bus. The released
-            # owner proves exclusion; the original keyed dispatch proves this
-            # exact publication. Resolve it before considering UNKNOWN input.
+            # The released owner and exact keyed dispatch prove an existing
+            # frozen reply before any attempt is marked uncertain.
             with VerifiedOwnerLoss.observe_native_release(store, pointer.execution_id) as loss:
                 loss.require_native_exit()
                 PublicationAppendDispatches.resolve_retained(

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

src/agent_comms/catalog_document.py

@@ -9,7 +9,7 @@
 from .channel_targets import BuiltinChannel, 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)

src/agent_comms/claim_admission.py

@@ -16,7 +16,7 @@
 from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.coordinator import Coordination
 
-from .bus_publication import CommittedDelivery, stable_thread_lookup
+from .bus_publication import CommittedDelivery
 from .channel_targets import BuiltinChannel
 from .cohort_schema import assert_cohort_schema
 from .comms import Comms
@@ -79,7 +79,7 @@ def _verify_selected_wake_state(
         raise IdentityConflict("Wake source or owner turn does not match") from error
     if initial.message.reference != admission.source:
         raise IdentityConflict("Wake source message changed")
-    if stable_thread_lookup(owner.created_at) != admission.recipient_lookup:
+    if owner.incarnation.lookup != admission.recipient_lookup:
         raise IdentityConflict("Wake recipient incarnation changed")
     if generation != admission.owner_admission_generation or turn.id != admission.turn_id:
         raise IdentityConflict("Wake source or owner turn does not match")

src/agent_comms/cli_commands.py

@@ -968,6 +968,11 @@ class StatusCliCommand(CliCommand):
     help = "Live activity: what each thread is doing right now"
 
     def apply(self, ctx: Comms) -> Any:
+        snapshot = ctx.registry.snapshot()
+        running = (
+            thread for name, thread in snapshot.threads.items()
+            if snapshot.statuses[name].active
+        )
         return {
             "status": [
                 {
@@ -976,7 +981,9 @@ def apply(self, ctx: Comms) -> Any:
                     "detail": activity.detail,
                     "ts": activity.timestamp,
                 }
-                for thread, activity in sorted(ctx.agents.all_activity().items())
+                for thread, activity in sorted(
+                    ctx.agents.all_activity(running, snapshot=snapshot).items()
+                )
             ]
         }
 

src/agent_comms/cohort_foreground.py

@@ -30,7 +30,6 @@
 from agent_comms.coordination_errors import IdentityConflict, PublicationActivationBlocked
 from agent_comms.coordinator import Coordination
 
-from .bus_publication import stable_thread_lookup
 from .child_process import ProcessIdentity
 from .cohort_schema import CohortDeliveryReceipts, assert_cohort_schema
 from .comms import Comms
@@ -197,7 +196,7 @@ def admit(resources):
             comms.registry.register(thread)
         with Coordination(str(root / "coordination.sqlite3")) as store:
             store.install_private_runtime()
-            lookup = stable_thread_lookup(comms.registry.require(name).created_at)
+            lookup = comms.registry.require(name).incarnation.lookup
             store.participants.register(lookup, name, name, committed=True)
         return thread, lookup, action
 

src/agent_comms/coordination_response.py

@@ -24,7 +24,6 @@
 from pathlib import Path
 from typing import Literal, Type
 
-from agent_comms.bus_publication import stable_thread_lookup
 from agent_comms.cohort_schema import assert_cohort_schema
 from agent_comms.coordination_errors import (
     IdentityConflict,
@@ -307,7 +306,7 @@ class LiveResponseOwner(RegistryOwner):
 
     @property
     def recipient_lookup(self) -> str:
-        return stable_thread_lookup(self.thread.created_at)
+        return self.thread.incarnation.lookup
 
     def require_live(self, snapshot: RegistrySnapshot, fence: OwnerFence) -> None:
         self.require_snapshot(snapshot, "response owner turn stopped or changed before publication")

src/agent_comms/coordination_tables/assignments.py

@@ -44,11 +44,10 @@ def recipient_identity(self):
 
     def require_selected_source(self, initial, owner) -> None:
         from agent_comms.audience_manifest import FrozenRecipient
-        from agent_comms.bus_publication import stable_thread_lookup
         from agent_comms.coordination_errors import IdentityConflict
 
         if self.recipient_identity != FrozenRecipient(
-            stable_thread_lookup(owner.created_at), owner.name
+            owner.incarnation.lookup, owner.name
         ):
             raise IdentityConflict("wake frame does not match a committed recipient")
         self.require_committed_source(initial)

src/agent_comms/cursor_owner.py

@@ -6,7 +6,6 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from .bus_publication import stable_thread_lookup
 from .coordinated_runtime_schema import assert_native_runtime_schema
 from .coordination_errors import IdentityConflict, StaleFence
 from .coordinator import Coordination
@@ -39,7 +38,7 @@ def participant_identity(self):
 
     @property
     def lookup(self) -> str:
-        return stable_thread_lookup(self.thread.created_at)
+        return self.thread.incarnation.lookup
 
     def require_live(self, bus: MessageBus, registry: RegistrySnapshot, reason: str) -> None:
         if bus.log._private_marker_unlocked().root_id != self.wire_root_id:

src/agent_comms/diagnostics.py

@@ -35,11 +35,9 @@ def owner_process_output(root: Path, thread: Thread):
     capture. Each launch gets its own private file, without changing thread or
     input disposition state.
     """
-    from .bus_publication import stable_thread_lookup
-
     directory = root / "diagnostics"
     directory.mkdir(mode=0o700, exist_ok=True)
-    path = directory / f"owner-{stable_thread_lookup(thread.created_at)}-{uuid4().hex}.log"
+    path = directory / f"owner-{thread.incarnation.lookup}-{uuid4().hex}.log"
     descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     with os.fdopen(descriptor, "w") as output:
         output.write(f"Owner launch: {thread.name}\n")

src/agent_comms/external_transcript.py

@@ -17,7 +17,6 @@
 from pathlib import Path
 from typing import ClassVar
 
-from .bus_publication import stable_thread_lookup
 from .native_runtime_input import PublishedReplyRevision
 from .native_transcript import RecordsCut
 from .thread_identity import ThreadIncarnation
@@ -153,7 +152,7 @@ def source_paths(cls, root: Path) -> tuple[Path, ...]:
 
     @property
     def lookup(self) -> str:
-        return stable_thread_lookup(self.incarnation.created_at)
+        return self.incarnation.lookup
 
     def finish(self, *, turn_id: str, item_id: str,
                events: tuple[TranscriptEvent, ...]) -> None: