diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0a2b25f30..1a782d3d3 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -1,5 +1,7 @@
 name: Local-equivalent verification
 on:
+  push:
+    branches: [main]
   pull_request:
     branches: [main]
   workflow_dispatch:
@@ -14,7 +16,7 @@ permissions:
   contents: read
 concurrency:
   group: toad-verification-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
+  cancel-in-progress: ${{ github.event_name != 'push' }}
 jobs:
   nra-gate:
     name: NRA gate
@@ -24,7 +26,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          ref: ${{ github.event.pull_request.head.sha || inputs.head }}
+          ref: ${{ github.event.pull_request.head.sha || inputs.head || github.sha }}
       - uses: actions/setup-python@v5
         with:
           python-version: '3.14'
@@ -44,8 +46,8 @@ jobs:
           # pull_request.base.sha is not refreshed when the base branch moves, so it can name a
           # revision many commits behind and charge the PR for debt main introduced. The PR's own
           # change starts at the merge-base with the base branch's current tip.
-          BASE_TIP: ${{ github.event.pull_request.base.ref && format('origin/{0}', github.event.pull_request.base.ref) || inputs.base }}
-          PR_HEAD: ${{ github.event.pull_request.head.sha || inputs.head }}
+          BASE_TIP: ${{ github.event.pull_request.base.ref && format('origin/{0}', github.event.pull_request.base.ref) || inputs.base || github.event.before }}
+          PR_HEAD: ${{ github.event.pull_request.head.sha || inputs.head || github.sha }}
         run: PYTHONPATH="$PWD/.core-gate/src" PATH="$PWD/.venv/bin:$PATH" .venv/bin/python .core-gate/tools/nra_client_gate.py --root src/toad --base "$(git merge-base "$BASE_TIP" "$PR_HEAD")" --head "$PR_HEAD"
       - name: Run surface guards
         run: .venv/bin/pytest tests/guards
@@ -56,7 +58,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          ref: ${{ github.event.pull_request.head.sha || inputs.head }}
+          ref: ${{ github.event.pull_request.head.sha || inputs.head || github.sha }}
       - uses: actions/setup-python@v5
         with:
           python-version: '3.14'
diff --git a/src/toad/markdown_links.py b/src/toad/markdown_links.py
index 9f34d2478..1cc35ab16 100644
--- a/src/toad/markdown_links.py
+++ b/src/toad/markdown_links.py
@@ -178,7 +178,7 @@ def resolve(cls, renderer: ProjectTokenRenderer, token: Token, context: _Project
         if context.linked:
             return [token]
         result, position = [], 0
-        for start, end, tokens in context.links.matches(token.content):
+        for start, end, tokens in cls.matches(token.content, context.links):
             if start > position:
                 result.append(Token(token.type, token.tag, 0, content=token.content[position:start]))
             # Retain inline-code formatting on both linked and unlinked pieces.
@@ -192,8 +192,17 @@ def resolve(cls, renderer: ProjectTokenRenderer, token: Token, context: _Project
         return result
 
 
+    @staticmethod
+    def matches(text: str, links: LinkContext):
+        return links.matches(text)
+
+
 class CodeInlineTokenRule(TextTokenRule):
-    pass
+    """A code span links only when the whole span is a reference: `run file.md` is a command."""
+
+    @staticmethod
+    def matches(text: str, links: LinkContext):
+        return [(start, end, tokens) for start, end, tokens in links.matches(text) if (start, end) == (0, len(text))]
 
 
 class ProjectTokenRenderer:
@@ -211,6 +220,9 @@ def resolve(self, tokens: list[Token], context: _ProjectTokens):
 class _MarkdownParserState(local):
     def __init__(self):
         self.parser = MarkdownIt("gfm-like")
+        # Linkify owns written schemes and e-mail addresses. A schemeless spelling such as
+        # file.md or declarations.py is a project path first, which TextTokenRule resolves.
+        self.parser.linkify.set({"fuzzy_link": False})
         self.parser.validateLink = ProjectLink.opens
         self.resolver = ProjectTokenRenderer()
         self.syntax_rules = tuple(self.parser.core.ruler.getRules(""))
diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py
index d5a2fe5d5..c78658cda 100644
--- a/src/toad/widgets/message_notifications.py
+++ b/src/toad/widgets/message_notifications.py
@@ -10,6 +10,7 @@
 from textual.app import ComposeResult
 from textual.content import Content
 from textual.widgets import Collapsible
+from textual.worker import Worker
 
 from agent_comms.ui_model.observation import ReadNotifications
 
@@ -81,6 +82,11 @@ def start(self, marker: VisibleReadMarking) -> None:
 
 
 class NothingToRead(PaintedRead):
+    is_finished = True
+
+    async def wait(self) -> None:
+        pass
+
     def start(self, marker: VisibleReadMarking) -> None:
         pass
 
@@ -90,18 +96,18 @@ class ReadWork(PaintedRead):
     work: Awaitable[None]
 
     def start(self, marker: VisibleReadMarking) -> None:
-        marker._read_worker = marker.run_worker(self.work, group="painted-read")
+        marker.read_worker = marker.run_worker(self.work, group="painted-read")
 
 
 class VisibleReadMarking:
     """One painted-read operation at a time, for pagers and native/CLI sessions."""
 
     def __init__(self, *args, **kwargs):
-        self._read_worker = None
+        self.read_worker: Worker[None] | NothingToRead = NothingToRead()
         super().__init__(*args, **kwargs)
 
     def mark_visible(self) -> None:
-        if self._read_worker is not None and not self._read_worker.is_finished:
+        if not self.read_worker.is_finished:
             return
         self.read_visible().start(self)
 
diff --git a/tests/app/test_channel_display_basis.py b/tests/app/test_channel_display_basis.py
index 925a25dc8..91ba1b4ca 100644
--- a/tests/app/test_channel_display_basis.py
+++ b/tests/app/test_channel_display_basis.py
@@ -1,5 +1,7 @@
 """A channel marks read exactly the messages whose text it showed, never the history it left out."""
+from agent_comms.read_basis import ChannelConversation
 from agent_comms.repo import Checkout
+from agent_comms.ui_model.unread import ExactUnread
 
 
 async def test_shown_messages_are_read_exactly_and_hidden_expansion_stays_unread(workspace):
@@ -13,8 +15,9 @@ async def test_shown_messages_are_read_exactly_and_hidden_expansion_stays_unread
     messages = [comms.messaging.send("alice", "#team", "\n".join(f"channel {i:02d} line" for _ in range(12))).message
                 for i in range(12)]
 
-    def seen():
-        return comms.bus.reads.seen_sequences(viewer, comms.registry.snapshot())
+    ledger = comms.bus.reads
+    snapshot = comms.registry.snapshot()
+    channel = ChannelConversation("#team")
 
     async with workspace.app(size=(120, 40)) as ui:
         owner = ui.selected_tab
@@ -24,14 +27,21 @@ def seen():
         def shown():
             return {message.seq for i, message in enumerate(messages) if ui.shows(f"channel {i:02d} line")}
 
-        await ui.until(lambda: seen() == shown(), why="the shown messages to be marked read")
+        await ui.until(lambda: all(
+            (message.seq in ledger.cursor(viewer, snapshot).for_conversation(channel))
+            == (message.seq in shown()) for message in messages
+        ), why="the shown messages to be marked read")
         # Earlier messages were read from Core for scrolling, but never shown.
         assert 0 < len(shown()) < len(messages) and not ui.shows("channel 00 line")
-        assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] == len(messages) - len(shown())
-        assert hidden.seq not in seen()
+        assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] == ExactUnread(
+            count=len(messages) - len(shown()))
+        assert hidden.seq not in ledger.cursor(viewer, snapshot).for_conversation(
+            ledger.conversation(hidden, snapshot))
         await ui.select_tab(owner)
         comms.channels.set_channel_any_mode("#team", True)
         comms.channels.update_tags("bob", add=frozenset({"team"}))
         # Expanding a hidden tab's view never manufactures read facts.
-        assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] >= len(messages) - len(shown()) + 1
-        assert hidden.seq not in seen()
+        assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] == ExactUnread(
+            count=len(messages) - len(shown()) + 1)
+        assert hidden.seq not in ledger.cursor(viewer, snapshot).for_conversation(
+            ledger.conversation(hidden, snapshot))
diff --git a/tests/app/test_channel_scope_expansion.py b/tests/app/test_channel_scope_expansion.py
index b5adb5890..ce7417d8a 100644
--- a/tests/app/test_channel_scope_expansion.py
+++ b/tests/app/test_channel_scope_expansion.py
@@ -1,5 +1,7 @@
 """A mounted any-mode channel reveals older messages entering its scope."""
 
+from agent_comms.repo import Checkout
+
 
 async def test_mounted_any_mode_channel_reveals_older_messages_entering_its_scope(workspace):
     for name, tags in (("alice", ("team",)), ("bob", ()), ("carol", ()), ("dave", ())):
@@ -8,7 +10,7 @@ async def test_mounted_any_mode_channel_reveals_older_messages_entering_its_scop
     comms.messaging.send("carol", "dave", "older newly visible DM")
     comms.messaging.send("alice", "#team", "current channel message")
     comms.channels.set_channel_any_mode("#team", True)
-    assert [m.body for m in comms.views.channel_display_page("#team", worktree=str(workspace.project)).messages] == [
+    assert [m.body for m in comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(workspace.project))).messages] == [
         "current channel message"]
 
     async with workspace.app(size=(100, 30)) as ui:
diff --git a/tests/app/test_markdown_file_links.py b/tests/app/test_markdown_file_links.py
index d8189c16c..b3a0436e2 100644
--- a/tests/app/test_markdown_file_links.py
+++ b/tests/app/test_markdown_file_links.py
@@ -2,6 +2,7 @@
 from toad.session_scope import AgentAttached
 
 from support.agents import peer
+from toad.project_link import DirectProjectFileLink, ExternalProjectLink
 
 # One reference per paragraph: ui.link_at reaches the first match on a drawn row.
 ANSWER_SOURCE = "see file.md\n\nsee `file.md`\n\nbut `run file.md` is a command\n\nsee [website](https://example.com)"
@@ -24,8 +25,8 @@ async def test_file_link_in_answer_opens_preview_and_return_keeps_draft(workspac
         await ui.until_shows("PEER_ANSWER")
         await ui.write_draft("KEEP_MARKDOWN_DRAFT")
         links = file_links(ui, "file.md")
-        assert sum(f"toad-file:{target}" in str(link) for link in links) == 2, links
-        assert any("https://example.com" in str(link) for link in file_links(ui, "website")), ui.visible_text()
+        assert links.count(DirectProjectFileLink(target.resolve(), original_root=workspace.project.resolve()).click) == 2, links
+        assert file_links(ui, "website") == [ExternalProjectLink("https://example.com").click], ui.visible_text()
         drawn = [i for i, line in enumerate(line for line in ui.visible_lines() if "file.md" in line)]
         await ui.click("file.md", occurrence=next(i for i in drawn if ui.link_at("file.md", occurrence=i) is not None))
         await ui.until_shows("REAL_MARKDOWN_FILE_PREVIEW")
diff --git a/tests/app/test_prompt_routing.py b/tests/app/test_prompt_routing.py
index 66920ddef..9531bfe0e 100644
--- a/tests/app/test_prompt_routing.py
+++ b/tests/app/test_prompt_routing.py
@@ -7,7 +7,8 @@
 async def test_command_looking_text_is_sent_to_the_agent_not_the_shell(workspace):
     async with workspace.app(binding=AgentAttached(peer("echo"))) as ui:
         await ui.until_agent_ready()
-        sent = ["openhcs", "open the project", "git status", "python", "$100 budget", "@peer", "!literal"]
+        # "!" on an empty prompt enters shell mode (cfe1bb492); that is tests/textual/test_prompt_shell_mode.py's.
+        sent = ["openhcs", "open the project", "git status", "python", "$100 budget", "@peer"]
         for text in sent:
             await ui.type(text)
             assert ui.draft() == text
@@ -32,12 +33,3 @@ async def test_enter_sends_the_whole_multiline_draft_and_ctrl_j_adds_a_line(work
         assert len(workspace.prompts_received()) == sent
         await ui.press("enter")
         await ui.until(lambda: workspace.prompts_received()[-1:] == ["line one\nline two\nline three"])
-
-
-async def test_an_exclamation_mark_is_text_not_a_shell_switch(workspace):
-    async with workspace.app(binding=AgentAttached(peer("echo"))) as ui:
-        await ui.until_agent_ready()
-        await ui.press("!")
-        assert ui.draft() == "!"
-        await ui.press("enter")
-        await ui.until(lambda: workspace.prompts_received() == ["!"], why="'!' at the agent")
diff --git a/tests/app/test_unread_clears_on_opening.py b/tests/app/test_unread_clears_on_opening.py
index b457b6c0d..d327e926e 100644
--- a/tests/app/test_unread_clears_on_opening.py
+++ b/tests/app/test_unread_clears_on_opening.py
@@ -1,5 +1,6 @@
 """A displayed channel page clears its own unread rows after every navigation."""
 from agent_comms.repo import Checkout
+from agent_comms.ui_model.unread import ExactUnread
 
 
 def unread_badge_gone(ui, identity):
@@ -11,23 +12,24 @@ async def test_channel_unread_clears_on_opening_and_on_return_to_the_open_tab(wo
     comms = workspace.comms
     comms.messaging.send("sender", "#all", "first-pending")
 
-    def unread() -> int:
-        return comms.views.viewer_snapshot(Checkout(workspace.project)).channel_unread["#all"]
-
-    assert unread() == 1
+    assert comms.views.viewer_snapshot(Checkout(workspace.project)).channel_unread["#all"] == ExactUnread(count=1)
     async with workspace.app(size=(110, 36)) as ui:
         owner = ui.selected_tab
         channel = await ui.open_channel("#all", actor=workspace.viewer)
         await ui.until_shows("first-pending", seconds=5)
-        await ui.until(lambda: unread() == 0, seconds=5, why="the channel to be read")
+        await ui.until(lambda: comms.views.viewer_snapshot(
+            Checkout(workspace.project)).channel_unread["#all"] == ExactUnread(),
+            seconds=5, why="the channel to be read")
         await ui.until(lambda: unread_badge_gone(ui, channel), seconds=5, why="the unread badge to clear")
 
         await ui.select_tab(owner)
         comms.messaging.send("sender", "#all", "second-pending")
-        assert unread() == 1
+        assert comms.views.viewer_snapshot(Checkout(workspace.project)).channel_unread["#all"] == ExactUnread(count=1)
         await ui.select_tab(channel)
         await ui.until_shows("second-pending", seconds=5)
-        await ui.until(lambda: unread() == 0, seconds=5, why="the returned channel to be read")
+        await ui.until(lambda: comms.views.viewer_snapshot(
+            Checkout(workspace.project)).channel_unread["#all"] == ExactUnread(),
+            seconds=5, why="the returned channel to be read")
         await ui.until(lambda: unread_badge_gone(ui, channel), seconds=5, why="the unread badge to clear on return")
 
 
@@ -35,16 +37,15 @@ async def test_direct_message_unread_clears_on_opening(workspace):
     workspace.thread("sender")
     comms = workspace.comms
 
-    def unread() -> int:
-        return comms.views.viewer_snapshot(Checkout(workspace.project)).unread.get("sender", 0)
-
     async with workspace.app(size=(110, 36)) as ui:
         owner = ui.selected_tab
         comms.messaging.send("sender", workspace.viewer, "dm-pending")
-        assert unread() == 1
+        assert comms.views.viewer_snapshot(Checkout(workspace.project)).unread["sender"] == ExactUnread(count=1)
         direct = await ui.open_direct("sender", actor=workspace.viewer, owner=owner)
         await ui.until_shows("dm-pending", seconds=5)
-        await ui.until(lambda: unread() == 0, seconds=5, why="the direct messages to be read")
+        await ui.until(lambda: comms.views.viewer_snapshot(
+            Checkout(workspace.project)).unread["sender"] == ExactUnread(),
+            seconds=5, why="the direct messages to be read")
         await ui.until(lambda: unread_badge_gone(ui, direct), seconds=5, why="the unread badge to clear")
 
 
@@ -52,14 +53,13 @@ async def test_aggregate_feed_unread_clears_on_opening(workspace):
     workspace.thread("sender")
     comms = workspace.comms
 
-    def unread() -> int:
-        return comms.views.viewer_snapshot(Checkout(workspace.project)).channel_unread.get("#any", 0)
-
     async with workspace.app(size=(110, 36)) as ui:
         owner = ui.selected_tab
         comms.messaging.send("sender", "#all", "aggregate-pending")
-        assert unread() >= 1
+        assert comms.views.viewer_snapshot(Checkout(workspace.project)).channel_unread["#any"] == ExactUnread(count=1)
         feed = await ui.open_channel("#any", actor=workspace.viewer, owner=owner)
         await ui.until_shows("aggregate-pending", seconds=5)
-        await ui.until(lambda: unread() == 0, seconds=5, why="the feed to be read")
+        await ui.until(lambda: comms.views.viewer_snapshot(
+            Checkout(workspace.project)).channel_unread["#any"] == ExactUnread(),
+            seconds=5, why="the feed to be read")
         await ui.until(lambda: unread_badge_gone(ui, feed), seconds=5, why="the unread badge to clear")
diff --git a/tests/conftest.py b/tests/conftest.py
index eead05fcc..00816bbad 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -121,6 +121,20 @@ def pytest_collection_modifyitems(config, items):
             item.add_report_section("call", "Recorded defect", note)
 
 
+def pytest_sessionfinish(session):
+    """The test's own process stops what its app left running before it exits.
+
+    Those helpers inherit this process's output pipes, which the runner reads to their end
+    (conftest's BoundedRun.run): left running after a failed close, they held a finished
+    test open until --test-timeout killed it.
+    """
+    if (attempt := os.environ.get(ATTEMPT)) is None:
+        return  # The runner; the processes it was given stop their own.
+    from support.processes import stop_test_children
+
+    asyncio.run(stop_test_children(attempt))
+
+
 def pytest_runtest_logreport(report):
     if (path := os.environ.get(RESULT)) is None:
         return
diff --git a/tests/support/overlapping_annotations.py b/tests/support/overlapping_annotations.py
index 3ac607dab..738983549 100644
--- a/tests/support/overlapping_annotations.py
+++ b/tests/support/overlapping_annotations.py
@@ -3,15 +3,13 @@
 from __future__ import annotations
 
 from agent_comms.repo import Checkout
-from agent_comms.active_route import resolve_comms_route
 
 from dataclasses import dataclass, replace
 from pathlib import Path
 
-from agent_comms.comms import Comms
+from agent_comms.context_store import AbsentContext, ContextManifestPage
 from agent_comms.session_revision import NoSessionObservation
 from agent_comms.thread_identity import RecordedThreadIncarnation, TurnId, TurnIdentity
-from agent_comms.registration import Registration
 from agent_comms.turn_context import (
     ContextManifest, ContextSpan, ContributionCoordinates, RecordedContextTurn,
     UnattributedProvenance, UserInputSegment,
@@ -22,6 +20,7 @@
 from agent_comms.working_memory_questions import CommitmentSpan, KindQuestion, OtherSpan, RuleSpan
 from agent_comms.threads import Thread
 from toad.core.context_inspection import ContextInspection
+from support.roots import installed_wire
 
 
 @dataclass(frozen=True)
@@ -38,8 +37,9 @@ class OverlappingAnnotations:
 
 def overlapping_annotations(directory: Path) -> OverlappingAnnotations:
     """A short and a long annotation over one segment, as an inspection shows them, and the same after correcting the short one."""
-    owner = Thread(
-        "authored-navigation", frozenset(), Checkout(directory), created_at=1.0)
+    service = installed_wire(directory / "unused-wire")
+    owner = service.registry.declare(Thread(
+        "authored-navigation", frozenset(), Checkout(directory), created_at=1.0))
     segment = UserInputSegment(content="Keep the original instruction.",
                                provenance=(UnattributedProvenance(),))
     observed = segment.manifest(0)
@@ -51,13 +51,14 @@ def overlapping_annotations(directory: Path) -> OverlappingAnnotations:
         classifier, (AnswerProbability(RuleSpan, .7), AnswerProbability(CommitmentSpan, .2),
                      AnswerProbability(OtherSpan, .1)), .8, f"authored-answer-{index}", classifier.pin)
         for index, span in enumerate(spans))
-    # This unsealed model only exercises navigation: no request, no captured
-    # public text, no wire message, no native owner and no saved session.
+    # A locally captured context exercises navigation without a native request,
+    # wire message or saved session.
     manifest = ContextManifest(owner.incarnation,
-        RecordedContextTurn(TurnId("authored-unsealed-navigation"), TurnIdentity(owner.incarnation, 1)),
+        RecordedContextTurn(TurnId.mint(), TurnIdentity(owner.incarnation, 1)),
         (observed,), "authored-navigation")
-    inspection = ContextInspection(owner, (manifest,), NoSessionObservation(),
-                                   Comms(resolve_comms_route(directory / "unused-wire")), labels)
+    captured = service.contexts.record_after(AbsentContext(), manifest)
+    inspection = ContextInspection(service.registry.entry(owner.name).entry,
+        ContextManifestPage((captured,)), NoSessionObservation(), service, labels)
     ((_, (source,), _),) = inspection.working_memory()
     short, long = source.children()
     corrected = HumanLabel.correct(labels[0], RuleSpan, RecordedThreadIncarnation("authored-user", 2.0))
diff --git a/tests/support/textual_steps.py b/tests/support/textual_steps.py
index e0f2d2c56..b9b073f12 100644
--- a/tests/support/textual_steps.py
+++ b/tests/support/textual_steps.py
@@ -96,7 +96,7 @@ def text_rows_of(chat, message) -> range | None:
     the window is not on them until that layout ends.
     """
     window = chat.window
-    if window.reader_position is not None or window.screen._stepped_layout is not None:
+    if not window.reader_position.laid_out(window) or window.screen._stepped_layout is not None:
         return None
     for page in chat.message_history.pages:
         for fragment, first, lines in page._layout:
diff --git a/tests/support/ui.py b/tests/support/ui.py
index 2bc5f99cc..fcaa3dd7d 100644
--- a/tests/support/ui.py
+++ b/tests/support/ui.py
@@ -103,10 +103,6 @@ async def click(self, text: str, *, occurrence: int = 0) -> None:
     def link_at(self, text: str, *, occurrence: int = 0) -> tuple[str, tuple] | None:
         """The action a click on `text` would trigger, as (name, arguments), if it is a link."""
 
-    @abstractmethod
-    def hyperlink_at(self, text: str, *, occurrence: int = 0) -> str | None:
-        """The address a click on `text` would open, if the drawn text is a hyperlink."""
-
     @abstractmethod
     async def resize(self, width: int, height: int) -> None: ...
 
@@ -333,9 +329,6 @@ def link_at(self, text: str, *, occurrence: int = 0) -> tuple[str, tuple] | None
             "@click"
         )
 
-    def hyperlink_at(self, text: str, *, occurrence: int = 0) -> str | None:
-        return self.app.screen.get_style_at(*self._cell_of(text, occurrence)).link
-
     async def resize(self, width: int, height: int) -> None:
         await self.pilot.resize_terminal(width, height)
         await self.step()
@@ -395,10 +388,10 @@ async def go_forward(self) -> None:
         await self.step()
 
     def can_go_back(self) -> bool:
-        return not self.app.screen.query_one("#tab-back").has_class("-unavailable")
+        return not self.app.screen.query_one("#tab-back").disabled
 
     def can_go_forward(self) -> bool:
-        return not self.app.screen.query_one("#tab-forward").has_class("-unavailable")
+        return not self.app.screen.query_one("#tab-forward").disabled
 
     async def open_preview(self, path) -> str:
         identity = await self.app.session_navigation.preview(path)
diff --git a/tests/textual/test_acp_log_link.py b/tests/textual/test_acp_log_link.py
index 10623d299..d264cd6c8 100644
--- a/tests/textual/test_acp_log_link.py
+++ b/tests/textual/test_acp_log_link.py
@@ -63,7 +63,7 @@ async def test_a_link_to_a_log_that_was_never_written_opens_the_tab_and_says_the
     missing = workspace.roots.attempt / "state/toad/logs/never-written.txt"
     async with workspace.app(size=(139, 30)) as ui:
         owner = ui.selected_tab
-        await ui.conversation.post(AgentResponse(AcpLogLink(missing).markdown))
+        await ui.conversation.post(AgentResponse(AcpLogLink.markdown(missing)))
         await ui.until(lambda: ui.shows(LINK) and ui.link_at(LINK) is not None, why="the log link drawn as a link")
         await ui.click(LINK)
         await ui.until(lambda: ui.selected_tab != owner, why="the log tab selected")
diff --git a/tests/textual/test_app_close_order.py b/tests/textual/test_app_close_order.py
index 34d1544b5..eaac02f14 100644
--- a/tests/textual/test_app_close_order.py
+++ b/tests/textual/test_app_close_order.py
@@ -6,13 +6,13 @@
 import os
 
 import pytest
-from agent_comms.acp_extension import CoordinationChangedUpdate, encode_updates
+from agent_comms.acp_extension import encode_updates
 from agent_comms.child_process import ProcessIdentity
-from agent_comms.context_reading import UnmeasuredContext
 from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.fresh_private_session import create_fresh_private_session
 
 from support.agents import deliver
+from support.coordination import registered_coordination_fact
 from toad.widgets.transcript_history import TranscriptHistory
 
 
@@ -36,17 +36,15 @@ async def test_closing_the_app_after_the_observation_process_exits_raises_nothin
         stream.write(json.dumps({"type": "message", "id": "ask",
             "timestamp": "2026-10-22T00:00:00Z", "message": {
         "role": "user", "content": [{"type": "text", "text": "ASK_ROW"}]}}) + "\n")
-    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
+    comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
     workspace.thread("peer")
     comms.messaging.send("peer", "source", "SAVED_NOTICE")
     async with workspace.app(size=(120, 35)) as ui:
         view = ui.conversation
         agent = ui.detached_agent("closing")
         view.bind_agent(agent)
-        agent.coordination = CoordinationChangedUpdate(
-            thread=thread.incarnation, wire_root=str(workspace.roots.wire), owner=ProcessIdentity.capture(os.getpid()),
-            worktree=str(project), model="test-model", thinking_level=None, title=thread.name,
-            context_usage=UnmeasuredContext())
+        agent.coordination = registered_coordination_fact(
+            comms, "source", owner=ProcessIdentity.capture(os.getpid()), model="test-model", worktree=project)
         await deliver(agent, {
             "sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": ""},
             "_meta": encode_updates(agent.coordination)})
diff --git a/tests/textual/test_arrival_fade.py b/tests/textual/test_arrival_fade.py
index 2dfea5aca..ded24fbda 100644
--- a/tests/textual/test_arrival_fade.py
+++ b/tests/textual/test_arrival_fade.py
@@ -5,9 +5,8 @@
 import json
 import os
 
-from agent_comms.acp_extension import CoordinationChangedUpdate, TurnChangedUpdate
+from agent_comms.acp_extension import TurnChangedUpdate
 from agent_comms.child_process import ProcessIdentity
-from agent_comms.context_reading import UnmeasuredContext
 from agent_comms.field_codec import TurnId
 from agent_comms.turn_lease import ActiveTurn, FinishedTurn, TurnState
 from agent_comms.turn_phase import PublishingPhase
@@ -16,7 +15,7 @@
 from ui_theme.effects import NEW_TEXT
 
 from support.agents import deliver
-from support.coordination import carrying
+from support.coordination import carrying, registered_coordination_fact
 from support.textual_steps import tool_status
 from toad.frame_presentation import EFFECT_UNSTARTED
 from toad.widgets.agent_response import AgentResponse
@@ -48,17 +47,15 @@ async def test_a_commit_retiring_a_live_block_still_fades_another_senders_row(wo
         "timestamp": "2026-10-22T00:00:00Z", "message": {
         "role": "user", "content": [{"type": "text", "text": "ASK_ROW"}]}}) + "\n"
     saved.write_text(prompt)
-    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
+    comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
     workspace.thread("peer")
     comms.messaging.send("peer", "source", "EARLIER_NOTICE")
     async with workspace.app(size=(120, 35)) as ui:
         view = ui.conversation
         agent = ui.detached_agent("arrival-fade")
         view.bind_agent(agent)
-        coordination = CoordinationChangedUpdate(
-            thread=thread.incarnation, wire_root=str(workspace.roots.wire), owner=ProcessIdentity.capture(os.getpid()),
-            worktree=str(project), model="test-model", thinking_level=None, title=thread.name,
-            context_usage=UnmeasuredContext())
+        coordination = registered_coordination_fact(
+            comms, "source", owner=ProcessIdentity.capture(os.getpid()), model="test-model", worktree=project)
         agent.coordination = coordination
         await deliver(agent, carrying(coordination))
         agent.presentation.managed_turns().receive(
diff --git a/tests/textual/test_channel_partial_paint.py b/tests/textual/test_channel_partial_paint.py
index 4886825f5..fd9f75b17 100644
--- a/tests/textual/test_channel_partial_paint.py
+++ b/tests/textual/test_channel_partial_paint.py
@@ -1,6 +1,6 @@
 """Textual-facing: read marks follow the message text the viewport has painted; scrolling the window reveals the rest."""
 
-from support.textual_steps import text_rows_of
+from agent_comms.read_basis import ChannelConversation
 
 
 async def test_partial_viewport_marks_only_painted_text_and_scrolling_marks_the_remaining_messages(workspace):
@@ -12,40 +12,59 @@ async def test_partial_viewport_marks_only_painted_text_and_scrolling_marks_the_
     viewer = workspace.viewer
     rows = [comms.messaging.send("peer", "#team", f"row {i}\n" + "body\n" * 10).message for i in range(8)]
 
-    def seen():
-        return comms.bus.reads.seen_sequences(viewer, comms.registry.snapshot())
+    ledger = comms.bus.reads
+    snapshot = comms.registry.snapshot()
+    channel = ChannelConversation("#team")
 
     async with workspace.app(size=(90, 24)) as ui:
         app = ui.textual[0]
         await ui.open_channel("#team", actor=viewer)
         chat = app.screen.query_one(CommsChatView)
         history, window = chat.message_history, chat.window
-        await ui.until(lambda: seen() and not history.marking, why="the first viewport to be marked")
+        await ui.until(lambda: any(row.seq in ledger.cursor(
+            viewer, snapshot).for_conversation(channel) for row in rows),
+            why="the first viewport to be marked")
+        await history.read_worker.wait()
         in_view = {fragment.message.seq for fragment in history.read_fragments()}
-        assert 0 < len(seen()) < len(rows)
-        assert seen() == in_view
-        # Messages beyond the first viewport stay unread until their text is scrolled to.
+        assert 0 < len(in_view) < len(rows)
+        cursor = ledger.cursor(viewer, snapshot).for_conversation(channel)
+        assert all((message.seq in cursor) == (message.seq in in_view) for message in rows)
+        # Address each original message through the history's own source
+        # fragments; revealing its head prepares virtualized rows on demand.
         for message in reversed(rows):
-            if message.seq in seen():
+            if message.seq in ledger.cursor(viewer, snapshot).for_conversation(channel):
                 continue
-            if text_rows_of(chat, message) is None:
-                # Earlier messages are laid out as the reader nears them.
-                window.scroll_to(y=0, animate=False, immediate=True)
-                await ui.until(lambda: text_rows_of(chat, message) is not None, why=f"row {message.seq} laid out")
-            window.scroll_to(y=text_rows_of(chat, message).start, animate=False, immediate=True)
-            await ui.until(lambda: message.seq in seen(), why=f"row {message.seq} to be marked once scrolled to")
-        assert seen() == {row.seq for row in rows}
+            fragment = next(part for part in history.fragments
+                            if part.message.view_key == message.view_key)
+            for _ in rows:
+                if history.draws_fragment(fragment):
+                    break
+                window.scroll_page_up(animate=False)
+                await ui.step()
+                await ui.until(lambda: window.reader_position.laid_out(window),
+                               why="the scrolled viewport to finish layout")
+            await history.reveal_message_head(fragment, lambda: True)
+            history.show_fragment(fragment)
+            await ui.until(lambda: message.seq in ledger.cursor(
+                viewer, snapshot).for_conversation(channel),
+                why=f"row {message.seq} to be marked once scrolled to")
+        cursor = ledger.cursor(viewer, snapshot).for_conversation(channel)
+        assert all(row.seq in cursor for row in rows)
         # A wholly read history has nothing left to mark: no further worker or route read.
-        await ui.until(lambda: not history.marking, why="marking to settle")
-        assert all(fragment.evidence is None for fragment in history.wire_fragments)
+        await history.read_worker.wait()
+        assert all(fragment.evidence is None for fragment in history.fragments)
+        completed = history.read_worker
         for _ in range(50):
             history.mark_visible()
-            assert not history.marking
-        recorded = tuple(message.view_key for message in history.messages)
+            assert history.read_worker is completed
+        window.scroll_end(animate=False, immediate=True)
+        await ui.until(lambda: ui.shows("row 7"), why="the newest row after returning to the bottom")
+        recorded = tuple(fragment.message.view_key for fragment in history.fragments)
         await ui.press("ctrl+t")
-        await ui.until(lambda: isinstance(history.style, MarkdownMessageStyle), why="the Markdown style")
+        await ui.until(lambda: isinstance(history.page_source.loader.style, MarkdownMessageStyle), why="the Markdown style")
         assert ui.shows("row 7")
         await ui.press("ctrl+t")
-        await ui.until(lambda: isinstance(history.style, IrcMessageStyle), why="the IRC style")
-        assert tuple(message.view_key for message in history.messages) == recorded
-        assert seen() == {row.seq for row in rows}
+        await ui.until(lambda: isinstance(history.page_source.loader.style, IrcMessageStyle), why="the IRC style")
+        assert tuple(fragment.message.view_key for fragment in history.fragments) == recorded
+        cursor = ledger.cursor(viewer, snapshot).for_conversation(channel)
+        assert all(row.seq in cursor for row in rows)
diff --git a/tests/textual/test_channel_read_ahead.py b/tests/textual/test_channel_read_ahead.py
index 0e29e002f..996753c6e 100644
--- a/tests/textual/test_channel_read_ahead.py
+++ b/tests/textual/test_channel_read_ahead.py
@@ -2,6 +2,8 @@
 
 from unittest.mock import patch
 
+from agent_comms.page_traversal import OlderTraversal
+
 from toad.widgets.comms_chat import CommsChatView
 
 
@@ -11,30 +13,30 @@ async def test_a_channel_reads_the_next_older_page_before_scrolling_reaches_its_
         workspace.comms.messaging.send("sender", "#scroll", f"Message {index}: " + "wrapped body " * 6)
     async with workspace.app(size=(145, 50)) as ui:
         app, _pilot = ui.textual
-        views = app.coordination_access.service.views
+        views = app.coordination_access.observation.service.views
         earlier_reads = []
         original = views.channel_display_page
 
-        def counted(target, **bounds):
-            if bounds.get("before") is not None:
-                earlier_reads.append(bounds["limit"])
-            return original(target, **bounds)
+        def counted(target, *, bounds, **request):
+            if isinstance(bounds.traversal, OlderTraversal):
+                earlier_reads.append(bounds.limit)
+            return original(target, bounds=bounds, **request)
 
         with patch.object(views, "channel_display_page", side_effect=counted):
             await ui.open_channel("#scroll", actor="sender")
             chat = app.screen.query_one(CommsChatView)
             history = chat.message_history
-            await ui.until(lambda: history.messages, why="channel messages")
-            source = history.source
+            await ui.until(lambda: history.fragments, why="channel messages")
+            source = history.page_source.loader
             assert source.comms.views is views
             # The newest messages come as a short page drawn at once; every earlier read is a whole page.
-            await ui.until(lambda: len(history.messages) > source.tail_limit and len(earlier_reads) >= 3
+            await ui.until(lambda: len(history.fragments) > source.tail_limit and len(earlier_reads) >= 3
                            and history.state.accepts_source_work, why="earlier messages and a page read ahead")
             assert earlier_reads[0] == source.tail_limit and set(earlier_reads[1:]) == {source.page_limit}, earlier_reads
-            oldest, reads = history.messages[0].seq, len(earlier_reads)
+            oldest, reads = history.fragments[0].message.seq, len(earlier_reads)
             chat.window.focus()
             for _ in range(6):
                 await ui.press("pageup")
-            await ui.until(lambda: history.messages[0].seq <= oldest - 24, why="earlier messages drawn while paging up")
+            await ui.until(lambda: history.fragments[0].message.seq <= oldest - 24, why="earlier messages drawn while paging up")
             # Six pages of scrolling did not cost six Core reads.
             assert len(earlier_reads) - reads <= 3 and set(earlier_reads[1:]) == {source.page_limit}, earlier_reads
diff --git a/tests/textual/test_channel_scope_rows.py b/tests/textual/test_channel_scope_rows.py
index b8e10e58e..c899e3cb1 100644
--- a/tests/textual/test_channel_scope_rows.py
+++ b/tests/textual/test_channel_scope_rows.py
@@ -11,11 +11,11 @@ async def test_any_mode_channel_rows_include_older_messages_after_a_refresh(work
     for name, tags in (("alice", ("team",)), ("bob", ()), ("carol", ()), ("dave", ())):
         workspace.thread(name, *tags, running=True)
     comms = workspace.comms
-    project = str(workspace.project)
+    viewer = comms.views.display_viewer(Checkout(workspace.project))
     comms.messaging.send("carol", "dave", "older newly visible DM")
     comms.messaging.send("alice", "#team", "current channel message")
     comms.channels.set_channel_any_mode("#team", True)
-    assert [m.body for m in comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project))).messages] == [
+    assert [m.body for m in comms.views.channel_display_page("#team", viewer=viewer).messages] == [
         "current channel message"]
 
     async with workspace.app(size=(100, 30)) as ui:
@@ -30,9 +30,9 @@ async def test_any_mode_channel_rows_include_older_messages_after_a_refresh(work
         await ui.app.coordination_access.stop_observing()
         history = chat.message_history
         source, snapshot = history.page_source, history.source_snapshot()
-        marked = comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project)))
+        marked = comms.views.channel_display_page("#team", viewer=viewer)
         comms.channels.update_tags("carol", add=frozenset({"team"}))
-        expanded = comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project)))
+        expanded = comms.views.channel_display_page("#team", viewer=viewer)
         assert [m.body for m in expanded.messages][:2] == ["older newly visible DM", "current channel message"]
         await history._mark_page(marked, (), snapshot)
         await ui.until(lambda: history.page_source is not source,
diff --git a/tests/textual/test_checkpoint_first_source.py b/tests/textual/test_checkpoint_first_source.py
index ea37657d5..4d7e1bbd2 100644
--- a/tests/textual/test_checkpoint_first_source.py
+++ b/tests/textual/test_checkpoint_first_source.py
@@ -8,13 +8,11 @@
 from dataclasses import replace
 import os
 
-from agent_comms.acp_extension import CoordinationChangedUpdate
 from agent_comms.child_process import ProcessIdentity
-from agent_comms.context_reading import UnmeasuredContext
 
 from support import journals
 from support.agents import deliver
-from support.coordination import carrying
+from support.coordination import carrying, registered_coordination_fact
 from toad.live_output import ResponseStream
 from toad.transcript_publication import CommitPublication
 from toad.widgets.agent_response import AgentResponse
@@ -31,10 +29,8 @@ async def test_first_native_file_replaces_the_wire_history_and_the_covered_live_
         view = ui.conversation
         agent = ui.detached_agent("checkpoint-first")
         ui.attach_agent(agent)
-        agent.coordination = coordination = CoordinationChangedUpdate(
-            thread=thread.incarnation, wire_root=str(workspace.roots.wire), owner=ProcessIdentity.capture(os.getpid()),
-            worktree=str(workspace.project), model="test-model", thinking_level=None, title=thread.name,
-            context_usage=UnmeasuredContext())
+        agent.coordination = coordination = registered_coordination_fact(
+            comms, "source", owner=ProcessIdentity.capture(os.getpid()), model="test-model", worktree=workspace.project)
         await deliver(agent, carrying(coordination))
         agent.session.settled.set()
         # The bound view mounts the wire history itself; the test reads that one, never a copy.
diff --git a/tests/textual/test_command_pane_and_resume_modal.py b/tests/textual/test_command_pane_and_resume_modal.py
index 64bdeb0f4..4bedca07c 100644
--- a/tests/textual/test_command_pane_and_resume_modal.py
+++ b/tests/textual/test_command_pane_and_resume_modal.py
@@ -58,5 +58,5 @@ async def test_the_resume_modal_lists_a_saved_session_and_returns_it_typed(tmp_p
             assert modal.session_table.get_row(str(pk))[:2] == ["Agent", "Saved session"]
             await modal.dissmiss_with_session(str(pk))
             await pilot.pause()
-            assert resumed[0].meta_json.cwd == tmp_path
+            assert resumed[0].meta_json.cwd == Checkout(tmp_path)
             assert resumed[0].agent_session_id == "saved-session"
diff --git a/tests/textual/test_context_tree_keeps_overlapping_annotations.py b/tests/textual/test_context_tree_keeps_overlapping_annotations.py
index adb6b4e2e..e7245d7e6 100644
--- a/tests/textual/test_context_tree_keeps_overlapping_annotations.py
+++ b/tests/textual/test_context_tree_keeps_overlapping_annotations.py
@@ -1,70 +1,70 @@
-"""Textual-facing: the context tree (a Textual Tree widget) selects, scrolls and remounts overlapping annotations without losing the reader's row or position."""
+"""The real context panel keeps mounted navigation and rebinds detail after tree retirement."""
 
-from textual.app import App
-from toad.textual_colours import ThemedApp
-from toad.widgets.context_explorer import ContextTree, ContextTreeIntent
+from toad.widgets.context_explorer import ContextExplorer, ContextTree
+from toad.widgets.side_bar import SideBar, SideBarCollapsible
 
 from support.overlapping_annotations import overlapping_annotations
 
 
-class TreeApp(ThemedApp, App):
-    def __init__(self, tree):
-        super().__init__()
-        self.subject = tree
-
-    def compose(self):
-        yield self.subject
-
-
-async def test_tree_rebinds_corrections_and_remounts_on_the_same_row_without_a_new_selection(tmp_path):
-    found = overlapping_annotations(tmp_path)
+async def test_tree_rebinds_corrections_and_remounts_on_the_same_row_without_a_new_selection(workspace):
+    found = overlapping_annotations(workspace.project)
     spans, labels, inspection, short, long = found.spans, found.labels, found.inspection, found.short, found.long
     groups = found.refined_groups
-    intent = ContextTreeIntent(selected=short)
-    detail = []
-    tree = ContextTree(intent, detail.append, lambda text: None)
-    tree.styles.height = 3
-    app = TreeApp(tree)
-    async with app.run_test() as pilot:
+    async with workspace.app(size=(120, 45)) as ui:
+        app = ui.textual[0]
+        sidebar = app.screen.query_one("#thread-sidebar", SideBar)
+        sidebar.reveal()
+        await sidebar.wait_content_ready()
+        next(panel for panel in sidebar.query(SideBarCollapsible) if panel.title == "Context").collapsed = False
+        explorer = sidebar.query_one(ContextExplorer)
+        await ui.step()
+        tree = explorer.query_one(ContextTree)
+        tree.styles.height = 3
+        tree.styles.min_height = 0
+        intent = explorer.intent
+        intent.selected = short
         tree.present(inspection.working_memory())
-        await pilot.pause()
-        short_node, long_node = tree.reveal(short.key), tree.reveal(long.key)
+        short_node, long_node = tree.resolve_reader(short.key), tree.resolve_reader(long.key)
+        tree.root.expand_all()
+        await ui.step()
+        tree.move_cursor(short_node, animate=False)
+        await ui.step()
         assert short_node is not long_node
         assert tree.owns_node(short_node) and tree.owns_node(long_node)
         assert tree.cursor_node is short_node and intent.selected is short_node.data
         assert short_node.data.annotation.span == spans[0]
         assert long_node.data.annotation.span == spans[1]
 
-        # A wheel scroll can leave the selected native row outside the
-        # viewport. Rebinding its source is not a new navigation request.
+        # Rebinding selected detail is not another native navigation request.
         tree.scroll_home(animate=False)
-        await pilot.pause()
+        await ui.step()
         assert short_node.line >= tree.scrollable_content_region.height
         reader_offset = tree.scroll_offset
-        detail.clear()
         tree.present(inspection.working_memory())
-        await pilot.pause()
+        await ui.step()
         assert tree.cursor_node is short_node
         assert tree.scroll_offset == reader_offset
-        assert detail == [short_node.data]
+        assert intent.selected is short_node.data
 
         tree.present(groups)
-        await pilot.pause()
-        short_node, long_node = tree.reveal(short.key), tree.reveal(long.key)
+        await ui.step()
+        short_node, long_node = tree.resolve_reader(short.key), tree.resolve_reader(long.key)
         assert tree.owns_node(short_node) and tree.owns_node(long_node)
-        assert tree.cursor_node is short_node and intent.selected is short_node.data
+        # A human correction moves the answer to another section. The old
+        # native row retires; detail rebinds without manufacturing navigation.
+        assert intent.selected is short_node.data
         assert short_node.data.annotation == found.corrected
         assert long_node.data.annotation == labels[1]
-        assert detail[-1] is short_node.data
 
         await tree.remove()
-        restored = ContextTree(intent, detail.append, lambda text: None)
-        await app.mount(restored)
+        restored = ContextTree(explorer._show_detail, explorer._show_placeholder)
+        await explorer.mount(restored)
         restored.present(groups)
-        await pilot.pause()
-        short_node, long_node = restored.reveal(short.key), restored.reveal(long.key)
+        await ui.step()
+        short_node, long_node = restored.resolve_reader(short.key), restored.resolve_reader(long.key)
         assert restored.owns_node(short_node) and restored.owns_node(long_node)
-        assert restored.cursor_node is short_node and intent.selected is short_node.data
+        # Detail intent survives. Retired native nodes do not own a new cursor.
+        assert intent.selected is short_node.data
+        assert restored.cursor_node is not short_node
         assert short_node.data.annotation == found.corrected and long_node.data.annotation == labels[1]
-        assert detail[-1] is short_node.data
     assert app._exception is None
diff --git a/tests/textual/test_conversation_value.py b/tests/textual/test_conversation_value.py
index d13e1f935..1964d34d6 100644
--- a/tests/textual/test_conversation_value.py
+++ b/tests/textual/test_conversation_value.py
@@ -190,7 +190,7 @@ async def test_recreated_dm_peer_retires_the_old_view_and_opens_a_new_incarnatio
         def replace_peer():
             registry = workspace.comms.registry
             registry.unregister(peer.name)
-            registry.remove_originals((registry.require(peer.name),))
+            workspace.comms.threads.delete(peer.name)
             return registry.declare(Thread(
                 peer.name, peer.tags, peer.worktree, created_at=peer.created_at + 1,
             ))
diff --git a/tests/textual/test_cursor_request_results.py b/tests/textual/test_cursor_request_results.py
index 52024ee4c..413f4ec14 100644
--- a/tests/textual/test_cursor_request_results.py
+++ b/tests/textual/test_cursor_request_results.py
@@ -9,6 +9,7 @@
     CursorAdvancedUpdate, CursorEnvelope, LiveCursorScope, VerifiedCursorObservation, encode_updates,
 )
 from agent_comms.child_process import ProcessIdentity
+from agent_comms.field_codec import AcpSessionId, RootId
 from agent_comms.native_input_record import FullNativeExecution, NumberNativeColumn, TextNativeColumn
 from agent_comms.native_runtime_input import CurrentNativeCursor
 from agent_comms.thread_identity import ClaimedIdentity, RecordedThreadIncarnation
@@ -22,13 +23,14 @@
 
 def response(session):
     owner = ClaimedIdentity(RecordedThreadIncarnation("owner", 1000.0), 1)
+    root = RootId("a" * 32)
     cursor = CurrentNativeCursor(
-        wire_root_id="a" * 32, recipient_lookup=owner.incarnation.lookup, owner_generation=1,
+        wire_root_id=root, recipient_lookup=owner.incarnation.lookup, owner_generation=1,
         owner_claimed_generation=1, covered_seq=2, injected_seq=2, input_id=TextNativeColumn("b" * 32),
         stage=TextNativeColumn(FullNativeExecution.declared_name), session_id=TextNativeColumn("native-session"),
         request_generation=NumberNativeColumn(1))
     fact = CursorAdvancedUpdate(CursorEnvelope(
-        LiveCursorScope(session_id=session, wire_root_id="a" * 32, claimed_identity=owner,
+        LiveCursorScope(session_id=AcpSessionId(session), wire_root_id=root, claimed_identity=owner,
                         owner=ProcessIdentity(1234, 1)), 1, VerifiedCursorObservation(cursor)))
     return NewSessionResponse(sessionId=session, _meta=encode_updates(fact))
 
diff --git a/tests/textual/test_divider_only_viewport_stays_unread.py b/tests/textual/test_divider_only_viewport_stays_unread.py
index f12b88a91..b0e8f86c5 100644
--- a/tests/textual/test_divider_only_viewport_stays_unread.py
+++ b/tests/textual/test_divider_only_viewport_stays_unread.py
@@ -1,6 +1,7 @@
 """Textual-facing: whether a channel message counts as read depends on which rows Textual painted, and a divider painted without its text is not enough."""
 
 from agent_comms.repo import Checkout
+from agent_comms.ui_model.unread import ExactUnread
 from unittest.mock import patch
 
 from support.textual_steps import text_rows_of
@@ -11,7 +12,7 @@
 async def test_divider_without_body_leaves_the_message_unread(workspace):
     workspace.thread("peer", "team")
     comms = workspace.comms
-    project = str(workspace.project)
+    project = workspace.project
     for index in range(7):
         comms.messaging.send("peer", "#team", f"earlier {index} " + "body " * 20)
     comms.views.mark_user_view_read("#team", worktree=Checkout(project))
@@ -49,6 +50,8 @@ def caption_is_last_row():
             assert all(fragment.message.seq != last.seq for fragment in history.read_fragments())
 
             mark_visible(history)
-            await ui.until(lambda: not history.marking, why="marking to settle")
-            await ui.step(.2)
-            assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] == 1
+            # Earlier visible messages may have work; completing that work
+            # must not acknowledge the unread message whose text was hidden.
+            await history.read_worker.wait()
+            await ui.step()
+            assert comms.views.viewer_snapshot(Checkout(project)).channel_unread["#team"] == ExactUnread(count=1)
diff --git a/tests/textual/test_dm_peer_replacement.py b/tests/textual/test_dm_peer_replacement.py
index 7dce38b89..89dd57652 100644
--- a/tests/textual/test_dm_peer_replacement.py
+++ b/tests/textual/test_dm_peer_replacement.py
@@ -4,6 +4,7 @@
 from threading import Event
 from unittest.mock import patch
 
+from agent_comms.channel_targets import CommandTarget
 from agent_comms.history_views import HistoryViews
 
 
@@ -13,25 +14,29 @@ async def test_a_page_painted_for_the_old_peer_does_not_mark_the_replacement_pee
     workspace.thread("peer", running=True)
     comms = workspace.comms
     viewer = workspace.viewer
+    target = CommandTarget.named("peer")
     comms.messaging.send("peer", viewer, "old peer painted")
-    entered, release = Event(), Event()
-    original_mark = HistoryViews.mark_dm_view_read
+    entered, release, marked = Event(), Event(), Event()
+    original_mark = HistoryViews.mark_page_read
 
-    def pause_old_mark(self, peer, **kwargs):
+    def pause_old_mark(self, page, **kwargs):
         entered.set()
-        if not release.wait(8):
-            raise TimeoutError("Old painted marker was not released")
-        return original_mark(self, peer, **kwargs)
+        try:
+            if not release.wait(8):
+                raise TimeoutError("Old painted marker was not released")
+            return original_mark(self, page, **kwargs)
+        finally:
+            marked.set()
 
     try:
-        with patch.object(HistoryViews, "mark_dm_view_read", pause_old_mark):
+        with patch.object(HistoryViews, "mark_page_read", pause_old_mark):
             async with workspace.app(size=(120, 40)) as ui:
                 owner = ui.selected_tab
                 await ui.open_direct("peer", actor=viewer)
                 chat = ui.textual[0].screen.query_one(CommsChatView)
                 assert await asyncio.to_thread(entered.wait, 4)
-                assert [message.body for message in chat.message_history.messages] == ["old peer painted"]
-                assert comms.bus.pending_count(viewer, "peer") == 1
+                assert [fragment.message.body for fragment in chat.message_history.fragments] == ["old peer painted"]
+                assert comms.bus.pending_count(viewer, target) == 1
 
                 comms.registry.unregister("peer")
                 comms.registry.remove("peer")
@@ -39,8 +44,10 @@ def pause_old_mark(self, peer, **kwargs):
                 comms.messaging.send("peer", viewer, "new peer never painted")
                 await ui.select_tab(owner)
                 release.set()
-                await ui.until(lambda: not chat.message_history.marking, why="the held mark to finish")
-                assert comms.bus.pending_count(viewer, "peer") == 1
-                assert not any(message.body == "new peer never painted" for message in chat.message_history.messages)
+                # The old page's mark runs in its thread even when leaving the tab cancels its worker.
+                assert await asyncio.to_thread(marked.wait, 8)
+                assert comms.bus.pending_count(viewer, target) == 1
+                assert not any(fragment.message.body == "new peer never painted"
+                               for fragment in chat.message_history.fragments)
     finally:
         release.set()
diff --git a/tests/textual/test_file_links_in_agent_text.py b/tests/textual/test_file_links_in_agent_text.py
index 3c8620188..10ea12c1f 100644
--- a/tests/textual/test_file_links_in_agent_text.py
+++ b/tests/textual/test_file_links_in_agent_text.py
@@ -2,6 +2,7 @@
 
 from unittest.mock import patch
 
+from toad.project_link import DirectProjectFileLink, ExternalProjectLink, SearchProjectFileLink
 from toad.widgets.comms_menu import ContextMenu, ContextMenuItem
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.project_panel import FilePreview
@@ -25,10 +26,13 @@ async def test_existing_project_paths_in_agent_text_are_links_and_other_text_is_
     async with workspace.app(size=(130, 35)) as ui:
         await say(ui, "See plans/tag-channel-view-plan.md:\n\n`plans/tag-channel-view-plan.md`\n\n"
                       "but not plans/missing.md\n\nor https://example.com/plans/tag-channel-view-plan.md.")
-        assert ui.hyperlink_at("plans/tag-channel-view-plan.md:", occurrence=0) == f"toad-file:{path}"
-        assert ui.hyperlink_at("plans/tag-channel-view-plan.md", occurrence=1) == f"toad-file:{path}"
-        assert ui.hyperlink_at("plans/missing.md") is None
-        assert ui.hyperlink_at("https://example.com/plans/tag-channel-view-plan.md.") is None
+        plan = DirectProjectFileLink(path, original_root=workspace.project.resolve()).click
+        assert ui.link_at("plans/tag-channel-view-plan.md:", occurrence=0) == plan
+        assert ui.link_at("plans/tag-channel-view-plan.md", occurrence=1) == plan
+        assert ui.link_at("plans/missing.md") is None
+        # Linkify's URLs are resolved syntax; the sentence's full stop is not part of the address.
+        assert ui.link_at("https://example.com/plans/tag-channel-view-plan.md") == ExternalProjectLink(
+            "https://example.com/plans/tag-channel-view-plan.md").click
 
 
 async def test_explicit_links_name_the_file_search_or_the_website(workspace):
@@ -36,9 +40,11 @@ async def test_explicit_links_name_the_file_search_or_the_website(workspace):
     async with workspace.app(size=(130, 35)) as ui:
         await say(ui, f"[Absolute]({path}) [Relative](plans/tag-channel-view-plan.md) "
                       "[Website](https://example.com/preview.py) [Missing](missing.py)")
-        assert [ui.hyperlink_at(name) for name in ("Absolute", "Relative", "Website", "Missing")] == [
-            f"toad-file:{path}", f"toad-file:{path}",
-            "https://example.com/preview.py", "toad-file-search:missing.py"]
+        root = workspace.project.resolve()
+        assert [ui.link_at(name) for name in ("Absolute", "Relative", "Website", "Missing")] == [
+            DirectProjectFileLink(path, original_root=root).click, DirectProjectFileLink(path, original_root=root).click,
+            ExternalProjectLink("https://example.com/preview.py").click,
+            SearchProjectFileLink("missing.py", original_root=root).click]
 
 
 async def test_clicking_a_file_link_opens_its_preview_tab(workspace):
@@ -82,7 +88,8 @@ async def test_a_bare_file_name_links_to_a_search_that_opens_the_one_matching_fi
         app, _ = ui.textual
         owner = ui.selected_tab
         await say(ui, "See declarations.py")
-        assert ui.hyperlink_at("declarations.py") == "toad-file-search:declarations.py"
+        assert ui.link_at("declarations.py") == SearchProjectFileLink(
+            "declarations.py", original_root=workspace.project.resolve()).click
         await ui.click("declarations.py")
         await ui.until(lambda: ui.selected_tab != owner, why="the preview of the nested file")
         preview = app.selected_session.query_one(FilePreview)
diff --git a/tests/textual/test_hidden_channel_refresh.py b/tests/textual/test_hidden_channel_refresh.py
index d9c383c5e..24e7a295a 100644
--- a/tests/textual/test_hidden_channel_refresh.py
+++ b/tests/textual/test_hidden_channel_refresh.py
@@ -28,9 +28,10 @@ async def test_hidden_channel_refresh_does_no_root_check_and_resumed_window_fetc
         with patch.object(RouteSelection, "root_is_current", side_effect=AssertionError("hidden route check")):
             for _ in range(40):
                 await refresh_chat(chat)
-            assert not any(m.body == "ARRIVED-WHILE-HIDDEN" for m in history.messages)
+            assert not any(fragment.message.body == "ARRIVED-WHILE-HIDDEN" for fragment in history.fragments)
         await ui.select_tab(first)
-        await ui.until(lambda: any(m.body == "ARRIVED-WHILE-HIDDEN" for m in history.messages),
+        await ui.until(lambda: any(fragment.message.body == "ARRIVED-WHILE-HIDDEN"
+                                   for fragment in history.fragments),
                        why="the resumed channel to fetch the current message")
         # The history holds the message before layout has measured its rows; the next layout draws it.
         await ui.until(lambda: ui.shows("ARRIVED-WHILE-HIDDEN"), why="the fetched message drawn")
diff --git a/tests/textual/test_hidden_session_retirement.py b/tests/textual/test_hidden_session_retirement.py
index fdf1135ff..6441608fd 100644
--- a/tests/textual/test_hidden_session_retirement.py
+++ b/tests/textual/test_hidden_session_retirement.py
@@ -2,7 +2,6 @@
 
 from agent_comms.repo import Checkout
 from collections import Counter
-from pathlib import Path
 import sys
 
 from agent_comms.comms import Comms
@@ -34,7 +33,7 @@ async def open_hidden_sessions(workspace, ui, owners):
     first = ui.selected_tab
 
     def source(owner):
-        return MainScreen.from_thread(Path(owner.worktree), str(workspace.comms.root), owner.incarnation)
+        return MainScreen.from_thread(owner.worktree, str(workspace.comms.root), owner.incarnation)
 
     modes = {}
     for owner in owners[1:]:
@@ -121,7 +120,7 @@ async def test_a_same_name_successor_does_not_keep_the_deleted_threads_hidden_se
         _, gone = app.session_navigation.get(modes["agent-3"]).original_threads(app.session_navigation)[0]
         old = registry.require(gone.name)
         registry.unregister(old.name)
-        registry.remove_originals((registry.require(old.name),))
+        workspace.comms.threads.delete(old.name)
         registry.declare(Thread(old.name, old.tags, old.worktree, created_at=old.created_at + 1))
         await app.session_navigation.retire_missing()
         await ui.until(lambda: app.session_navigation.get(modes["agent-3"]) is None, seconds=15,
diff --git a/tests/textual/test_saved_view_read_only.py b/tests/textual/test_saved_view_read_only.py
index b03cc3939..c1fb1fe47 100644
--- a/tests/textual/test_saved_view_read_only.py
+++ b/tests/textual/test_saved_view_read_only.py
@@ -25,18 +25,19 @@ async def test_saved_view_shows_union_history_keeps_its_pin_and_refuses_to_send(
             "engineering", ViewPredicate(AnyOfMatch, frozenset({"api", "ui"})), 17,
             frozenset({"#engineering"}))},
     ))
-    before = comms.bus.log.path.read_bytes()
+    before = comms.views.latest_sequence()
 
     async with workspace.app(size=(110, 35)) as ui:
         sidebar = await channel_roster(ui, "#engineering")
         next(group for group in sidebar.query(ChannelGroup) if group.row.target_name == "#engineering").row.focus()
         await ui.press("enter")
         chat = ui.textual[0].screen.query_one(CommsChatView)
-        await ui.until(lambda: [m.body for m in chat.message_history.messages] == [m.body for m in rows],
+        await ui.until(lambda: [fragment.message.body for fragment in chat.message_history.fragments]
+                       == [message.body for message in rows],
                        why="the saved view's history")
         assert chat.prompt.prompt_text_area.disabled
-        await ui.until(lambda: "Read-only" in chat.status, why="the read-only status")
+        await ui.until(lambda: "Read-only" in chat.thread_status, why="the read-only status")
         await chat.submit_input(input_events.UserInputSubmitted("must not publish"))
         assert chat.prompt.text == "must not publish"
-        assert comms.bus.log.path.read_bytes() == before
+        assert comms.views.latest_sequence() == before
         assert next(view for view in comms.views.channel_views() if view.channel.name == "#engineering").channel.pinned
diff --git a/tests/textual/test_snapshot_cancellation.py b/tests/textual/test_snapshot_cancellation.py
index c859d7b2c..b6430c7f4 100644
--- a/tests/textual/test_snapshot_cancellation.py
+++ b/tests/textual/test_snapshot_cancellation.py
@@ -9,15 +9,13 @@
 import os
 from unittest.mock import patch
 
-from agent_comms.acp_extension import CoordinationChangedUpdate
 from agent_comms.child_process import ProcessIdentity
-from agent_comms.context_reading import UnmeasuredContext
 import pytest
 from textual.worker import WorkerCancelled
 
 from support import journals
 from support.agents import deliver
-from support.coordination import carrying
+from support.coordination import carrying, registered_coordination_fact
 from toad.live_output import ResponseStream
 from support.textual_steps import saved_history
 
@@ -42,11 +40,8 @@ async def test_cancelled_commit_with_the_output_owner_busy_retires_the_old_histo
         old = await saved_history(view, old_page, view.transcript.get_transcript_page)
         await view.contents.mount(old)
         await ui.until(lambda: old in view.window.histories, why="the old history")
-        thread = workspace.comms.registry.require("replacement")
-        agent.coordination = coordination = CoordinationChangedUpdate(
-            thread=thread.incarnation, wire_root=str(workspace.roots.wire), owner=ProcessIdentity.capture(os.getpid()),
-            worktree=str(workspace.project), model="test-model", thinking_level=None, title=thread.name,
-            context_usage=UnmeasuredContext())
+        agent.coordination = coordination = registered_coordination_fact(
+            workspace.comms, "replacement", owner=ProcessIdentity.capture(os.getpid()), model="test-model", worktree=workspace.project)
         await deliver(agent, carrying(coordination))
         agent.session.settled.set()
         # A live stream block that the saved page already covers.