P434: present other-route threads with incarnation-safe display references

src/agent_comms/display_references.py

@@ -9,9 +9,13 @@
 from dataclasses import dataclass, field
 from functools import cached_property
 import re
+from typing import TYPE_CHECKING
 
-from .declared_family import DeclaredFamily
-from .thread_identity import ThreadIncarnation
+from agent_comms.declared_family import DeclaredFamily
+from agent_comms.thread_identity import ThreadIncarnation
+
+if TYPE_CHECKING:
+    from agent_comms.registry_document import RegistrySnapshot
 
 
 @dataclass(frozen=True)
@@ -142,15 +146,6 @@ class DisplayReferences(RootedReferences):
     description = ""
     targets: Mapping[str, DisplayReference] = field(default_factory=dict)
 
-    @classmethod
-    def capture(cls, root, registry, channels) -> DisplayReferences:
-        targets = {name: ThreadDisplayReference(str(root), thread.incarnation)
-                   for name, thread in registry.threads.items()}
-        targets.update((alias, targets[name]) for alias, name in registry.aliases.items()
-                       if name in targets)
-        targets.update((name, ChannelDisplayReference(str(root), name)) for name in channels)
-        return cls(str(root), targets)
-
     @classmethod
     def recorded(cls, root, source) -> DisplayReferences:
         registry = source.provenance
@@ -188,3 +183,27 @@ def accepts_history(self, root):
 
     def contains(self, reference: DisplayReference) -> bool:
         return self.targets.get(reference.name) == reference
+
+
+@dataclass(frozen=True)
+class CurrentDisplayReferences(DisplayReferences):
+    """Current targets retain their registry cut, not another identity index.
+
+    Recorded catalogs share name recognition but cannot resolve a live target.
+    """
+
+    registry: RegistrySnapshot = field(kw_only=True, repr=False, compare=False)
+
+    @classmethod
+    def capture(cls, root, registry: RegistrySnapshot, channels) -> CurrentDisplayReferences:
+        targets = {name: ThreadDisplayReference(str(root), thread.incarnation)
+                   for name, thread in registry.threads.items()}
+        targets.update((alias, targets[name]) for alias, name in registry.aliases.items()
+                       if name in targets)
+        targets.update((name, ChannelDisplayReference(str(root), name)) for name in channels)
+        return cls(str(root), targets, registry=registry)
+
+    def thread(self, incarnation: ThreadIncarnation) -> ThreadDisplayReference:
+        """Resolve an exact birth to its current name; a missing birth fails."""
+        current = self.registry.by_lookup[incarnation.require_recorded().lookup]
+        return ThreadDisplayReference(self.root, current.incarnation)

src/agent_comms/errors.py

@@ -17,6 +17,12 @@ class UnknownChannelError(ToolRefusedError):
     """A target is absent from the catalog and current thread tags."""
 
 
+class RouteChangedError(ValueError):
+    """The selected Comms route was replaced; reopen against the current one."""
+
+    reader_sentence = "On another Comms route; reopen this view"
+
+
 class RelationViolationError(ValueError):
     """A required relation between declarations cannot be proved."""
 

src/agent_comms/presentation.py

@@ -25,7 +25,7 @@
 from .coordination_errors import DMPaintedBoundChanged, StaleRevision
 from .declared_family import DeclaredFamily
 from .display_order import ChannelSort, NameChannelSort
-from .display_references import DisplayReferences
+from agent_comms.display_references import CurrentDisplayReferences
 from .external_transcript import ExternalTranscript
 from .field_codec import (
     FieldCodec,
@@ -720,7 +720,7 @@ class CoordinationSnapshot:
     unread: Mapping[str, UnreadPresentation]
     channel_unread: Mapping[str, UnreadPresentation]
     last_sent: Mapping[str, float]
-    references: DisplayReferences
+    references: CurrentDisplayReferences
     channel_order: type[ChannelSort] = NameChannelSort
     thread_unread: Mapping[str, UnreadPresentation] = field(default_factory=dict)
     read_marker_notice: ReadNotice = field(default_factory=NoReadNotice)
@@ -748,7 +748,7 @@ def capture(
             channel_unread={name: ExactUnread(count) for name, count in channel_unread.items()},
             last_sent=sent,
             thread_unread={name: UnobservedUnread() for name in threads},
-            references=DisplayReferences.capture(root, registry, declarations),
+            references=CurrentDisplayReferences.capture(root, registry, declarations),
             channel_order=catalog.list_order, read_marker_notice=read_marker_notice,
         )
 

src/agent_comms/registry_document.py

@@ -6,9 +6,9 @@
 import time
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass, field, fields, replace
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Literal, Annotated
+from typing import TYPE_CHECKING, Any, Literal, Annotated, Self
 
 
 from .child_process import NoProcess, ProcessIdentity
@@ -710,6 +710,19 @@ class RegistrySnapshot(RegistryPresence[Mapping[str, ThreadStatus]], RegistryPro
     _claimed_generations: Mapping[str, int] = field(repr=False)
     by_lookup: Mapping[str, Thread] = field(repr=False, compare=False)
 
+    def __post_init__(self) -> None:
+        """Every construction owns and freezes its declared registry maps."""
+        for declaration in fields(self):
+            object.__setattr__(
+                self, declaration.name, MappingProxyType(dict(getattr(self, declaration.name)))
+            )
+
+    def __reduce__(self) -> tuple[type[Self], tuple[dict[str, object], ...]]:
+        """Preparation workers restore snapshots through their constructor."""
+        return type(self), tuple(
+            dict(getattr(self, declaration.name)) for declaration in fields(self)
+        )
+
     def claim_generation(self, name: str) -> int:
         """The allocated input generation follows this registry's alias lineage."""
         return self._claimed_generations[self.canonical_name(name)]
@@ -745,13 +758,13 @@ def capture(cls, document: RegistryDocument) -> RegistrySnapshot:
                 raise RelationViolationError("Registry creation identities collide.")
             indexed[lookup] = thread
         return cls(
-            threads=MappingProxyType(dict(document.threads)),
-            statuses=MappingProxyType(dict(document.statuses)),
-            last_seen=MappingProxyType(dict(document.last_seen)),
-            aliases=MappingProxyType(dict(document.aliases)),
-            owner_generations=MappingProxyType(dict(document.owners.generations)),
-            _claimed_generations=MappingProxyType(dict(document.claims.generations)),
-            by_lookup=MappingProxyType(indexed),
+            threads=document.threads,
+            statuses=document.statuses,
+            last_seen=document.last_seen,
+            aliases=document.aliases,
+            owner_generations=document.owners.generations,
+            _claimed_generations=document.claims.generations,
+            by_lookup=indexed,
         )
 
     def observed_activity(

src/agent_comms/route_selection.py

@@ -14,13 +14,12 @@
 from pathlib import Path
 from typing import TypeVar
 
+from agent_comms import errors
+
 from .active_route import CommsRoute, resolve_comms_route
 from .launch_root import LaunchRoot
 
 
-class RouteChanged(ValueError):
-    """The selected Comms route was replaced; reopen against the current one."""
-
 T = TypeVar("T")
 
 
@@ -49,7 +48,7 @@ def capture(cls, source: str | Path | None = None) -> RouteSelection:
         route = resolve_comms_route()
         selection = cls(route, route.observe_root())
         if source is not None and Path(source).expanduser().resolve() != selection.root:
-            raise RouteChanged("Comms route changed; reopen this view")
+            raise errors.RouteChangedError(errors.RouteChangedError.reader_sentence)
         return selection
 
     @classmethod

src/agent_comms/thread_presentation.py

@@ -7,7 +7,7 @@
 from typing import TYPE_CHECKING, ClassVar, Self
 
 from agent_comms.coordination_errors import StaleRevision
-from agent_comms.errors import RelationViolationError, UnregisteredThreadError
+from agent_comms.errors import RelationViolationError, RouteChangedError, UnregisteredThreadError
 
 from .declared_family import DeclaredFamily
 from .thread_identity import OwnerIdentity
@@ -125,6 +125,14 @@ def presentation(self, title: str) -> ThreadPresentation:
         return ThreadPresentation(title, "○", "Thread is no longer available", status=self)
 
 
+@dataclass(frozen=True, slots=True)
+class OtherRouteThread(ThreadPresentationStatus):
+    """The view thread lives on a Comms route that is not selected."""
+
+    def presentation(self, title: str) -> ThreadPresentation:
+        return ThreadPresentation(title, "↗", RouteChangedError.reader_sentence, status=self)
+
+
 class ThreadOwnerBinding(DeclaredFamily, affix="ThreadOwnerBinding"):
     """A projection of the registry lease, never an alternative owner store."""
 

src/agent_comms/ui_model/status.py

@@ -13,6 +13,7 @@
 
 from agent_comms.thread_presentation import (
     IdleThreadPresentationStatus,
+    OtherRouteThread,
     SessionDisconnected,
     StatusPending,
     ThreadRetired,
@@ -84,6 +85,10 @@ def pending(cls, name: str) -> ThreadStatusModel:
     def retired(cls, name: str) -> ThreadStatusModel:
         return cls.known(name, ThreadRetired().presentation(name))
 
+    @classmethod
+    def other_route(cls, name: str) -> ThreadStatusModel:
+        return cls.known(name, OtherRouteThread().presentation(name))
+
     @classmethod
     def unreadable(cls, name: str, error: UnreadableHistory) -> ThreadStatusModel:
         return cls.known(name, ThreadPresentation(

tests/test_sidebar_observation.py

@@ -1,7 +1,11 @@
 """Sidebar states answer the same questions before and after a real read."""
 
+import pickle
+from dataclasses import fields, replace
+
 import pytest
 
+from agent_comms.thread_status import RunningThreadStatus, StoppedThreadStatus
 from agent_comms.ui_model.sidebar import SidebarModel
 from agent_comms.ui_model.unread import ExactUnread, UnobservedUnread
 
@@ -17,7 +21,8 @@ def test_sidebar_apply_and_failure_publish_their_original_answers(comms):
     assert model.observation.require_snapshot() is snapshot
     assert model.observation.channel_order() is snapshot.channel_order
     assert model.observation.channel_unread("#any") == ExactUnread(0)
-    assert model.observation.bus_unread("missing") == ExactUnread(0)
+    with pytest.raises(KeyError):
+        model.observation.bus_unread("missing")
     assert set(model.channels.rows) == set(snapshot.channels)
     failure = ValueError("original sidebar read failure")
     model.clear(failure)
@@ -26,3 +31,53 @@ def test_sidebar_apply_and_failure_publish_their_original_answers(comms):
     with pytest.raises(ValueError) as caught:
         model.observation.require_snapshot()
     assert caught.value is failure
+
+
+@pytest.fixture
+def declare_reference(comms):
+    def declare(name="reference", status=StoppedThreadStatus()):
+        return comms.registry.declare(comms.registry.new_thread(
+            name=name, tags=frozenset(), worktree=str(comms.root.parent),
+        ), status=status)
+    return declare
+
+
+def test_registry_snapshot_constructor_and_pickle_own_immutable_maps(comms, declare_reference):
+    original = declare_reference()
+    captured = comms.registry.snapshot()
+    supplied = dict(captured.threads)
+    snapshot = replace(captured, threads=supplied)
+    supplied.clear()
+    for restored in (snapshot, pickle.loads(pickle.dumps(snapshot, protocol=5))):
+        assert restored.threads[original.name] == original
+        assert restored.by_lookup[original.incarnation.require_recorded().lookup] == original
+        for declaration in fields(restored):
+            with pytest.raises(TypeError):
+                getattr(restored, declaration.name)["unowned"] = None
+
+
+def test_current_reference_resolves_original_birth_after_registry_rename(comms, declare_reference):
+    original = declare_reference(status=RunningThreadStatus())
+    comms.registry.rename(original.name, "renamed")
+    snapshot = comms.views.viewer_snapshot(str(comms.root.parent))
+    references = pickle.loads(pickle.dumps(snapshot, protocol=5)).references
+    current = references.thread(original.incarnation)
+    assert current.name == "renamed"
+    assert current.incarnation == original.incarnation
+    assert current.current_in(references)
+
+
+def test_recreated_thread_reference_resolves_new_birth_not_same_name(comms, declare_reference):
+    original = declare_reference()
+    before = comms.views.viewer_snapshot(str(comms.root.parent)).references
+    comms.threads.delete(original.name)
+    recreated = declare_reference(original.name)
+    snapshot = comms.views.viewer_snapshot(str(comms.root.parent))
+    references = pickle.loads(pickle.dumps(snapshot, protocol=5)).references
+    assert recreated.incarnation != original.incarnation
+    assert references.thread(recreated.incarnation).incarnation == recreated.incarnation
+    assert before.thread(original.incarnation).incarnation == original.incarnation
+    with pytest.raises(KeyError):
+        references.thread(original.incarnation)
+    with pytest.raises(TypeError):
+        references.registry.by_lookup[original.incarnation.require_recorded().lookup] = original

tests/test_thread_presentation.py

@@ -4,11 +4,11 @@
 
 import pytest
 
-from agent_comms.errors import UnregisteredThreadError
+from agent_comms.errors import RouteChangedError, UnregisteredThreadError
 from agent_comms.field_codec import FieldCodec
 from agent_comms.notification_progress import SendingProgress
 from agent_comms.thread_presentation import (
-    IdleThreadPresentationStatus, NoThreadPresentation, SessionDisconnected,
+    IdleThreadPresentationStatus, NoThreadPresentation, OtherRouteThread, SessionDisconnected,
     StatusPending, ThreadPresentationStatus, WorkingThreadPresentationStatus,
 )
 from agent_comms.ui_model.delivery import InputDelivery
@@ -52,6 +52,14 @@ def test_missing_presentation_owns_source_absence_and_retired_status(comms):
         presentations.require("missing")
 
 
+def test_other_route_status_presents_the_route_owners_explanation():
+    row = ThreadStatusModel.other_route("x")
+    assert row.name == "x"
+    assert isinstance(row.status, OtherRouteThread)
+    assert row.lines == row.summary == (RouteChangedError.reader_sentence,)
+    assert row.presentation.title == "x"
+
+
 @pytest.mark.parametrize("original", [
     IdleThreadPresentationStatus(), WorkingThreadPresentationStatus(),
     SendingProgress(23), SessionDisconnected("original disconnect"),