Toad suite: find the conftest.py:93 hang, fix link and Core-type drift

.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'

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(""))

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)
 

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))

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:

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")

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")

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,55 +12,54 @@ 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")
 
 
 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")
 
 
 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")

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

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))

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: