diff --git a/src/agent_comms/display_references.py b/src/agent_comms/display_references.py index 7925401011..5d2026172b 100644 --- a/src/agent_comms/display_references.py +++ b/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) diff --git a/src/agent_comms/errors.py b/src/agent_comms/errors.py index acbf3e9ec7..fa7f5065d9 100644 --- a/src/agent_comms/errors.py +++ b/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.""" diff --git a/src/agent_comms/presentation.py b/src/agent_comms/presentation.py index faf0059e1b..98eddb5d4a 100644 --- a/src/agent_comms/presentation.py +++ b/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, ) diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index b038dca498..dc32cccad0 100644 --- a/src/agent_comms/registry_document.py +++ b/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( diff --git a/src/agent_comms/route_selection.py b/src/agent_comms/route_selection.py index a99f1d95d7..69f5b702a4 100644 --- a/src/agent_comms/route_selection.py +++ b/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 diff --git a/src/agent_comms/thread_presentation.py b/src/agent_comms/thread_presentation.py index 8987ff88bc..06871e45a6 100644 --- a/src/agent_comms/thread_presentation.py +++ b/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.""" diff --git a/src/agent_comms/ui_model/status.py b/src/agent_comms/ui_model/status.py index 121bf207a0..ca4d19e142 100644 --- a/src/agent_comms/ui_model/status.py +++ b/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( diff --git a/tests/test_sidebar_observation.py b/tests/test_sidebar_observation.py index 13fddb768f..62e1905dce 100644 --- a/tests/test_sidebar_observation.py +++ b/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 diff --git a/tests/test_thread_presentation.py b/tests/test_thread_presentation.py index 4c7ef7f6cf..a52a53a0bb 100644 --- a/tests/test_thread_presentation.py +++ b/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"),