P954: reland paired narrow relationship readers and complete controls

src/toad/conversation_kind.py

@@ -300,8 +300,8 @@ def prompt(self, project: Checkout):
 
     async def status_text(self, comms: Comms, *, read_only: bool) -> str:
         def read():
-            references = CurrentDisplayReferences.from_registry(str(comms.root), comms.registry.snapshot())
-            return comms.configuration.answer(self.destination(references).name)
+            snapshot = comms.registry.snapshot()
+            return comms.configuration.answer(self.peer.require_current(snapshot).name)
 
         return (await asyncio.to_thread(read)).status_label
 

src/toad/widgets/comms_chat.py

@@ -114,7 +114,6 @@ async def initialize_view(self) -> None:
                 f"Conversation root {self._bound_root} differs from observed root {shared.root}"
             )
         comms = shared
-        await self.app.coordination_access.current_sidebar()
         self.message_history.initialize_source(WireSource(
             comms, self.conversation, self.project_path, self._me,
             IrcMessageStyle(),

src/toad/widgets/thread_comms.py

@@ -10,6 +10,7 @@
 from toad.navigation_target import NavigationTarget, ThreadTarget
 
 import asyncio
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 
 from agent_comms.display_order import ThreadSort
@@ -44,10 +45,9 @@
 
 
 def entry_target(entry: RelationshipEntry) -> NavigationTarget:
-    person = entry.person
-    if person is None:
-        return NavigationTarget.named(entry.target)
-    return ThreadTarget(person.thread.name)
+    for person in entry.person.present():
+        return ThreadTarget(person.thread.name)
+    return NavigationTarget.named(entry.target)
 
 
 @dataclass
@@ -111,7 +111,9 @@ class RelationshipRow(ThreadStatusRow, actions=(CopyIdentityAction,)):
     available = True
     entry: RelationshipEntry | None = None
 
-    def present(self, entry, model) -> None:
+    def present(
+        self, entry: RelationshipEntry, rows: Mapping[str, ThreadRowModel]
+    ) -> None:
         previous, self.entry = self.entry, entry
         self.available, self.target = entry.available, entry_target(entry)
         if not entry.available:
@@ -120,36 +122,39 @@ def present(self, entry, model) -> None:
                 Content(f"? {entry.target}\n  Unavailable · Ctrl+C copies name"),
                 layout=False,
             )
-        elif entry.person is not None:
-            if (
-                model is not None
-                and model.incarnation == entry.person.thread.incarnation
-            ):
+            self._show_entry_detail(entry)
+            return
+        for person in entry.person.present():
+            model = rows.get(person.thread.name)
+            if model is not None and model.incarnation == person.thread.incarnation:
                 self.workers.cancel_group(self, "relationship-row")
                 self._show_person(entry, model)
             elif previous != entry:
                 self.run_worker(
-                    self._present_person(entry),
+                    self._present_person(entry, person),
                     group="relationship-row",
                     exclusive=True,
                 )
-        elif previous != entry:
+            return
+        if previous != entry:
             self.set_label(entry.target)
-        if entry.person is None:
-            self.tooltip = (
-                Content(f"{entry.target}\n{entry.detail}")
-                if entry.detail
-                else entry.target
-            )
+        self._show_entry_detail(entry)
+
+    def _show_entry_detail(self, entry):
+        self.tooltip = (
+            Content(f"{entry.target}\n{entry.detail}")
+            if entry.detail
+            else entry.target
+        )
 
     def _show_person(self, entry, model):
         self.show(model)
         if entry.detail:
-            self.tooltip = Content(f"{entry.target}\n{entry.detail}")
+            self._show_entry_detail(entry)
 
-    async def _present_person(self, entry):
+    async def _present_person(self, entry, person):
         model = await self.app.preparation.run_thread(
-            ThreadRowModel.of, entry.person, UnobservedUnread()
+            ThreadRowModel.of, person, UnobservedUnread()
         )
         if self.entry == entry:
             self._show_person(entry, model)
@@ -207,7 +212,8 @@ def keys_for_threads(self, names: frozenset[str]) -> frozenset[str]:
         return frozenset(
             entry.target
             for entry in self.model.entries.values()
-            if entry.person is not None and entry.person.thread.name in names
+            for person in entry.person.present()
+            if person.thread.name in names
         )
 
     def new_row(self, key: str):
@@ -217,10 +223,7 @@ def new_row(self, key: str):
     def show_member(self, key: str, row: RelationshipRow) -> None:
         entry = self.model.entries[key]
         threads = self.query_ancestor(ThreadCommsSidebar).model_state.shown.threads.rows
-        row.present(
-            entry,
-            threads.get(entry.person.thread.name) if entry.person is not None else None,
-        )
+        row.present(entry, threads)
 
     def sync_members(self, changed: frozenset[str]) -> None:
         """Reconcile member rows from this group's entries and the tree's derived thread rows."""

src/toad/wire_history.py

@@ -16,7 +16,6 @@
 
 from agent_comms.comms import Comms
 from agent_comms.repo import Checkout
-from agent_comms.display_references import CurrentDisplayReferences
 from agent_comms.historical_views import (
     ConversationHistoryView,
     ConversationObservation,
@@ -301,10 +300,6 @@ class WireSource:
     def viewer(self) -> NamedViewer:
         return self.comms.views.display_viewer(self.project)
 
-    def references(self) -> CurrentDisplayReferences:
-        """Resolve stable destinations through this read's registry cut."""
-        return CurrentDisplayReferences.from_registry(str(self.comms.root), self.comms.registry.snapshot())
-
     def observe(self, seen: ConversationObservation) -> SourceObservation:
         """The observed state owns how changed or unchanged history is drawn."""
         revision = self.conversation.history_view.revision(

tests/textual/test_channel_updates_and_returns.py

@@ -1,5 +1,7 @@
 """A real shared observer updates a shown channel and leaves its parked view alone."""
 
+import asyncio
+
 async def test_live_update_paints_parked_channel_reads_nothing_and_return_shows_the_update(workspace):
     from toad.widgets.comms_chat import CommsChatView, session_thread_name
 
@@ -32,3 +34,64 @@ async def test_live_update_paints_parked_channel_reads_nothing_and_return_shows_
         await ui.until_shows("PARKED UPDATE")
         await ui.until(lambda: bool(access.notification_interest.reads), why="the returned channel to resubscribe")
         assert app.selected_session.query_one(CommsChatView) is chat
+
+
+async def test_first_wire_page_paints_before_pending_sidebar_and_keeps_source_fences(workspace):
+    from agent_comms.launch_root import LaunchRoot
+    from support.roots import installed_wire
+    from toad.preferences import SETTINGS
+    from toad.widgets.channel_prompt import ChannelPrompt
+    from toad.widgets.comms_chat import CommsChatView
+
+    sender = workspace.thread("sender", "first-page")
+    workspace.thread("reader", "first-page")
+    workspace.comms.messaging.send("sender", "#first-page", "FIRST_PAGE_WITHOUT_SIDEBAR")
+    other = installed_wire(workspace.roots.attempt / "other-route")
+    async with workspace.app(size=(113, 34)) as ui:
+        app = ui.app
+        access = app.coordination_access
+        await access.current_sidebar()
+        observed = access.observation.observed()
+        process = observed.process
+        # Join the actual receiver, not a replacement service or fake answer.
+        # Its child keeps running and the real sidebar reply remains in its pipe.
+        await process._reader.aclose()
+        async with asyncio.timeout(20), asyncio.TaskGroup() as work:
+            try:
+                app.saved_settings.toggle(SETTINGS.sidebar.show_archived)
+                sidebar = work.create_task(access.current_sidebar())
+                references = work.create_task(
+                    observed.current_references(access, str(workspace.comms.root)),
+                )
+                await ui.until(lambda: bool(access.sidebar_waiters),
+                               why="the genuine changed-interest sidebar answer to remain pending")
+                opening = work.create_task(ui.open_channel("#first-page", actor="reader"))
+                await ui.until_shows("FIRST_PAGE_WITHOUT_SIDEBAR")
+                chat = app.selected_session.query_one(CommsChatView)
+                history = chat.message_history
+                source = history.page_source
+                assert not history.undrawn
+                assert not sidebar.done() and not references.done()
+                assert not chat.query_one(ChannelPrompt)._candidates
+            finally:
+                # Reattach the same observer's one reader; deliver its original reply.
+                process.start_receiving(observed.receive, observed.receive_failed)
+            mode = await opening
+            answer = await sidebar
+            catalog = await references
+        assert catalog.thread(sender.incarnation).name == sender.name
+        assert answer.require_snapshot().references.thread(sender.incarnation).name == sender.name
+        await ui.until(
+            lambda: any(candidate.name == sender.name
+                        for candidate in chat.query_one(ChannelPrompt)._candidates),
+            why="the genuine channel mention label to finish after its sidebar answer",
+        )
+        assert chat.agent_info.plain == chat.conversation.label(catalog)
+        snapshot = history.source_snapshot()
+        source = history.page_source
+        with LaunchRoot.from_path(other.root).selected():
+            assert not await history.source_is_current(snapshot)
+        assert await history.source_is_current(snapshot)
+        await app.session_navigation.close(mode)
+        assert source.closed
+        assert not await source.route_is_current()

tests/textual/test_conversation_kind.py

@@ -9,9 +9,11 @@
 from agent_comms.comms import Comms
 from agent_comms.active_route import resolve_comms_route
 from support.roots import installed_wire
+from agent_comms.errors import UnregisteredThreadError
 from agent_comms.field_codec import FieldCodec
 from agent_comms.installer.backup import StoreBackup
 from agent_comms.message_page import PageBounds
+from agent_comms.thread_creation import RecordedCreationTime
 from agent_comms.threads import Thread
 
 from toad.constants import ALL_COMMS_TARGET
@@ -107,11 +109,14 @@ def test_kind_and_owner_are_part_of_identity_but_channel_address_is_shared(copie
     )
 
 
-def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copied_conversations):
+async def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copied_conversations):
     comms, viewer = copied_conversations
     direct = DirectTarget("peer").prepare(comms, viewer).conversation
     key = CommsViewKey(str(comms.root), "owner", direct)
     scope = direct.input_scope
+    assert await direct.status_text(comms, read_only=False) == (
+        comms.configuration.answer("peer").status_label
+    )
     comms.registry.rename("peer", "renamed")
     reopened = DirectTarget("renamed").prepare(comms, viewer).conversation
     references = comms.views.viewer_snapshot(Checkout(comms.root.parent)).references
@@ -121,6 +126,9 @@ def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copie
     assert direct.label(references) == "@renamed"
     assert direct.send_target().selected_name(comms.registry.snapshot()) == "renamed"
     assert HistoryTarget.for_conversation(direct, references) == DirectTarget("renamed")
+    assert await direct.status_text(comms, read_only=False) == (
+        comms.configuration.answer("renamed").status_label
+    )
 
 
 def test_recreated_name_cannot_resolve_the_retired_dm(copied_conversations):
@@ -138,6 +146,27 @@ def test_recreated_name_cannot_resolve_the_retired_dm(copied_conversations):
         direct.destination(references)
 
 
+@pytest.mark.parametrize("recreate", (False, True))
+async def test_dm_status_refuses_deleted_or_recreated_peer(copied_conversations, recreate):
+    comms, viewer = copied_conversations
+    original = comms.registry.require("peer")
+    direct = DirectTarget("peer").prepare(comms, viewer).conversation
+    comms.registry.unregister("peer")
+    comms.threads.delete("peer")
+    if recreate:
+        comms.registry.declare(Thread(
+            "peer", frozenset(), Checkout(comms.root.parent),
+            created_at=RecordedCreationTime(original.created_at + 1),
+        ))
+        replacement = DirectTarget("peer").prepare(comms, viewer).conversation
+        assert replacement != direct
+        assert await replacement.status_text(comms, read_only=False) == (
+            comms.configuration.answer("peer").status_label
+        )
+    with pytest.raises(UnregisteredThreadError):
+        await direct.status_text(comms, read_only=False)
+
+
 def test_history_picker_and_channel_status_use_the_same_declared_value(copied_conversations):
     from toad.screens.historical_sessions import HistoricalSessions
 

tests/textual/test_thread_relationship_sidebar.py

@@ -76,6 +76,48 @@ async def test_relationship_groups_follow_core_edits_tools_sorting_and_renames(w
         assert peer_group.entries["owner"].target == "owner"
         assert peer_group.entries["owner"].detail == "Review the sidebar"
 
+        # Use Core's actual entries and models in this already-open panel.
+        collaboration = tree.groups[CollaboratingRelationshipGroup]
+        person_entry = collaboration.model.entries["peer"]
+        person, = person_entry.person.present()
+        person_row = collaboration.new_row("peer")
+        channel_group = tree.groups[OutboundRelationshipGroup]
+        channel_entry = channel_group.model.entries["#team"]
+        channel_row = channel_group.new_row("#team")
+        await collaboration.member_container.mount(person_row)
+        await channel_group.member_container.mount(channel_row)
+        try:
+            person_row.present(person_entry, {})
+            await ui.until(
+                lambda: person_row.thread_row is not None,
+                why="an uncached person to finish real model preparation",
+            )
+            assert person_row.thread_row is not None
+            assert person_row.thread_row.incarnation == person.thread.incarnation
+            assert person_row.thread_row.name == "peer"
+            assert "Review the sidebar" in str(person_row.tooltip)
+
+            cached = tree.model_state.shown.threads.rows["peer"]
+            person_row.retire_thread_row()
+            person_row.present(person_entry, {"peer": cached})
+            assert person_row.thread_row is cached
+
+            assert tuple(channel_entry.person.present()) == ()
+            channel_row.present(channel_entry, tree.model_state.shown.threads.rows)
+            await ui.until(
+                lambda: "#team" in ui.painted(channel_row),
+                why="the nominally absent person to keep its channel label",
+            )
+            assert channel_row.thread_row is None
+            assert channel_row.entry == channel_entry
+            assert str(channel_row.tooltip) == (
+                f"{channel_entry.target}\n{channel_entry.detail}"
+                if channel_entry.detail else channel_entry.target
+            )
+        finally:
+            await person_row.remove()
+            await channel_row.remove()
+
         invoke_tool(comms, "comms_collaboration", {"action": "remove", "peer": "owner"},
                     caller=ThreadCaller("peer"))
         await ui.until(lambda: not tree.groups[CollaboratingRelationshipGroup].model.entries