Resolve routes from the held launch root

src/agent_comms/acp.py

@@ -574,7 +574,7 @@ class CommsClient(CommsAgent):
 
 
 def main() -> int:
-    LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
+    root = LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
     if "--login" in sys.argv[1:]:
         from .login import run_login
 
@@ -592,7 +592,7 @@ def main() -> int:
     # The stdio ACP client only attaches to a separately owned worker. A
     # private launch is checked before it can request an owner; the worker
     # independently repeats the same preflight.
-    comms = wire()
+    comms = wire(root.path)
     comms.owners.require_launch_authority()
 
     async def run() -> None:

src/agent_comms/cli.py

@@ -49,11 +49,11 @@ def execute(self, arguments: Sequence[str], output: TextIO) -> int:
         if unknown and (len(unknown) != 2 or unknown[0] != LaunchRoot.argument):
             raise ValueError(f"unrecognized arguments: {' '.join(unknown)}")
         declaration = args.command_declaration
-        LaunchRoot.startup(
+        launch_root = LaunchRoot.startup(
             arguments,
             lambda: args.root.resolve(declaration.published_launch),
         )
-        route = resolve_comms_route(args.root.resolve(LaunchRoot.current).path)
+        route = resolve_comms_route(args.root.resolve(lambda: launch_root).path)
         with LaunchRoot.from_path(route.root).selected():
             command = declaration.from_arguments(args)
             return command.execute(route, output)

src/agent_comms/commander/cli.py

@@ -37,9 +37,9 @@ def main() -> None:
 
     try:
         # Every comms process selects its root once at startup, as the comms CLI does.
-        LaunchRoot.startup(argv, lambda: (LaunchRoot.from_path(options.root) if options.root
-                                          else ActiveRoute.published_launch()))
-        context = Commander.read_root(resolve_comms_route(options.root).root)
+        root = LaunchRoot.startup(argv, lambda: (LaunchRoot.from_path(options.root) if options.root
+                                                 else ActiveRoute.published_launch()))
+        context = Commander.read_root(resolve_comms_route(root.path).root)
         tools[argv[0]].main(argv[1:], context)
     except CLI_FAILURES as refused:
         # The comms CLI's one failure set: a refusal the command raises prints, never a traceback.

src/agent_comms/launch_root.py

@@ -83,16 +83,18 @@ def arguments(self) -> tuple[str, str]:
         return self.argument, str(self.path)
 
     @classmethod
-    def startup(cls, arguments: list[str], published: Callable[[], LaunchRoot]) -> None:
+    def startup(cls, arguments: list[str], published: Callable[[], LaunchRoot]) -> LaunchRoot:
         if cls.argument in arguments:
             index = arguments.index(cls.argument)
             if index + 1 == len(arguments):
                 raise ValueError(f"{cls.argument} requires a root")
             record = FieldCodec.decode(cls, {"path": arguments[index + 1]})
-            cls.from_path(record.path).install()
+            root = cls.from_path(record.path)
             del arguments[index : index + 2]
         else:
-            published().install()
+            root = published()
+        root.install()
+        return root
 
     @classmethod
     def environment(cls, source: Mapping[str, str]) -> dict[str, str]:

src/agent_comms/native_pi.py

@@ -147,8 +147,8 @@ def diagnostic_evidence(self) -> dict[str, object]:
 
 def main() -> int:
     """Run the active route's pinned Pi for ordinary ACP owner sessions."""
-    LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
-    package = resolve_comms_route().launch().require_private().validate()
+    root = LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
+    package = resolve_comms_route(root.path).launch().require_private().validate()
     environment = dict(os.environ)
     argv, environment = NativePiRpcLaunch.bootstrap(
         package, tuple(sys.argv[1:]), Path.cwd(), environment,

src/agent_comms/route_selection.py

@@ -8,7 +8,6 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -20,40 +19,27 @@
 from .recorded_bus import RecordedBusIdentity
 
 
-def current_root() -> Path:
-    """Resolve the explicit override or the validated default Comms route."""
-    return resolve_comms_route().observe_root()
-
-
-def root_is_current(root: str | Path) -> bool:
-    """Fail closed if a client's root is no longer the current route."""
-    return current_root() == Path(root).expanduser().resolve()
-
-
-def child_root(env: Mapping[str, str], cwd: str | Path) -> Path:
-    """The explicit process launch selects its child; HOME never supplies a route."""
-    return LaunchRoot.current().path
-
-
 @dataclass(frozen=True)
 class RouteSelection:
     route: CommsRoute
     root: Path
     bus_identity: RecordedBusIdentity
 
     @classmethod
-    def capture(cls, source: str | Path | None = None) -> RouteSelection:
-        route = resolve_comms_route()
-        root = route.observe_root()
-        selection = cls(route, root, RecordedBusIdentity.capture(Bus.path_in(root)))
-        if source is not None and Path(source).expanduser().resolve() != selection.root:
-            raise errors.RouteChangedError(errors.RouteChangedError.reader_sentence)
-        return selection
-
-    @classmethod
-    def for_child(cls, env: Mapping[str, str], cwd: str | Path) -> RouteSelection:
-        """Capture the explicit process selection before launching its child."""
-        root = child_root(env, cwd)
+    def of(cls, root: LaunchRoot) -> RouteSelection:
+        """Resolve the root held by this client, independent of the process default."""
+        route = resolve_comms_route(root.path)
+        selected_root = route.observe_root()
         return cls(
-            resolve_comms_route(root), root, RecordedBusIdentity.capture(Bus.path_in(root))
+            route, selected_root, RecordedBusIdentity.capture(Bus.path_in(selected_root))
         )
+
+    def root_is_current(self, source: str | Path) -> bool:
+        """Whether a source belongs to this selection, not the process default."""
+        return Path(source).expanduser().resolve() == self.root
+
+    def require_source(self, source: str | Path) -> RouteSelection:
+        """Refuse a foreign source before using this client's selection."""
+        if not self.root_is_current(source):
+            raise errors.RouteChangedError(errors.RouteChangedError.reader_sentence)
+        return self

src/agent_comms/worker.py

@@ -17,11 +17,11 @@
 
 
 async def run() -> None:
-    LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
+    root = LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
     parser = argparse.ArgumentParser(prog="agent-comms-agent", description=__doc__)
     parser.add_argument("--listener-fd", type=int, required=True)
     options = parser.parse_args()
-    await run_host(wire(), options.listener_fd)
+    await run_host(wire(root.path), options.listener_fd)
 
 
 async def run_host(comms: Comms, listener_fd: int) -> None:

tests/selected_write_owner_child.py

@@ -26,8 +26,8 @@ async def wait_for(path: Path):
 
 
 async def main(base: Path) -> None:
-    LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
-    comms = wire()
+    root = LaunchRoot.startup(sys.argv, ActiveRoute.published_launch)
+    comms = wire(root.path)
     await wait_for(base / "registered")
     stopped = asyncio.Event()
     loop = asyncio.get_running_loop()