P1022: migrate stale Core test callers and verify main pushes

.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,18 +58,10 @@ 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'
-      - name: Install Neovim
-        run: |
-          curl --fail --location --silent --show-error \
-            https://github.com/neovim/neovim/releases/download/v0.12.4/nvim-linux-x86_64.tar.gz \
-            --output "$RUNNER_TEMP/neovim.tar.gz"
-          tar --extract --gzip --file "$RUNNER_TEMP/neovim.tar.gz" --directory "$RUNNER_TEMP"
-          echo "$RUNNER_TEMP/nvim-linux-x86_64/bin" >> "$GITHUB_PATH"
-          "$RUNNER_TEMP/nvim-linux-x86_64/bin/nvim" --version
       - uses: ./.github/actions/core-from-main
         with:
           path: .artifacts/core
@@ -88,6 +82,8 @@ jobs:
           with open(os.environ['GITHUB_ENV'], 'a') as env:
               env.write(f'AC_NATIVE_COPIED_PACKAGE={package.resolve()}\n')
           PYCORE
+      - name: Prepare the declared Neovim requirement
+        run: .venv/bin/python -m toad.neovim_runtime
       # xdist reports a collection error as a worker crash; collect alone first so it reads as one.
       - run: .venv/bin/pytest --collect-only -q
       # A test stuck past 120 s dumps every thread, so the red names where it hung.

pyproject.toml

@@ -52,6 +52,9 @@ textual-diff-view = { git = "https://github.com/OpenHCSDev/textual-diff-view.git
 [project.scripts]
 toad = "toad.cli:main"
 
+[project.entry-points."agent_comms.runtime_preparation"]
+neovim = "toad.neovim_runtime:Neovim"
+
 [project.entry-points."agent_comms.writer_owned_stores"]
 toad = "toad.db"
 preferences = "toad.preferences"

src/toad/about.py

@@ -6,7 +6,7 @@
 import platform
 from string import Template
 
-from agent_comms.launch_root import LaunchRoot
+from pathlib import Path
 from toad.preferences import SavedSettings
 from toad.terminal_attention import TerminalAttention
 from toad import paths
@@ -78,7 +78,7 @@
 )
 
 
-def render() -> str:
+def render(root: Path) -> str:
     """Render about markdown.
 
     Returns:
@@ -95,7 +95,7 @@ def render() -> str:
         "COLORTERM": os.environ.get("COLORTERM", ""),
         "COMMS_AGENT_ARGS": shlex.join(InheritedArguments().configured().argv),
         "COMMS_AGENT_BIN": ConfiguredBinary().configured(),
-        "COMMS_ROOT": str(LaunchRoot.current().path),
+        "COMMS_ROOT": str(root),
         "CONFIG": config,
         "DATA_PATH": paths.get_data(),
         "LOG_PATH": paths.get_log(),

src/toad/acp/agent_controller.py

@@ -21,6 +21,7 @@
 )
 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.ui_model.delivery import DeliveryView
 from agent_comms.ui_model.observation import ReadDeliveryOverview
@@ -390,10 +391,6 @@ async def _cancel_prompt(self, authority: ClientSessionRequest):
     async def request_owner(self, method: str, **params: JsonValue):
         if self.coordination is None:
             raise ValueError("This action requires an agent-comms thread.")
-        from agent_comms.runtime import RuntimeProxy, socket_path
-
-        from toad.owner_preparation import OwnerRequestContext
-
         coordination = self.coordination
         authority = ClientSessionRequest(self.agent, self.agent.session_id)
         self.require_owner(coordination, authority)
@@ -403,8 +400,8 @@ def resolve():
             owner = comms.registry.require(coordination.thread.name)
             if owner.incarnation != coordination.thread:
                 raise ValueError("The owner incarnation changed while preparing the request.")
-            return RuntimeProxy(
-                OwnerRequestContext(comms), owner.name, socket_path(comms.root, owner.require_process().pid)
+            return RuntimeConnection(
+                comms, owner.incarnation, socket_path(comms.root, owner.require_process().pid)
             )
 
         proxy = await asyncio.to_thread(resolve)

src/toad/acp/agent_process.py

@@ -70,7 +70,7 @@ async def start(self, agent, processes: tuple[AgentProcess, ...]=()):
             agent.presentation.log_path.parent.mkdir, parents=True, exist_ok=True)
         env = os.environ.copy()
         cwd = str(agent.project_root_path.resolve())
-        selection = RouteSelection.capture(agent.controller.transcripts.access.launch_root.path)
+        selection = agent.controller.transcripts.access.selection
         if not agent.session.accepts_updates:
             return self
         for connection in dict.fromkeys((self, *processes)):

src/toad/acp/agent_session.py

@@ -18,7 +18,6 @@
 from agent_comms.errors import RouteChangedError
 from agent_comms.input_attempt import NotSentInput
 from agent_comms.owner_protocol import IncompatibleOwner
-from agent_comms.route_selection import RouteSelection
 from agent_comms.session_load import (
     EnsuringSessionLoadPlan,
     ExistingSessionLoadPlan,
@@ -362,7 +361,7 @@ async def reconnect(self, plan: SessionLoadPlan | None = None) -> None:
         try:
             try:
                 await asyncio.to_thread(
-                    RouteSelection.capture, self.agent.controller.transcripts.access.launch_root.path)
+                    lambda: self.agent.controller.transcripts.access.selection)
             except RouteChangedError as error:
                 self.startup_failed(f"Reconnect blocked before restart: {error}")
                 return

src/toad/acp/comms_updates.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from pathlib import Path
 from typing import TYPE_CHECKING
 
 from agent_comms.acp_extension import (
@@ -29,7 +28,6 @@
     TurnChangedUpdate,
 )
 from agent_comms.context_reading import UnmeasuredContext
-from agent_comms.errors import RouteChangedError
 from agent_comms.native_output import AssistantOutput, ToolOutput
 from agent_comms.thread_presentation import NoThreadPresentation, ThreadPresentation
 from agent_comms.transcripts import NativeCommittedPrefix
@@ -186,9 +184,7 @@ def _post_unidentified(self, update: InputStartedUpdate) -> None:
     @handles(CoordinationChangedUpdate)
     def coordination_changed(self, update: CoordinationChangedUpdate) -> None:
         agent = self.agent
-        root = agent.controller.transcripts.access.launch_root.path
-        if Path(update.wire_root).resolve() != root.resolve():
-            raise RouteChangedError("ACP owner reported a different coordination root")
+        agent.controller.transcripts.access.selected(update.wire_root)
         agent.coordination = update
         agent.project_root_path = update.worktree.path
         agent.context_measurement = ContextMeasurement.saved(update.context_usage)

src/toad/acp/transcript_reader.py

@@ -16,7 +16,6 @@
 from functools import partial
 
 from agent_comms.acp_extension import TranscriptDraftUpdate, TranscriptSnapshotUpdate
-from agent_comms.route_selection import RouteSelection
 from agent_comms.thread_presentation import ThreadPresentation
 from agent_comms.ui_model.observation import (
     AwaitedRead,
@@ -44,7 +43,7 @@ async def read(self, request: AwaitedRead):
 
     async def service(self, root):
         return await self.access.preparation.run_thread(
-            lambda: self.access.require(RouteSelection.capture(root)).service
+            lambda: self.access.require(self.access.selected(root)).service
         )
 
     async def thread_presentation(

src/toad/app.py

@@ -18,6 +18,7 @@
 from typing import TYPE_CHECKING, Callable, Never, TypeVar, cast
 
 from agent_comms.acp_extension import TranscriptChangedUpdate
+from agent_comms.launch_root import LaunchRoot
 from agent_comms.repo import Checkout
 from toad.core import input_events
 from rich.console import RenderableType
@@ -150,6 +151,7 @@ def __init__(
         self,
         agent_binding: AgentBinding,
         project_dir: Checkout,
+        launch_root: LaunchRoot,
         mode: str | None = None,
         agent_session_id: str | None = None,
     ) -> None:
@@ -158,6 +160,7 @@ def __init__(
         Args:
             agent_binding: The agent selected at startup.
             project_dir: Project directory.
+            launch_root: Coordination root selected by the launch owner.
             mode: Initial mode.
         """
         from toad.render_processes import RenderProcessPool
@@ -189,7 +192,7 @@ def __init__(
             db=DB(),
         )
         self._session_tracker = SessionTracker(self.tab_order)
-        self.coordination_access = CoordinationAccess(self)
+        self.coordination_access = CoordinationAccess(self, launch_root)
 
         from toad.file_kind import FileKind
         FileKind.prepare_terminal()

src/toad/cli.py

@@ -6,6 +6,7 @@
 import click
 
 from agent_comms.field_codec import FieldCodec
+from agent_comms.launch_root import LaunchRoot
 from agent_comms.repo import Checkout
 from toad.session_scope import AgentAttached, AgentBinding, NoAgent
 
@@ -75,10 +76,9 @@ def parse_args(self, ctx, args):
         if "--version" in args or "-v" in args:
             return super().parse_args(ctx, args)
         from agent_comms.active_route import ActiveRoute
-        from agent_comms.launch_root import LaunchRoot
 
-        # One selection, as in every Comms process: --comms-root if given, else the published route.
-        LaunchRoot.startup(args, ActiveRoute.published_launch)
+        # Select once at the process boundary; apps receive this captured value.
+        ctx.obj = LaunchRoot.startup(args, ActiveRoute.published_launch)
         # Check if first arg is a known subcommand
         if not args or args[0] not in self.commands:
             # If not a subcommand, prepend the default command name
@@ -123,7 +123,8 @@ def main(ctx, version):
 @main.command("run")
 @click.argument("project_dir", metavar="PATH", required=True)
 @click.option("-a", "--agent", metavar="AGENT", default="")
-def run(project_dir: str, agent: str = "1"):
+@click.pass_obj
+def run(launch_root: LaunchRoot, project_dir: str, agent: str = "1"):
     """Run an installed agent."""
 
     project = FieldCodec.decode(Checkout, project_dir).resolved(strict=True)
@@ -137,6 +138,7 @@ def run(project_dir: str, agent: str = "1"):
         mode=binding.startup_mode,
         agent_binding=binding,
         project_dir=project,
+        launch_root=launch_root,
     )
     run_terminal(app)
 
@@ -158,7 +160,9 @@ def run(project_dir: str, agent: str = "1"):
     default=None,
 )
 @click.option("-d", "--project-dir", "project_option", metavar="PATH", default=None)
+@click.pass_obj
 def acp(
+    launch_root: LaunchRoot,
     command: str,
     title: str | None,
     project_dir: str | None,
@@ -195,6 +199,7 @@ def acp(
     app = ToadApp(
         agent_binding=AgentAttached(agent_data),
         project_dir=FieldCodec.decode(Checkout, project_dir).resolved(strict=True),
+        launch_root=launch_root,
         agent_session_id=session_id,
     )
     run_terminal(app)
@@ -269,12 +274,13 @@ def serve(port: int, host: str, toad_args: tuple[str, ...]) -> None:
 
 
 @main.command("about")
-def about() -> None:
+@click.pass_obj
+def about(launch_root: LaunchRoot) -> None:
     """Show about information."""
 
     from toad import about
 
-    print(about.render())
+    print(about.render(launch_root.path))
 
 
 if __name__ == "__main__":

src/toad/comms_root.py

@@ -24,6 +24,7 @@
 from agent_comms.comms import Comms
 from agent_comms.display_references import PendingReferences, UnavailableReferences
 from agent_comms.errors import RouteChangedError
+from agent_comms.launch_root import LaunchRoot
 from agent_comms.launch_runtime import LaunchRuntime
 from metaclass_registry.mro_dispatch import MroDispatch, handles
 from agent_comms.presentation import MessageNotification, NoRevision, ReadState
@@ -329,11 +330,9 @@ def observe_commander(self, wanted: ObserveCommanderHud | StopCommanderHud):
 class CoordinationAccess(MroDispatch):
     """A validated Core route owns cached access, observations and UI write checks."""
 
-    def __init__(self, app: ToadApp) -> None:
-        from agent_comms.launch_root import LaunchRoot
-
+    def __init__(self, app: ToadApp, launch_root: LaunchRoot) -> None:
         self.app = app
-        self.launch_root = LaunchRoot.current()
+        self.launch_root = launch_root
         from toad.acp.transcript_reader import CoordinationTranscriptReader
 
         self.transcripts = CoordinationTranscriptReader(self)
@@ -361,18 +360,31 @@ def __init__(self, app: ToadApp) -> None:
         self.commander_pages: WeakSet[CommanderPage] = WeakSet()
         self.commander_interest: ObserveCommanderHud | StopCommanderHud = StopCommanderHud()
 
+    @property
+    def selection(self) -> RouteSelection:
+        """The current route resolved from this app's held launch root."""
+        return RouteSelection.of(self.launch_root)
+
+    def selected(self, source: str | Path) -> RouteSelection:
+        """Select this app's route and refuse a foreign source."""
+        return self.selection.require_source(source)
+
+    def current(self, source: str | Path) -> bool:
+        """Whether a source still belongs to this app's current route."""
+        return self.selection.root_is_current(source)
+
     def open_current(self) -> Observing | FailedObservation:
         """Capture the current route and open its service, or expose the opening error."""
         with self.opening:
             selected: RouteSelection | UnselectedRoute = UnselectedRoute()
             try:
-                selected = RouteSelection.capture(self.launch_root.path)
+                selected = self.selection
                 return self.require(selected)
             except (OSError, ValueError, RuntimeError) as error:
                 return self.observation.failed(error, selected)
 
     def confirm(self, selected: RouteSelection) -> None:
-        if RouteSelection.capture() != selected:
+        if self.selection != selected:
             raise RouteChangedError("Comms route changed before the operation")
 
     def require(self, selected: RouteSelection) -> Observing:
@@ -649,5 +661,5 @@ async def observe_route(self) -> None:
     def write(self, selected: RouteSelection, operation: Callable[..., T], *args: object, **kwargs: object) -> T:
         # This method runs in the same worker as the actual sink. No route lock: the
         # store refuses a write an install holds, inside its own write transaction.
-        RouteSelection.capture(selected.root)
+        self.confirm(selected)
         return operation(*args, **kwargs)

src/toad/core/context_inspection.py

@@ -15,7 +15,7 @@
 from agent_comms.field_codec import FieldCodec
 from agent_comms.importing import ImportedSessionMetadata
 from metaclass_registry.mro_dispatch import handles
-from agent_comms.native_turn_context import NativeContextData
+from agent_comms.pi_payloads import NativeContextData
 from agent_comms.pi_payloads import PiMessage
 from agent_comms.presentation import ReadState
 from agent_comms.registry_document import RegistryEntry
@@ -539,7 +539,7 @@ def read(
         page = comms.contexts.page(owner, comms.registry,
             cursor=cursor, previous=previous)
         annotations = WorkingMemoryAnnotations.for_context(
-            comms.root / "coordination.sqlite3",
+            comms.registry.store.coordinator_reader,
             tuple(resource.value for resource in page.resources), JevClassifier.version())
         imported = ImportedSessionMetadata.sources_for_owner(comms.registry, thread)
         inspection_type = ThreadInspectionSource().project_value(entry.status)
@@ -622,7 +622,7 @@ async def _request(self, action: str, **parameters: JsonValue | tuple[int, ...])
         if owner.incarnation != self.owner.incarnation:
             raise ValueError("Selected context thread incarnation changed")
         process = owner.require_process()
-        connection = RuntimeConnection(self.service, owner.name, socket_path(self.service.root, process.pid))
+        connection = RuntimeConnection(self.service, owner.incarnation, socket_path(self.service.root, process.pid))
         try:
             return await connection.request(action, **parameters)
         finally: