diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py
index a9040b693..b3b75b4ec 100644
--- a/src/toad/acp/agent_controller.py
+++ b/src/toad/acp/agent_controller.py
@@ -21,7 +21,6 @@
     encode_request,
 )
 from agent_comms.acp_failure import ACPFailure, RequestFailure
-from agent_comms.coordination_errors import Refused
 from agent_comms.runtime import RuntimeConnection, socket_path
 from agent_comms.transcripts import TranscriptReadIdentity
 from agent_comms.order_tools import InputDeliveryRead
@@ -411,8 +410,6 @@ def resolve():
             result = await proxy.request(method, **params)
             self.require_owner(coordination, authority)
             return result
-        except (RuntimeError, Refused) as error:
-            raise ValueError(str(error)) from error
         finally:
             await proxy.close()
 
diff --git a/src/toad/block_navigation.py b/src/toad/block_navigation.py
index 0bb61afae..027cdb3b9 100644
--- a/src/toad/block_navigation.py
+++ b/src/toad/block_navigation.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
+from collections.abc import Iterable
 from functools import cached_property
 from typing import TYPE_CHECKING, Callable, ClassVar, cast
 
@@ -15,6 +16,7 @@
 
 from toad.block_content import BlockContent
 from toad.frame_presentation import AnimationEffects
+from toad.notification_status import MessageQueueEntry
 from toad.widgets.presentation_window import NewerEnd, OlderEnd, PageEnd
 
 if TYPE_CHECKING:
@@ -402,6 +404,10 @@ def notification_identity_for(self, references: frozenset[MessageReference]) ->
         """Live entries have no captured BUS display cut; source rows override with theirs."""
         return NoDisplaySelection()
 
+    def queue_entries(self) -> Iterable[MessageQueueEntry]:
+        """Live entries supply no captured BUS delivery rows; source rows override with theirs."""
+        return ()
+
     @cached_property
     def block_cursor(self) -> BlockCursor:
         return AtomicBlockCursor(self)
diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py
index ac3891d0c..3e442566a 100644
--- a/src/toad/conversation_submission.py
+++ b/src/toad/conversation_submission.py
@@ -23,7 +23,7 @@
 from toad.command_catalog import CommandCatalog
 from toad.acp.agent import Agent
 from toad.acp.queue_attachment import QueueActionRefused
-from toad.submission_queue import QueueEvidence
+from toad.submission_queue import QueueEvidence, QueueSource
 from toad.core.input_events import UserInputSubmitted
 from toad.core_event_carrier import CoreEventMessage
 from toad.input_decision import InputDecision, InputRefused
@@ -334,10 +334,11 @@ def reset(self):
         self.publish_pending()
 
     def send_now(self):
-        source = self.queue_source
+        source = self.view.queue_source
+        presentation = self.view.agent_binding.presentation
         return self.view.run_worker(
-            partial(self._send_now, source, self.view.agent_binding.presentation,
-                    source.queue_scope(self.view.agent_binding.presentation)),
+            partial(self._send_now, source, presentation,
+                    source.queue_scope(presentation)),
             group="send-queued-now", exclusive=True,
         )
 
diff --git a/src/toad/goal_interaction.py b/src/toad/goal_interaction.py
index 433b4f7ed..37cfa8418 100644
--- a/src/toad/goal_interaction.py
+++ b/src/toad/goal_interaction.py
@@ -5,6 +5,8 @@
 from typing import ClassVar, TYPE_CHECKING
 from weakref import ref
 
+from agent_comms.coordination_errors import Refused
+from agent_comms.errors import GoalRefused
 from agent_comms.nominal.declared_family import DeclaredFamily
 from agent_comms.goal_actions import ClearGoalAction, GoalAction
 from agent_comms.goal_presentation import ControlledExecution, GoalExecution
@@ -103,7 +105,7 @@ async def apply(cls, session):
 
         async def save(text: str) -> None:
             if not session.owns(agent):
-                raise ValueError("The goal presentation changed; reopen its editor.")
+                raise GoalRefused("The goal presentation changed; reopen its editor.")
             await session.write(agent, agent.edit_goal(goal, text))
 
         observed = await view.app.preparation.run_thread(view.app.coordination_access.observation.observed)
@@ -159,15 +161,18 @@ async def activate(self, action: type[GoalInteraction]) -> None:
         if not action.enabled(self):
             view.feedback.flash(view.goal_display.heading(view.goal_execution))
             return
-        await action.apply(self)
+        try:
+            await action.apply(self)
+        except Refused as error:
+            if not self.owns(agent):
+                raise
+            view.app.show_error(error, "Goal action failed")
 
     async def change(self, action: type[GoalAction], text: str = "") -> None:
         view = self.view
-        if not view.is_attached:
-            raise ValueError("The goal presentation changed; reopen its controls.")
         agent = view.agent_binding
         if not self.owns(agent):
-            raise ValueError("The goal presentation changed; reopen its controls.")
+            raise GoalRefused("The goal presentation changed; reopen its controls.")
         # The existing Agent boundary encodes the declared goal operation.
         # Goal state governs continuation, never interruption of an active turn.
         await self.write(agent, agent.update_goal(action, text))
@@ -177,15 +182,10 @@ async def change(self, action: type[GoalAction], text: str = "") -> None:
     async def write(self, agent, operation):
         """Every write settles by reading the owner, including rejected writes."""
         try:
-            try:
-                return await operation
-            finally:
-                if self.owns(agent):
-                    await self.view.goal_observation.refresh()
-        except (OSError, ValueError) as error:
+            return await operation
+        finally:
             if self.owns(agent):
-                raise ValueError(self.view.goal_display.action_failure(error)) from error
-            raise
+                await self.view.goal_observation.refresh()
 
     def close(self) -> None:
         modal = self.modal
diff --git a/src/toad/screens/goal_details.py b/src/toad/screens/goal_details.py
index 8a6974487..6788204c4 100644
--- a/src/toad/screens/goal_details.py
+++ b/src/toad/screens/goal_details.py
@@ -1,23 +1,18 @@
 """A full-height, scrollable projection of a persistent goal."""
-from toad.core_event_carrier import CoreEventReceiver, CoreEventMessage
-from metaclass_registry.mro_dispatch import handles
-
 from datetime import datetime
 
 from agent_comms.goal_history import RecordedGoalHistory
 from textual.app import ComposeResult
 from textual.containers import Vertical, VerticalScroll
-from toad.screens.modal import DismissibleScreen
+from toad.screens.source_modal import SourceModal
 from toad.widget_actions import DismissAction
 from textual.widgets import Button, Static
 
-from toad.navigation_target import SelectTarget
 from toad.goal_interaction import GoalSession
-from toad.project_path_owner import ProjectPathOwner
 from toad.widgets.goal_text import GoalText
 
 
-class GoalDetails(CoreEventReceiver, ProjectPathOwner, DismissibleScreen[None]):
+class GoalDetails(SourceModal):
     AUTO_FOCUS = "#goal-document"
     DEFAULT_CSS = """
     GoalDetails { align: center middle; background: $background 40%; }
@@ -37,16 +32,8 @@ def __init__(self, source: GoalSession, *, history: RecordedGoalHistory):
         self.failures = history.failures
 
     @property
-    def project_root(self):
-        return ProjectPathOwner.containing(self.source.view).project_root
-
-    @property
-    def coordination_root(self):
-        return ProjectPathOwner.containing(self.source.view).coordination_root
-
-    @property
-    def link_context_events(self):
-        return ProjectPathOwner.containing(self.source.view).link_context_events
+    def source_view(self):
+        return self.source.view
 
     @property
     def goal_display(self):
@@ -109,9 +96,3 @@ def _update_current(self) -> None:
         progress = self.query_one("#goal-current-progress", GoalText)
         progress.update_goal_text(f"Progress: {goal.progress}")
         progress.display = goal.shows_progress
-
-    @handles(SelectTarget)
-    def open_target(self, event: SelectTarget, message: CoreEventMessage) -> None:
-        message.stop()
-        self.dismiss(None)
-        self.source.view.publish_core(event)
diff --git a/src/toad/screens/goal_edit.py b/src/toad/screens/goal_edit.py
index 79f72a6b0..6a0edf303 100644
--- a/src/toad/screens/goal_edit.py
+++ b/src/toad/screens/goal_edit.py
@@ -80,7 +80,7 @@ async def save(self, text: str) -> None:
         save_button.disabled = True
         try:
             await self.on_save(text.strip())
-        except (Refused, ValueError) as error:
+        except Refused as error:
             self.query_one("#goal-edit-error", Static).update(str(error))
         else:
             self.dismiss(text.strip())
diff --git a/src/toad/screens/queue.py b/src/toad/screens/queue.py
index 816ef9ece..75c1f707a 100644
--- a/src/toad/screens/queue.py
+++ b/src/toad/screens/queue.py
@@ -1,4 +1,5 @@
 """Inspect the captured queue source through the actual conversation view."""
+from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
 
@@ -11,13 +12,12 @@
 
 from toad.acp.queue_attachment import QueueActionRefused
 from toad.agent_presentation import AgentPresentation
+from toad.conversation_submission import ConversationSubmissions
 from toad.line_blocks import SpanColumns
-from toad.screens.modal import DismissibleScreen
-from toad.submission_queue import QueueEvidence, QueueSource
+from toad.screens.source_modal import SourceModal
 from toad.widgets.tool_content import ToolContent
 
-
-class QueueInspection(DismissibleScreen[None]):
+class QueueInspection(SourceModal):
     """One screen captures the source, presentation and scope it actually displays."""
 
     DEFAULT_CSS = """
@@ -28,11 +28,16 @@ class QueueInspection(DismissibleScreen[None]):
     QueueInspection #queue-error { height: auto; color: $error; }
     """
 
-    def __init__(self, source: QueueSource, evidence: QueueEvidence, changed: Callable[[], None]) -> None:
+    def __init__(self, submissions: ConversationSubmissions) -> None:
         super().__init__()
-        self.source, self.evidence = source, evidence
-        self.scope = source.queue_scope(evidence.presentation)
-        self.changed = changed
+        self.submissions = submissions
+        self.source = submissions.view.queue_source
+        self.evidence = submissions.queue_evidence
+        self.scope = self.source.queue_scope(self.evidence.presentation)
+
+    @property
+    def source_view(self):
+        return self.submissions.view
 
     def compose(self) -> ComposeResult:
         with Vertical():
@@ -55,7 +60,7 @@ async def apply_action(self, action: Callable[[AgentPresentation, QueueScope], A
         except (QueueActionRefused, RequestError) as error:
             self.query_one("#queue-error", Static).update(str(error))
             return
-        self.changed()
+        self.submissions.publish_pending()
         self.dismiss()
 
     @on(Button.Pressed, "#queue-send-now")
diff --git a/src/toad/screens/source_modal.py b/src/toad/screens/source_modal.py
new file mode 100644
index 000000000..621c06c31
--- /dev/null
+++ b/src/toad/screens/source_modal.py
@@ -0,0 +1,35 @@
+"""Rich modals retain their source's project and navigation capabilities."""
+from __future__ import annotations
+
+from abc import abstractmethod
+
+from metaclass_registry.mro_dispatch import handles
+from toad.core_event_carrier import CoreEventMessage, CoreEventReceiver
+from toad.navigation_target import SelectTarget
+from toad.project_path_owner import ProjectPathOwner
+from toad.screens.modal import DismissibleScreen
+
+class SourceModal(CoreEventReceiver, ProjectPathOwner, DismissibleScreen[None]):
+    """Rich modal content keeps the original view's project and navigation."""
+
+    @property
+    @abstractmethod
+    def source_view(self) -> CoreEventReceiver: ...
+
+    @property
+    def project_root(self):
+        return ProjectPathOwner.containing(self.source_view).project_root
+
+    @property
+    def coordination_root(self):
+        return ProjectPathOwner.containing(self.source_view).coordination_root
+
+    @property
+    def link_context_events(self):
+        return ProjectPathOwner.containing(self.source_view).link_context_events
+
+    @handles(SelectTarget)
+    def open_target(self, event: SelectTarget, message: CoreEventMessage) -> None:
+        message.stop()
+        self.dismiss(None)
+        self.source_view.publish_core(event)
diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py
index 8c3d57bf9..8186e6c20 100644
--- a/src/toad/submission_queue.py
+++ b/src/toad/submission_queue.py
@@ -13,9 +13,9 @@
 
 from toad.acp.queue_attachment import QueueActionRefused
 from toad.agent_presentation import AgentPresentation
+from toad.block_navigation import ConversationEntry
 from toad.line_blocks import MutedRole, Span
 from toad.notification_status import DetailLines, MessageQueueEntry
-from toad.widgets.message_notifications import SourceNotifications
 
 
 @dataclass(frozen=True)
@@ -25,7 +25,7 @@ class QueueEvidence:
     presentation: AgentPresentation
     delivering: tuple[str, ...]
     transport: str
-    read_targets: Callable[[], tuple[SourceNotifications, ...]]
+    read_targets: Callable[[], tuple[ConversationEntry, ...]]
 
     @cached_property
     def entries(self) -> tuple[MessageQueueEntry, ...]:
diff --git a/src/toad/toad.tcss b/src/toad/toad.tcss
index d4a7b3079..84ff6836e 100644
--- a/src/toad/toad.tcss
+++ b/src/toad/toad.tcss
@@ -148,6 +148,15 @@ HelpPanel {
 
 
 
+/* Notifications reserve their own row instead of owning clicks over the editor. */
+ToastRack {
+    layer: default;
+    dock: none;
+    split: bottom;
+    max-height: 25vh;
+    margin-bottom: 0;
+}
+
 Toast {    
     margin: 0 0;
     padding: 0 1;
diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py
index dd14c4757..91e2c5093 100644
--- a/src/toad/widgets/conversation.py
+++ b/src/toad/widgets/conversation.py
@@ -1693,8 +1693,7 @@ def queue_observation_changed(self, event: QueueEvidenceChanged) -> None:
         self.queue_evidence_changed()
 
     def open_queue_menu(self) -> None:
-        self.app.push_screen(QueueInspection(
-            self.queue_source, self.submissions.queue_evidence, self.submissions.publish_pending))
+        self.app.push_screen(QueueInspection(self.submissions))
 
     @handles(PreferenceChanged)
     async def _settings_changed(self, event: PreferenceChanged) -> None:
diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py
index 7a1bee21a..bcb8dc28f 100644
--- a/src/toad/widgets/message_notifications.py
+++ b/src/toad/widgets/message_notifications.py
@@ -11,10 +11,12 @@
 from textual.content import Content
 from textual.widgets import Collapsible
 
+from agent_comms.maintenance_barrier import MaintenanceHeldError
 from agent_comms.notification_reads import LiveNotificationRead
 from agent_comms.ui_model.observation import ReadNotifications
 from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection
 from agent_comms.message_reference import MessageReference
+from toad.app_node import ToadNode
 from toad.messages import QueueEvidenceChanged
 
 from toad.notification_status import (
@@ -93,11 +95,17 @@ def start(self, marker: VisibleReadMarking) -> None:
 class ReadWork(PaintedRead):
     work: Awaitable[None]
 
+    async def run(self, marker: VisibleReadMarking) -> None:
+        try:
+            await self.work
+        except MaintenanceHeldError as error:
+            marker.app.show_error(error, "Read not marked", severity="warning")
+
     def start(self, marker: VisibleReadMarking) -> None:
-        marker._read_worker = marker.run_worker(self.work, group="painted-read")
+        marker._read_worker = marker.run_worker(self.run(marker), group="painted-read")
 
 
-class VisibleReadMarking:
+class VisibleReadMarking(ToadNode):
     """One painted-read operation at a time, for pagers and native/CLI sessions."""
 
     def __init__(self, *args, **kwargs):
diff --git a/src/toad/widgets/outgoing_message.py b/src/toad/widgets/outgoing_message.py
index 9d53f56f5..b26e94d99 100644
--- a/src/toad/widgets/outgoing_message.py
+++ b/src/toad/widgets/outgoing_message.py
@@ -14,7 +14,7 @@
 from toad.widgets.message_notifications import MessageNotifications
 from toad.widgets.linked_rows import MessageRowBlock
 from toad.widgets.message_row_owner import MessageRowOwner
-from toad.notification_status import CheckedNotifications, NotificationStatus
+from toad.notification_status import CheckedNotifications, MessageQueueEntry, NotificationStatus
 from toad.widgets.committed_presentation import CommitParticipant, OutgoingMessageClaim
 
 
@@ -40,6 +40,9 @@ def message_reference(self):
     def notification_targets(self):
         return (self,) if self.message_dividers and self.display else ()
 
+    def queue_entries(self) -> tuple[MessageQueueEntry, ...]:
+        return (MessageQueueEntry(self.message_reference, self.query_one(MessageNotifications).status),)
+
     @property
     def notification_references(self):
         return (self.message_reference,)
diff --git a/tests/model/test_goal_display.py b/tests/model/test_goal_display.py
index 60a739985..80a3671b4 100644
--- a/tests/model/test_goal_display.py
+++ b/tests/model/test_goal_display.py
@@ -10,7 +10,7 @@ def test_each_goal_state_answers_visibility_control_and_snapshot():
     goal = published_goal("Keep the objective", "display-state")
     idle = IdleGoalExecution()
     for state, visible, enabled, snapshot in (
-        (NoGoal(), False, False, NoGoal()),  # nothing to control: Core answers a control with "This goal has been cleared."
+        (NoGoal(), False, False, NoGoal()),
         (goal, True, True, goal),
         (GoalUnavailable(goal), True, False, goal),
         (GoalUnavailable(NoGoal()), True, False, NoGoal()),
@@ -39,9 +39,6 @@ def heading(self, execution):
         def standby(self, execution):
             return False
 
-        def action_failure(self, error):
-            return str(error)
-
     state = OwnerReconnecting()
     assert state.visible
     assert not state.can_control
diff --git a/tests/model/test_queue_attachment.py b/tests/model/test_queue_attachment.py
index 018340508..078b99671 100644
--- a/tests/model/test_queue_attachment.py
+++ b/tests/model/test_queue_attachment.py
@@ -8,6 +8,7 @@
     AvailableQueueProjection, PendingQueueProjection, QueueChangedUpdate, LiveQueueScope,
 )
 from agent_comms.child_process import ProcessIdentity
+from agent_comms.nominal.field_codec import AcpSessionId
 from agent_comms.input_origin import HumanInputOrigin
 from agent_comms.threads import Thread
 
@@ -22,7 +23,7 @@ def declared_scope(tmp_path):
     comms.registry.declare(Thread("source", frozenset(), Checkout(tmp_path),
                                   process_identity=owner))
     return comms, LiveQueueScope(
-        session_id="source", claimed_identity=comms.registry.snapshot().claimed_identity("source"), owner=owner,
+        session_id=AcpSessionId("source"), claimed_identity=comms.registry.snapshot().claimed_identity("source"), owner=owner,
     )
 
 
diff --git a/tests/native/test_native_goal_bar.py b/tests/native/test_native_goal_bar.py
index 91a229bfa..3ea8a06af 100644
--- a/tests/native/test_native_goal_bar.py
+++ b/tests/native/test_native_goal_bar.py
@@ -176,7 +176,7 @@ async def test_standby_goal_mention_opens_its_peer_and_returns_to_the_draft(work
             "peer", frozenset(), Checkout(native.project), model=owner.model,
             thinking_level=owner.thinking_level,
         ))
-        await asyncio.to_thread(comms.owners.start, "peer")
+        await asyncio.to_thread(comms.owners.start_many, ("peer",))
         goal = comms.goals.update_goal(native.thread, SetGoalAction(text="Coordinate with @peer"))
         comms.goals.update_goal(native.thread, StandbyGoalAction(
             wait_for=("peer",), expect=GoalPrecondition(goal_id=OneGoal(goal.id))))
diff --git a/tests/native/test_native_goal_owner.py b/tests/native/test_native_goal_owner.py
index 8c7cf29dd..6151c2bc1 100644
--- a/tests/native/test_native_goal_owner.py
+++ b/tests/native/test_native_goal_owner.py
@@ -1,6 +1,7 @@
 """Goal requests reach the managed native owner through Toad's real controller."""
 
 import pytest
+from agent_comms.coordination_errors import Refused
 from agent_comms.goal_actions import SetGoalAction
 from agent_comms.goal_history import RecordedGoalHistory
 from agent_comms.input_attempt import OwnerQueueInputDeliveryScope
@@ -28,7 +29,7 @@ async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(worksp
         assert changed.revision == first.revision + 1
         assert changed.state == first.state
         assert comms.registry.require(session).goal == changed
-        with pytest.raises(ValueError):
+        with pytest.raises(Refused):
             await agent.edit_goal(first, "Stale overwrite")
         history: RecordedGoalHistory = await agent.get_goal_history(changed.id)
         assert any(entry.after.shows_document and entry.after.require().text == "First objective"
diff --git a/tests/native/test_native_journey.py b/tests/native/test_native_journey.py
index 0427d4a35..6a4e8f1d9 100644
--- a/tests/native/test_native_journey.py
+++ b/tests/native/test_native_journey.py
@@ -75,7 +75,7 @@ async def test_owner_is_attached_and_navigable(workspace, native_package):
         owner = native.comms.registry.require("beta")
         assert owner.process_identity is not None and owner.process_alive
         navigation = await asyncio.to_thread(
-            ThreadNavigationRequest(native.comms, "beta", ui.app.project_dir, ()).read)
+            ThreadNavigationRequest(native.comms, "beta", native.ui.app.project_dir, ()).read)
         assert navigation.attachable and navigation.thread.process_alive
 
 
@@ -103,7 +103,9 @@ def beta_row():
 
 
 async def test_input_queued_behind_a_running_turn_is_consumed_and_saved(workspace, native_package):
-    async with native_session(workspace) as native:
+    # Send now explicitly aborts the held first reply before consuming the queued input.
+    provider = ControlledProvider(expected_disconnects=frozenset({1}))
+    async with native_session(workspace, provider=provider) as native:
         ui, agent, comms = native.ui, native.agent, native.comms
         view = ui.conversation
         await send_first_input_and_hold_its_reply(native)
@@ -112,6 +114,7 @@ async def test_input_queued_behind_a_running_turn_is_consumed_and_saved(workspac
         queued = await queue_behind_the_held_turn(native, queued_text)
         assert view.submissions.queue_projection.status == "available"
         assert len(view.submissions.queue_projection.items[0].preview) <= 100
+        await view.submissions.send_now().wait()
         await finish_both_turns(native)
         assert ui.draft() == "unsent local draft"
 
@@ -129,6 +132,10 @@ async def test_input_queued_behind_a_running_turn_is_consumed_and_saved(workspac
 
         native_rows = [json.loads(line) for line in
                        Path(comms.registry.require("beta").runtime.path).read_text().splitlines()]
+        first_answer = next(row["message"] for row in native_rows
+                            if row.get("message", {}).get("role") == "assistant")
+        assert first_answer["stopReason"] == "aborted"
+        assert len(provider.requests) == 2, "Send now replayed or lost a native input"
         disposition = InputDispositions(comms.root).lookup(AcpInputKey(queued))
         assert isinstance(disposition, StartedInput)
         original = next(row["message"] for row in native_rows
diff --git a/tests/native/test_stopped_thread_start.py b/tests/native/test_stopped_thread_start.py
index 536a67bb7..b7afc462c 100644
--- a/tests/native/test_stopped_thread_start.py
+++ b/tests/native/test_stopped_thread_start.py
@@ -46,7 +46,7 @@ async def test_start_attaches_stopped_saved_tab_and_receives_dm(workspace, nativ
                 assert not provider.requests, 'Start must not submit a prompt'
                 before = comms.registry.require('stopped').process_identity
                 # A second successful Start must still nominate this tab for attachment.
-                result = await asyncio.to_thread(comms.owners.start, 'stopped')
+                result, = await asyncio.to_thread(comms.owners.start_many, ('stopped',))
                 assert not result.launched
                 assert result.reconnect_targets() == ('stopped',)
                 assert comms.registry.require('stopped').process_identity == before
diff --git a/tests/support/goals.py b/tests/support/goals.py
index 62923fe8c..27ea991f6 100644
--- a/tests/support/goals.py
+++ b/tests/support/goals.py
@@ -4,6 +4,7 @@
 
 from dataclasses import replace
 
+from agent_comms.errors import GoalRefused
 from agent_comms.goal_actions import GoalAction
 from agent_comms.goal_mentions import bind_goal_mentions
 from agent_comms.goal_presentation import GoalExecution, GoalSnapshot, RunnableExecution
@@ -56,16 +57,19 @@ def owning(cls, app: ToadApp, goal: GoalDeclaration,
     async def get_goal_snapshot(self):
         return GoalSnapshot(self.goal, self.execution)
 
+    def accept_write(self, kind: str, details: object) -> None:
+        self.requests.append((kind, details))
+        if self.reject:
+            raise GoalRefused("Goal changed; refresh before writing")
+
     async def update_goal(self, action: type[GoalAction], text: str = ""):
-        self.requests.append(("update", action))
+        self.accept_write("update", action)
         self.goal = replace(self.goal, state=ActiveGoal())
         self.execution = RunnableExecution(self.goal.id)
         return self.goal
 
     async def edit_goal(self, expected, text):
-        self.requests.append(("edit", (expected.id, expected.revision, text)))
-        if self.reject:
-            raise ValueError("Goal changed; refresh before editing")
+        self.accept_write("edit", (expected.id, expected.revision, text))
         self.goal = replace(self.goal, text=text, revision=self.goal.revision + 1)
         return self.goal
 
diff --git a/tests/support/native.py b/tests/support/native.py
index b27e7b6fc..ac28b5966 100644
--- a/tests/support/native.py
+++ b/tests/support/native.py
@@ -275,7 +275,7 @@ async def native_session(workspace, *, provider: ControlledProvider | None = Non
             if prepare is not None:
                 await prepare(comms, workspace.project, provider)
             if attached:
-                await asyncio.to_thread(comms.owners.start, NativeSession.thread)
+                await asyncio.to_thread(comms.owners.start_many, (NativeSession.thread,))
             async with attached_app(workspace, comms, NativeSession.thread, size=size, application=application,
                                     attached=attached) as (ui, agent):
                 yield NativeSession(ui, agent, comms, provider, workspace.project)
diff --git a/tests/support/workspace.py b/tests/support/workspace.py
index f5f00ff23..4691c6424 100644
--- a/tests/support/workspace.py
+++ b/tests/support/workspace.py
@@ -94,7 +94,8 @@ def agents(self) -> AppAgents:
 
     @asynccontextmanager
     async def app(self, *, size: tuple[int, int] = (120, 40), binding: AgentBinding = NoAgent(),
-                  application: type | None = None, session: bool = True, resume: str | None = None):
+                  application: type | None = None, session: bool = True, resume: str | None = None,
+                  notifications: bool = False):
         """Start Toad headless on this project; yields once its first session is shown.
 
         Toad itself starts with no tab open unless the launch names a session
@@ -105,7 +106,7 @@ async def app(self, *, size: tuple[int, int] = (120, 40), binding: AgentBinding
         launch = {} if resume is None else {"agent_session_id": resume}
         app = (application or ToadApp)(agent_binding=binding, project_dir=Checkout(self.project),
                                       launch_root=self.launch_root, **launch)
-        async with app.run_test(size=size) as pilot:
+        async with app.run_test(size=size, notifications=notifications) as pilot:
             await app.application.start_worker.wait()
             ui = TextualUi(app, pilot)
             if session:
diff --git a/tests/textual/test_goal_owner_views.py b/tests/textual/test_goal_owner_views.py
index 7efca58d1..06d7595b3 100644
--- a/tests/textual/test_goal_owner_views.py
+++ b/tests/textual/test_goal_owner_views.py
@@ -3,8 +3,11 @@
 import asyncio
 from dataclasses import replace
 
+import pytest
+
 from agent_comms.acp_extension import GoalChangedUpdate
-from agent_comms.nominal.field_codec import TurnId
+from agent_comms.goal_actions import ClearGoalAction
+from agent_comms.nominal.field_codec import FieldCodec, MalformedRecord, TurnId
 from agent_comms.goal_failure_observation import FailedTurnEvidence
 from agent_comms.goal_history import (
     RecordedGoalHistory, RecordedGoalRevision, BaselineRevision, TransitionRevision, ObservedGapRevision,
@@ -13,7 +16,7 @@
     GoalWaitTarget, IdleGoalExecution, PausedExecution, RunnableExecution, StandbyExecution,
 )
 from agent_comms.goal_states import PausedGoal
-from agent_comms.goals import NoGoal
+from agent_comms.goals import Goal, NoGoal
 from agent_comms.repo import Checkout
 from agent_comms.thread_identity import RecordedThreadIncarnation
 from agent_comms.turn_failure import NativeChildSignaled
@@ -26,7 +29,7 @@
 from toad.navigation_target import SelectTarget
 from toad.screens.goal_details import GoalDetails
 from toad.screens.goal_edit import GoalEdit
-from toad.widgets.goal_bar import GoalBar
+from toad.widgets.goal_bar import GoalBar, GoalControl
 
 
 def goal_owner(workspace, app):
@@ -183,6 +186,42 @@ async def test_goal_history_lists_earlier_revisions(workspace):
         assert view.prompt.text == "Composer draft stays"
 
 
+async def test_refused_goal_control_preserves_owner_state_and_composer(workspace):
+    async with workspace.app(size=(100, 35), notifications=True) as ui:
+        owner, view, bar, _status = await shown_goal(workspace, ui)
+        original = owner.goal
+        clear = bar.query_one("#goal-clear", GoalControl)
+        hit = ui.app.screen.get_widget_at(*clear.region.offset)[0]
+        print(f"Goal clear region={clear.region!r}; hit={hit!r}; disabled={clear.disabled}")
+        assert await ui.pilot.click(clear), "the real clear control must receive the click"
+        await ui.until(lambda: owner.requests == [("update", ClearGoalAction)],
+                       why="one goal-clear request at the scripted owner")
+        await ui.until_shows("Goal action failed")
+        assert owner.goal == original
+        assert view.goal_display.snapshot == original
+        assert view.prompt.text == "Composer draft stays"
+
+
+async def test_goal_editor_does_not_relabel_a_malformed_reply_as_a_refusal(workspace):
+    async with workspace.app(size=(100, 35)) as ui:
+        owner, view, _bar, _status = await shown_goal(workspace, ui)
+
+        async def save(text):
+            async def decode_reply():
+                return FieldCodec.decode(Goal, {})
+            await view.goal_controls.write(owner, decode_reply())
+
+        editor = GoalEdit(owner.goal, project=Checkout(workspace.project), on_save=save)
+        await ui.app.push_screen(editor)
+        editor.editor.text = "Keep this draft after the bad reply"
+        with pytest.raises(MalformedRecord):
+            await editor.save(editor.editor.text)
+        assert ui.app.screen is editor
+        assert editor.editor.text == "Keep this draft after the bad reply"
+        assert owner.requests == []
+        assert view.goal_display.snapshot == owner.goal
+
+
 async def test_owner_pushes_replace_and_then_remove_the_goal(workspace):
     async with workspace.app(size=(100, 35)) as ui:
         owner, view, bar, status = await shown_goal(workspace, ui)
diff --git a/tests/textual/test_live_output_streams.py b/tests/textual/test_live_output_streams.py
index 6272fdf8e..f1809173d 100644
--- a/tests/textual/test_live_output_streams.py
+++ b/tests/textual/test_live_output_streams.py
@@ -3,12 +3,14 @@
 from agent_comms.routing import MessageRoute
 from support.agents import RecordedEvents
 
+from toad.acp.comms_updates import CommsUpdateConsumer
 from toad.core.events import CoreEvent, Update
 from toad.core_event_carrier import CoreEventMessage, dispatch_publication
-from toad.live_output import OutputStream, ResponseStream, ThoughtStream
-from toad.response_delivery import RoutedResponse, UnroutedResponse
+from toad.live_output import CompleteResponseStream, OutputStream, ResponseStream, ThoughtStream
+from toad.response_delivery import RoutedResponse
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.agent_thought import AgentThought
+from toad.widgets.committed_presentation import CapturedClaim
 
 
 class DiagnosticStream(OutputStream):
@@ -19,6 +21,11 @@ def create(self, fragment):
         return AgentThought(fragment)
 
 
+def consumed_stream(agent):
+    """The stream one incoming agent message gets from the update consumer that owns it."""
+    return CommsUpdateConsumer(agent, "session").response_stream()
+
+
 def painted_window(ui):
     window = ui.conversation.window.region
     return "\n".join(strip.crop(window.x, window.right).text
@@ -61,8 +68,8 @@ async def test_queued_client_replies_of_one_local_turn_form_one_reply(workspace)
         ui.attach_agent(agent)
         subscription = agent.controller.surface.subscription
         view.turns.start_client()
-        first = view.turns.owner.response_stream(UnroutedResponse())
-        second = view.turns.owner.response_stream(UnroutedResponse())
+        first = consumed_stream(agent)
+        second = consumed_stream(agent)
         view.turns.finish_client()
         await dispatch_publication(view, CoreEventMessage(Update("text", "Local queued", first), subscription))
         await dispatch_publication(view, CoreEventMessage(Update("text", " response", second), subscription))
@@ -71,14 +78,14 @@ async def test_queued_client_replies_of_one_local_turn_form_one_reply(workspace)
         assert local.body.text == "Local queued response"
 
 
-async def test_a_different_turn_id_or_owner_notice_starts_a_new_block(workspace):
+async def test_a_different_append_claim_or_notice_starts_a_new_block(workspace):
     async with workspace.app() as ui:
         view = ui.conversation
-        one = await view.output.append(ResponseStream(turn_id="one"), "First turn")
-        two = await view.output.append(ResponseStream(turn_id="two"), "Second turn")
+        one = await view.output.append(ResponseStream(source_claim=CapturedClaim()), "First reply")
+        two = await view.output.append(ResponseStream(source_claim=CapturedClaim()), "Second reply")
         assert one is not two
-        notice = await view.output.append(view.turns.owner.response_stream(UnroutedResponse()), "Owner notice")
-        another = await view.output.append(view.turns.owner.response_stream(UnroutedResponse()), "Next owner notice")
+        notice = await view.output.append(CompleteResponseStream(), "Owner notice")
+        another = await view.output.append(CompleteResponseStream(), "Next owner notice")
         assert notice is not another
 
 
@@ -89,7 +96,7 @@ async def test_an_update_published_by_another_agent_adds_no_block(workspace):
         ui.attach_agent(agent)
         subscription = agent.controller.surface.subscription
         view.turns.start_client()
-        queued = view.turns.owner.response_stream(UnroutedResponse())
+        queued = consumed_stream(agent)
         view.turns.finish_client()
         await dispatch_publication(view, CoreEventMessage(Update("text", "Local queued", queued), subscription))
         await view.output.finish(ResponseStream)
diff --git a/tests/textual/test_painted_read_maintenance.py b/tests/textual/test_painted_read_maintenance.py
new file mode 100644
index 000000000..c821d7ab1
--- /dev/null
+++ b/tests/textual/test_painted_read_maintenance.py
@@ -0,0 +1,51 @@
+"""Real read-ledger refusal through the shared Textual read worker, not a fake error."""
+
+import asyncio
+
+from agent_comms.maintenance_barrier import MaintenanceOperator, NoMaintenanceHolder
+from agent_comms.read_ledger import ReadLedger
+
+from toad.widgets.message_notifications import ReadWork
+
+
+async def test_maintenance_keeps_read_marks_and_fresh_work_succeeds(workspace):
+    source = workspace.project / "read-metadata-control.txt"
+    source.write_text("First painted row\n", encoding="utf-8")
+    first = source.stat()
+    viewer = workspace.viewer
+    ledger = ReadLedger.at(workspace.comms.root)
+    barrier = workspace.comms.owners.maintenance
+    original = ledger.read()
+
+    with NoMaintenanceHolder().selected():
+        async with workspace.app(size=(100, 35), notifications=True) as ui:
+            marker = ui.app.selected_session
+            hold = barrier.begin(MaintenanceOperator.current())
+            try:
+                previous_workers = set(marker.workers)
+                ReadWork(asyncio.to_thread(
+                    ledger.mark_transcript, viewer, str(source), first.st_ino,
+                    first.st_size, first.st_size,
+                )).start(marker)
+                worker, = set(marker.workers) - previous_workers
+                await worker.wait()  # A WorkerFailed is a failed control, not a warning.
+                await ui.until_shows("Read not marked")
+                assert ledger.read() == original
+                assert ledger.transcript_seen(viewer, str(source), first.st_ino) == 0
+                assert ui.failure() is None
+            finally:
+                barrier.clear(hold)
+
+            # A new source/cursor observation after relief, not an automatic replay.
+            with source.open("a", encoding="utf-8") as output:
+                output.write("Newly painted row after maintenance\n")
+            fresh = source.stat()
+            previous_workers = set(marker.workers)
+            ReadWork(asyncio.to_thread(
+                ledger.mark_transcript, viewer, str(source), fresh.st_ino,
+                fresh.st_size, fresh.st_size,
+            )).start(marker)
+            worker, = set(marker.workers) - previous_workers
+            await worker.wait()
+            assert ledger.transcript_seen(viewer, str(source), fresh.st_ino) == fresh.st_size
+            assert ui.failure() is None
diff --git a/tests/textual/test_queue_inspection.py b/tests/textual/test_queue_inspection.py
new file mode 100644
index 000000000..1996fccef
--- /dev/null
+++ b/tests/textual/test_queue_inspection.py
@@ -0,0 +1,28 @@
+"""Queue details keep their original project while mounted as a separate screen."""
+
+from textual.widgets import Button
+
+from toad.project_path_owner import ProjectPathOwner
+from toad.screens.queue import QueueInspection
+from toad.widgets.tool_content import ToolContent
+
+
+async def test_queue_details_mount_with_the_original_project_and_keep_the_draft(workspace):
+    async with workspace.app(size=(115, 38)) as ui:
+        view = ui.conversation
+        owner = ProjectPathOwner.containing(view)
+        view.prompt.text = "Keep this unsent queue draft"
+        view.open_queue_menu()
+        await ui.until(lambda: isinstance(ui.app.screen, QueueInspection), why="the queue inspection screen")
+        screen = ui.app.screen
+        await ui.until(lambda: bool(screen.query(ToolContent)), why="the mounted rich queue details")
+        details = screen.query_one(ToolContent)
+        assert ProjectPathOwner.containing(details) is screen
+        assert screen.source_view is view
+        assert screen.project_root == owner.project_root
+        assert screen.coordination_root == owner.coordination_root
+        assert screen.link_context_events is owner.link_context_events
+        assert screen.query_one("#queue-send-now", Button).disabled
+        await ui.press("escape")
+        await ui.until(lambda: ui.app.screen is not screen, why="the dismissed queue inspection")
+        assert view.prompt.text == "Keep this unsent queue draft"
diff --git a/tests/textual/test_queue_request_results.py b/tests/textual/test_queue_request_results.py
index 62c9240c9..6a8a3dfd7 100644
--- a/tests/textual/test_queue_request_results.py
+++ b/tests/textual/test_queue_request_results.py
@@ -8,7 +8,7 @@
 from agent_comms.acp_extension import (
     AvailableQueueProjection, QueueChangedUpdate, QueueItem, LiveQueueScope, encode_updates,
 )
-from agent_comms.nominal.field_codec import InputId
+from agent_comms.nominal.field_codec import AcpSessionId, InputId
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.thread_identity import ClaimedIdentity, RecordedThreadIncarnation
 
@@ -20,7 +20,7 @@
 def response(session, next_owner=False):
     generation = 2 if next_owner else 1
     scope = LiveQueueScope(
-        session_id=session,
+        session_id=AcpSessionId(session),
         claimed_identity=ClaimedIdentity(RecordedThreadIncarnation(session, 1000.0), generation),
         owner=ProcessIdentity(1234, 1),
     )