Port current display identities without losing snapshot policy

src/agent_comms/display_references.py

@@ -6,12 +6,17 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 from functools import cached_property
 import re
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Self
 
-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,23 +147,12 @@ 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:
+    @staticmethod
+    def recorded(root, source) -> DisplayReferences:
         registry = source.provenance
         targets = {name: HistoricalDisplayReference(str(root), thread.incarnation, source.key)
-                   for name, thread in registry.threads.items()}
-        targets.update((alias, targets[name]) for alias, name in registry.aliases.items()
-                       if name in targets)
-        return cls(str(root), targets)
+                   for name, thread in registry.named_members()}
+        return DisplayReferences(str(root), targets)
 
     def for_source(self, source) -> DisplayReferences:
         if source is None:
@@ -188,3 +182,32 @@ 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.
+    """
+
+    targets: Mapping[str, DisplayReference] = field(init=False)
+    registry: RegistrySnapshot = field(repr=False, compare=False)
+    channels: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        targets = {name: ThreadDisplayReference(self.root, thread.incarnation)
+                   for name, thread in self.registry.named_members()}
+        targets.update((name, ChannelDisplayReference(self.root, name)) for name in self.channels)
+        object.__setattr__(self, "targets", MappingProxyType(targets))
+
+    def __reduce__(self) -> tuple[type[Self], tuple[object, ...]]:
+        """Restore declared inputs; rebuild the target view from their owners."""
+        return type(self), tuple(
+            getattr(self, declaration.name) for declaration in fields(self) if declaration.init
+        )
+
+    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

@@ -22,6 +22,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

@@ -23,7 +23,7 @@
 from .coordination_errors import DMPaintedBoundChanged, StaleRevision
 from .declared_family import DeclaredFamily
 from .display_order import ChannelSort, NameChannelSort
-from .display_references import DisplayReferences
+from .display_references import CurrentDisplayReferences
 from .external_transcript import ExternalTranscript
 from .field_codec import (
     Differs,
@@ -683,7 +683,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)
@@ -711,7 +711,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(str(root), registry, tuple(declarations)),
             channel_order=catalog.list_order, read_marker_notice=read_marker_notice,
         )
 

src/agent_comms/registry_document.py

@@ -8,7 +8,7 @@
 from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
 from dataclasses import dataclass, field, 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
@@ -198,26 +198,45 @@ def of(plan: FieldPlan) -> type[ThreadSlice]:
                 return policy.value
         raise TypeError(f"{plan.name} declares no thread slice")
 
-    @staticmethod
-    def required[Document: type](document: Document) -> Document:
+    @classmethod
+    def required[Document: type](cls, document: Document) -> Document:
         """Refuse, at class definition, a document field without a slice kind."""
         for plan in FieldCodec.field_plans(document):
             if plan.declared.init and plan.records:
-                ThreadSlice.of(plan)
+                cls.of(plan)
         return document
 
 
-class OwnRow(ThreadSlice[dict]):
+class SnapshotValue[Value](ThreadSlice[Value]):
+    """A declared registry field owns both editable and snapshot copies."""
+
+    @classmethod
+    @abstractmethod
+    def freeze(cls, value: Value) -> object: ...
+
+    @staticmethod
+    def of(plan: FieldPlan) -> type[SnapshotValue]:
+        policy = ThreadSlice.of(plan)
+        if not issubclass(policy, SnapshotValue):
+            raise TypeError(f"{plan.name} declares no snapshot value policy")
+        return policy
+
+
+class OwnRow(SnapshotValue[dict]):
     """The row under the thread's own name."""
 
     @staticmethod
     def cut(raw: dict, canonical: str) -> dict:
         return {canonical: raw[canonical]} if canonical in raw else {}
 
     @classmethod
-    def copy(cls, value: dict) -> dict:
+    def copy(cls, value: Mapping) -> dict:
         return dict(value)
 
+    @classmethod
+    def freeze(cls, value: Mapping) -> Mapping:
+        return MappingProxyType(cls.copy(value))
+
 
 class AliasesOf(OwnRow):
     """The aliases that point at the thread; they also resolve its canonical name."""
@@ -243,7 +262,7 @@ def copy(cls, value: GenerationCounter) -> GenerationCounter:
         return GenerationCounter(value.counter, dict(value.generations))
 
 
-class WholeDocumentValue[Value](ThreadSlice[Value]):
+class WholeDocumentValue[Value](SnapshotValue[Value]):
     """A scalar shared by the whole document and every one-thread view."""
 
     @staticmethod
@@ -254,6 +273,10 @@ def cut(raw: Value, canonical: str) -> Value:
     def copy(cls, value: Value) -> Value:
         return value
 
+    @classmethod
+    def freeze(cls, value: Value) -> Value:
+        return cls.copy(value)
+
 
 class RootPurpose(DeclaredFamily, affix="RootPurpose"):
     """The registry's durable permission to mutate this root."""
@@ -718,16 +741,40 @@ def owner_identity(self) -> OwnerIdentity:
         return OwnerIdentity(self.thread.incarnation, self.owner_generation)
 
 
+@SnapshotValue.required
 @dataclass(frozen=True, slots=True)
 class RegistrySnapshot(RegistryPresence[Mapping[str, ThreadStatus]], RegistryProvenance):
-    purpose: RootPurpose
-    threads: Mapping[str, Thread]
-    statuses: Mapping[str, ThreadStatus]
-    last_seen: Mapping[str, float]
-    aliases: Mapping[str, str]
-    owner_generations: Mapping[str, int]
-    _claimed_generations: Mapping[str, int] = field(repr=False)
-    by_lookup: Mapping[str, Thread] = field(repr=False, compare=False)
+    purpose: Annotated[RootPurpose, SlicePolicy(WholeDocumentValue)]
+    threads: Annotated[Mapping[str, Thread], SlicePolicy(OwnRow)]
+    statuses: Annotated[Mapping[str, ThreadStatus], SlicePolicy(OwnRow)]
+    last_seen: Annotated[Mapping[str, float], SlicePolicy(OwnRow)]
+    aliases: Annotated[Mapping[str, str], SlicePolicy(OwnRow)]
+    owner_generations: Annotated[Mapping[str, int], SlicePolicy(OwnRow)]
+    _claimed_generations: Annotated[Mapping[str, int], SlicePolicy(OwnRow)] = field(repr=False)
+    by_lookup: Mapping[str, Thread] = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self) -> None:
+        """Own each copied value and derive the birth index from the frozen threads."""
+        for plan in FieldCodec.field_plans(type(self)):
+            if plan.declared.init:
+                object.__setattr__(
+                    self, plan.name, SnapshotValue.of(plan).freeze(getattr(self, plan.name)),
+                )
+        indexed: dict[str, Thread] = {}
+        for thread in self.threads.values():
+            lookup = thread.incarnation.require_recorded().lookup
+            if lookup in indexed:
+                raise RelationViolationError("Registry creation identities collide.")
+            indexed[lookup] = thread
+        object.__setattr__(self, "by_lookup", MappingProxyType(indexed))
+        super(RegistrySnapshot, self).__post_init__()
+
+    def __reduce__(self) -> tuple[type[Self], tuple[object, ...]]:
+        """Worker restoration preserves scalar policy and re-enters the constructor."""
+        return type(self), tuple(
+            SnapshotValue.of(plan).copy(getattr(self, plan.name))
+            for plan in FieldCodec.field_plans(type(self)) if plan.declared.init
+        )
 
     def claim_generation(self, name: str) -> int:
         """The allocated input generation follows this registry's alias lineage."""
@@ -756,21 +803,14 @@ def idle_after(self, fence: FinishedTurnFence) -> tuple[Thread, ...]:
     @classmethod
     def capture(cls, document: RegistryDocument) -> RegistrySnapshot:
         """One live registry cut owns the index and all its source maps."""
-        indexed: dict[str, Thread] = {}
-        for thread in document.threads.values():
-            lookup = thread.incarnation.require_recorded().lookup
-            if lookup in indexed:
-                raise RelationViolationError("Registry creation identities collide.")
-            indexed[lookup] = thread
         return cls(
             purpose=document.purpose,
-            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,
         )
 
     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(