Preserve sends and original DM identities through transcript publication

docs/agents/TESTS.md

@@ -210,7 +210,7 @@ The starting point was `main` on 2026-10-11 against the published runtime: 363 i
 | `channel_any_mode_ui_pilot` | 1 | rewritten | `tests/textual/test_channel_any_mode.py` | The menu item now carries Core's one label, "Toggle member activity"; the old two labels were Toad's own. |
 | `channel_display_basis_pilot` | 1 | rewritten | `tests/app/test_channel_display_basis.py` | Painted rows are asked as screen text plus Core seen_sequences; has_older/painted_keys became 'channel 9 not shown'. |
 | `channel_mentions_pilot` | 1 | rewritten | `tests/app/test_channel_mentions.py` | Asserted before the mention candidates had loaded; waits for the list. |
-| `channel_participant_publication_pilot` | 1 | rewritten | `tests/textual/test_channel_participants.py` | (ChannelParticipants widget, sys.monitoring count of prepare_participants on worker vs UI thread, kept as counts). Dropped: the ThreadView.presentation counter (no longer called by the participants path) and the loop-gap measurement (never asserted). |
+| `channel_participant_publication_pilot` | 1 | superseded | `tests/textual/test_conversation_value.py` | ChannelParticipants and its preparation method were deleted. The current ChannelActivity receives actual member changes through the shared sidebar observer; the conversation journey checks that new members arrive. |
 | `channel_scope_expansion_pilot` | 1 | rewritten | `tests/app/test_channel_scope_expansion.py`, `tests/textual/test_channel_scope_rows.py` | App test waits on screen text with no explicit refresh; textual test keeps the explicit refresh and row/has_older checks. Both pass; the old AssertionError did not reproduce (old file used runtime_fixture refresh_comms). |
 | `channel_visibility_observation_pilot` | 1 | rewritten | `tests/textual/test_hidden_channel_refresh.py` | (hidden view's refresh callback, reader pending set). The unused checkpoint_checks helper in the old file was dead code and is not carried over. |
 | `checkpoint_source_pivot_pilot` | 1 | rewritten | `tests/textual/test_checkpoint_first_source.py` | Expected the saved answer as a response widget inside the history; history is drawn as lines, so the test now requires no live response widget left and the answer drawn exactly once. The covered live response is retired again since the answers-drawn-twice fix (agent_response.py). |
@@ -293,7 +293,7 @@ The starting point was `main` on 2026-10-11 against the published runtime: 363 i
 | `offscreen_tool_hydration_pilot` | 1 | rewritten | `tests/textual/test_offscreen_tool_diff.py` | kept the not-yet-mounted-then-hydrated-on-scroll assertions. Dropped the TOAD_BENCH_EAGER_TOOLS environment switch variant (a benchmark mode that patched _visible_in_window) and the printed millisecond figures. |
 | `owner_navigation_pilot` | 1 | open defect | `tests/app/test_channel_tab_sharing.py` | A channel has one tab for all sessions (deliberate). What the tab does for a second session is open defect shared-channel-tab. |
 | `owner_poll_latency_pilot` | 1 | deleted | — | Only printed timings. |
-| `owner_reader_reuse_pilot` | 1 | rewritten | `tests/app/test_owner_reader_reuse.py` | (workspace fixture, one test also starts the app). Three tests: shared reader and root change, replaced owner receives no request, bound reader is the app's own service. Uses support.agents.detached and support.coordination.registered_coordination_fact. |
+| `owner_reader_reuse_pilot` | 1 | rewritten | `tests/app/test_owner_reader_reuse.py` | Two real-app checks: agents share the application's existing service and another explicit root is refused. Deleted the obsolete per-agent reader reconstruction and root-rebinding expectations. |
 | `path_filter_pilot` | 1 | rewritten | `tests/model/test_path_filter.py`, `tests/textual/test_path_filter_tree.py` | PathFilter assertions go to model; tree root change and hidden-session deferred reload go to Textual. Hidden session publishes on the MainScreen captured as app.selected_session (WorkspaceScreen has no publish_core); return is ui.select_tab. |
 | `pending_thread_open_pilot` | 1 | rewritten (needs native package) | `tests/native/test_thread_open_during_slow_load.py`, `tests/support/peers/delayed_load.py`, `tests/native/test_native_journey.py::test_a_second_thread_opened_from_the_app_answers_its_first_message` | The scripted peer (was acp_navigation_server.py) claims a Core thread but publishes no input queue; |
 | `permission_pending_mount_pilot` | 1 | rewritten | `tests/textual/test_permission_requests.py`, `tests/app/test_permission_across_tabs.py` | The hand-rebinding cases are replaced by app tests through a real scripted agent and real tab switches; all pass. |
@@ -327,7 +327,7 @@ The starting point was `main` on 2026-10-11 against the published runtime: 363 i
 | `stream_style_work_pilot` | 1 | rewritten | `tests/textual/test_old_output_is_not_restyled.py` | kept zero-restyle counts for append and sidebar toggle; dropped printed millisecond figures. |
 | `t8_mounted_pilot` | 1 | rewritten | `tests/textual/test_command_pane_and_resume_modal.py`, `tests/model/test_saved_sessions_of_this_machine_read_back_typed.py` | Command pane environment and resume modal to Textual (bare App now ThemedApp, saved AgentDefinition has a name). The read-only check of this machine's ~/.local/state/toad/toad.db is its own model test and skips when absent; |
 | `tab_order_pilot` | 1 | rewritten | `tests/app/test_tab_order.py` | Selected sessions through Textual's switch_mode; asks the harness now. |
-| `terminal_crash_pilot` | 1 | rewritten (needs a terminal emulator) | `tests/terminal/test_terminal_crash.py`, `tests/support/terminal_crash_app.py` | Real PTY via support.terminal; script app moved into support; asserts exit 1, one 0600 Terminal_crash file naming fail_after_paint, no locals, error echoed on the terminal. |
+| `terminal_crash_pilot` | 1 | rewritten (needs a terminal emulator) | `tests/terminal/test_terminal_crash.py`, `tests/support/terminal_crash_app.py` | Real PTY: a mounted handler failure is displayed, then a channel opens, accepts a draft and quits cleanly. Fatal startup/teardown and recoverable mounted errors have different contracts. |
 | `test_agent_configuration` | 1 | rewritten | `tests/model/test_agent_configuration.py` | Model layer; the fake that collected post_message is a detached Agent whose events stream is subscribed (ConfigurationChanged from the agent is the old 'message.agent is agent'). |
 | `test_fixture_journal_retention` | 13 | deleted | — | 13 tests of a cold-storage utility for record files that nothing used. |
 | `test_shared_acp_transport` | 2 | rewritten | `tests/model/test_sessions_share_one_stdio_process.py`, `tests/textual/test_workspace_tabs_share_one_agent_connection.py`, `tests/support/shared_stdio.py` | Split into the Agent-objects test (model) and the two-tab app test (textual). claim_thread lost start_at_latest (inbox now starts at creation), dropped. The app test needs AGENT_COMMS_AGENT_MODELS=test/one via monkeypatch for the model catalogue. |

src/toad/acp/agent.py

@@ -274,11 +274,11 @@ async def update_project(self, path: str) -> str:
             raise ValueError("Project changes require an agent-comms thread.")
         authority = ClientSessionRequest(self, self.session_id)
         coordination = self.coordination
-        async with self.controller.transcripts.bind(coordination.wire_root) as comms:
-            authority.require()
-            result = await asyncio.to_thread(
-                comms.threads.set_project, coordination.thread.name, path,
-            )
+        comms = await self.controller.transcripts.service(coordination.wire_root)
+        authority.require()
+        result = await asyncio.to_thread(
+            comms.threads.set_project, coordination.thread.name, path,
+        )
         authority.require()
         self.coordination = replace(self.coordination, worktree=result.current)
         self.project_root_path = Path(result.current)

src/toad/acp/agent_controller.py

@@ -210,9 +210,9 @@ async def _submit(self, prompt, command, authority, queue_scope, project):
             coordination = self.coordination
             if coordination is not None:
                 self.require_owner(coordination, authority)
-                async with self.transcripts.bind(coordination.wire_root) as comms:
-                    origin = await asyncio.to_thread(
-                        self.agent.queue_attachment.capture_human_input, comms, queue_scope)
+                comms = await self.transcripts.service(coordination.wire_root)
+                origin = await asyncio.to_thread(
+                    self.agent.queue_attachment.capture_human_input, comms, queue_scope)
                 self.require_owner(coordination, authority)
                 command = replace(command, origin=origin)
             content = await asyncio.to_thread(build_prompt, project, prompt)
@@ -319,13 +319,13 @@ async def request_owner(self, method: str, **params):
         coordination = self.coordination
         authority = ClientSessionRequest(self.agent, self.agent.session_id)
         self.require_owner(coordination, authority)
-        async with self.transcripts.bind(coordination.wire_root) as comms:
-            def resolve():
-                owner = comms.registry.require(coordination.thread.name)
-                if owner.incarnation != coordination.thread:
-                    raise ValueError('The owner incarnation changed while preparing the request.')
-                return RuntimeProxy(OwnerRequestContext(comms), owner.name, socket_path(comms.root, owner.require_process().pid))
-            proxy = await asyncio.to_thread(resolve)
+        comms = await self.transcripts.service(coordination.wire_root)
+        def resolve():
+            owner = comms.registry.require(coordination.thread.name)
+            if owner.incarnation != coordination.thread:
+                raise ValueError('The owner incarnation changed while preparing the request.')
+            return RuntimeProxy(OwnerRequestContext(comms), owner.name, socket_path(comms.root, owner.require_process().pid))
+        proxy = await asyncio.to_thread(resolve)
         try:
             self.require_owner(coordination, authority)
             result = await proxy.request(method, **params)

src/toad/acp/agent_session.py

@@ -1,12 +1,14 @@
 """ACP session handshake, durable metadata and reconnection."""
 from __future__ import annotations
+
+from agent_comms.errors import RouteChangedError
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.child_process import join_retirement
 import asyncio
 import os
 from dataclasses import dataclass, replace
 from collections.abc import Callable
-from agent_comms.route_selection import RouteChanged, RouteSelection
+from agent_comms.route_selection import RouteSelection
 from agent_comms.thread_presentation import SessionDisconnected
 import toad
 from toad import constants
@@ -340,7 +342,7 @@ async def reconnect(self, plan: SessionLoadPlan | None = None) -> None:
             try:
                 await asyncio.to_thread(
                     RouteSelection.capture, self.agent.controller.transcripts.access.launch_root.path)
-            except RouteChanged as error:
+            except RouteChangedError as error:
                 self.startup_failed(f"Reconnect blocked before restart: {error}")
                 return
             await self.agent.stop()
@@ -506,12 +508,12 @@ async def rename_coordination(self, display_name: str) -> None:
 
         authority = ClientSessionRequest(self.agent, self.agent.session_id)
         owner_pid = self.agent.coordination.owner_pid
-        async with self.agent.controller.transcripts.bind(wire_root) as comms:
-            authority.require()
-            result = await asyncio.to_thread(
-                comms.threads.rename_managed_thread,
-                thread, display_name, owner_pid=owner_pid,
-            )
+        comms = await self.agent.controller.transcripts.service(wire_root)
+        authority.require()
+        result = await asyncio.to_thread(
+            comms.threads.rename_managed_thread,
+            thread, display_name, owner_pid=owner_pid,
+        )
         authority.require()
         self.agent.coordination = replace(
             self.agent.coordination,

src/toad/acp/comms_updates.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from agent_comms.errors import RouteChangedError
+
 from dataclasses import dataclass
 from typing import Any
 
@@ -121,12 +123,11 @@ def input_started(self, update: InputStartedUpdate) -> None:
     @handles(CoordinationChangedUpdate)
     def coordination_changed(self, update: CoordinationChangedUpdate) -> None:
         from pathlib import Path
-        from agent_comms.route_selection import RouteChanged
 
         agent = self.agent
         root = agent.controller.transcripts.access.launch_root.path
         if Path(update.wire_root).resolve() != root.resolve():
-            raise RouteChanged("ACP owner reported a different coordination root")
+            raise RouteChangedError("ACP owner reported a different coordination root")
         agent.coordination = update
         agent.project_root_path = Path(update.worktree)
         agent.context_measurement = ContextMeasurement.saved(update.context_usage)

src/toad/acp/transcript_reader.py

@@ -13,7 +13,6 @@
 from __future__ import annotations
 
 import asyncio
-from contextlib import asynccontextmanager
 
 from agent_comms.route_selection import RouteSelection
 from agent_comms.acp_extension import TranscriptDraftUpdate, TranscriptSnapshotUpdate
@@ -41,12 +40,8 @@ async def thread_presentation(self, root: str, thread: str):
         """Read one thread presentation through this reader's bound service."""
         from toad.owner_preparation import read_thread_presentation
 
-        async with self.bind(root) as reader:
-            return await asyncio.to_thread(read_thread_presentation, reader, thread)
-
-    @asynccontextmanager
-    async def bind(self, root: str):
-        yield await self.service(root)
+        reader = await self.service(root)
+        return await asyncio.to_thread(read_thread_presentation, reader, thread)
 
     async def publication(self, update: TranscriptSnapshotUpdate) -> TranscriptSnapshotUpdate:
         """The published page while its witness is current, else the current page."""

src/toad/app.py

@@ -202,11 +202,13 @@ async def _close_all(self) -> None:
         # View release can fail. The application still owns its observation
         # pipe and process pools; an Unmount event may never follow that error.
         async with AsyncResourceRelease() as release:
-            release.push_async_callback(self.render_processes.aclose)
-            release.push_async_callback(self.navigation_reader.aclose)
             release.callback(self.terminal_attention.close)
             release.callback(self.colours.stop)
             release.push_async_callback(self.coordination_access.close)
+            # Accepted preparation survives its UI waiter. Keep the observer
+            # available until those workers and their read replies finish.
+            release.push_async_callback(self.render_processes.aclose)
+            release.push_async_callback(self.navigation_reader.aclose)
             release.push_async_callback(super()._close_all)
             release.push_async_callback(self.coordination_access.stop_observing)
             release.push_async_callback(self.thread_actions.close)

src/toad/app_node.py

@@ -0,0 +1,16 @@
+"""Native nodes bound to Toad's application contract."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from textual.dom import DOMNode
+
+if TYPE_CHECKING:
+    from toad.app import ToadApp
+
+
+class ToadNode(DOMNode):
+    """Keep the application refinement on one shared native parent."""
+
+    app: ToadApp