Read process arguments and environment through Platform

src/agent_comms/active_route.py

@@ -18,7 +18,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, ClassVar
 
-from .child_process import Platform
+from .child_process import Platform, ProcessIdentity
 from .coordination_errors import CoordinationError
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
@@ -295,11 +295,6 @@ def detail(self) -> str:
         return "process exited"
 
 
-class EmptyCommandLockHolder(IdentifiedLockHolder):
-    def detail(self) -> str:
-        return "empty command"
-
-
 class UnidentifiedLockHolder(RouteLockHolder):
     def describe(self) -> str:
         return "kernel did not identify the owner"
@@ -324,11 +319,11 @@ def route_lock_holders(directory: int) -> tuple[RouteLockHolder, ...]:
                 holders.append(UnidentifiedLockHolder())
                 continue
             try:
-                command = Path(f'/proc/{pid}/cmdline').read_bytes().replace(b'\0', b' ').decode('utf-8', 'replace').strip()
-            except FileNotFoundError:
+                command = " ".join(ProcessIdentity.capture(pid).arguments())
+            except ProcessLookupError:
                 holders.append(ExitedLockHolder(pid))
             else:
-                holders.append(RunningLockHolder(pid, command) if command else EmptyCommandLockHolder(pid))
+                holders.append(RunningLockHolder(pid, command))
     return tuple(holders)
 
 

src/agent_comms/child_process.py

@@ -247,6 +247,12 @@ def __eq__(self, other: object) -> bool:
     def alive(self) -> bool:
         return Platform.current().matches(self)
 
+    def arguments(self) -> tuple[str, ...]:
+        return Platform.current().arguments(self)
+
+    def environment(self) -> dict[str, str]:
+        return Platform.current().environment(self)
+
     def require_alive(self) -> ProcessIdentity:
         """Validate this recorded PID and start time against the live process."""
         return Platform.current().require(self)
@@ -410,6 +416,12 @@ def force_group(self, identity: ProcessIdentity) -> None:
     @abstractmethod
     def identity(self, pid: int) -> ProcessIdentity: ...
 
+    @abstractmethod
+    def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: ...
+
+    @abstractmethod
+    def environment(self, process: ProcessIdentity) -> dict[str, str]: ...
+
     def parent_identity(self, pid: int) -> ProcessIdentity:
         raise NotImplementedError('This platform does not expose exact parent identities')
 
@@ -550,6 +562,23 @@ def _stat(pid: int) -> LinuxProcessRecord:
     def identity(self, pid: int) -> ProcessIdentity:
         return self._stat(pid).identity
 
+    def arguments(self, process: ProcessIdentity) -> tuple[str, ...]:
+        try:
+            raw = Path(f"/proc/{process.pid}/cmdline").read_bytes()
+        except FileNotFoundError:
+            raise ProcessLookupError(process.pid) from None
+        if not raw:
+            raise ProcessLookupError(process.pid)
+        return tuple(os.fsdecode(word) for word in raw.split(b"\0")[:-1])
+
+    def environment(self, process: ProcessIdentity) -> dict[str, str]:
+        try:
+            raw = Path(f"/proc/{process.pid}/environ").read_bytes()
+        except FileNotFoundError:
+            raise ProcessLookupError(process.pid) from None
+        values = (item.split(b"=", 1) for item in raw.split(b"\0") if b"=" in item)
+        return {os.fsdecode(key): os.fsdecode(value) for key, value in values}
+
     def parent_identity(self, pid: int) -> ProcessIdentity:
         return self.identity(self._stat(pid).parent)
 
@@ -683,6 +712,12 @@ def identity(self, pid: int) -> ProcessIdentity:
             pid, info.start_seconds * 1_000_000 + info.start_microseconds, info.pgid,
         )
 
+    def arguments(self, process: ProcessIdentity) -> tuple[str, ...]:
+        raise NotImplementedError(f"{type(self).__name__} does not read process arguments")
+
+    def environment(self, process: ProcessIdentity) -> dict[str, str]:
+        raise NotImplementedError(f"{type(self).__name__} does not read process environment")
+
     def members_of_group(self, pgid: int) -> tuple[ProcessIdentity, ...]:
         # proc_listpids returns bytes. Grow until one query fits; never mistake
         # a full buffer for a complete process-group observation.
@@ -875,6 +910,12 @@ def identity(self, pid: int) -> ProcessIdentity:
         with self.process_handle(pid) as handle:
             return self._identity(pid, handle)
 
+    def arguments(self, process: ProcessIdentity) -> tuple[str, ...]:
+        raise NotImplementedError(f"{type(self).__name__} does not read process arguments")
+
+    def environment(self, process: ProcessIdentity) -> dict[str, str]:
+        raise NotImplementedError(f"{type(self).__name__} does not read process environment")
+
     def wait_any_exited(
         self, identities: tuple[ProcessIdentity, ...], timeout: float,
     ) -> None:

src/agent_comms/host_services/__init__.py

@@ -19,6 +19,7 @@
 
 from aiohttp.web import TCPSite
 
+from ..child_process import ProcessIdentity
 from ..declared_family import DeclaredFamily
 from ..field_codec import PathText
 from ..launch_runtime import LaunchRuntime
@@ -196,15 +197,14 @@ def running(cls) -> tuple[Self, ...]:
         return () if pid == 0 else (cls.from_process(pid),)
 
     @classmethod
-    def process_arguments(cls, pid: int) -> tuple[str, ...]:
-        arguments = tuple(os.fsdecode(word) for word in
-                          Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")[:-1])
-        if not arguments or not arguments[0]:
-            raise ServiceNotServingError(f"{cls.unit} has no running command")
-        return arguments
+    def running_arguments(cls, pid: int) -> tuple[str, ...]:
+        try:
+            return ProcessIdentity.capture(pid).arguments()
+        except ProcessLookupError:
+            raise ServiceNotServingError(f"{cls.unit} has no running command") from None
 
     def runtime(self) -> Path:
-        return Path(self.process_arguments(self.pid())[0])
+        return Path(self.running_arguments(self.pid())[0])
 
     def restart(self) -> None:
         subprocess.run(("systemctl", "--user", "restart", self.unit), check=True)

src/agent_comms/host_services/connector.py

@@ -81,7 +81,7 @@ class SystemdConnectorService(ConnectorService, SystemdUserService):
     @classmethod
     def from_process(cls, pid: int) -> Self:
         # Our installed console script's argv is interpreter, script, verb, options.
-        endpoint = ServeVerb.parse(cls.process_arguments(pid)[2:])
+        endpoint = ServeVerb.parse(cls.running_arguments(pid)[2:])
         return cls(launch=LaunchRoot.from_path(endpoint.root), port=endpoint.port)
 
     def exec_arguments(self, issuer: str) -> tuple[str, ...]:

src/agent_comms/owner_launch.py

@@ -2,9 +2,7 @@
 
 from __future__ import annotations
 
-import os
 from dataclasses import dataclass, field
-from pathlib import Path
 from typing import TYPE_CHECKING
 
 from .child_process import Platform, ProcessIdentity
@@ -37,13 +35,10 @@ def capture(
         process = selection.process
         platform = Platform.current()
         platform.require(process)
-        command = Path(f"/proc/{process.pid}/cmdline").read_bytes().split(b"\0", 1)[0]
-        observed_interpreter = os.fsdecode(command)
+        observed_interpreter = process.arguments()[0]
         if observed_interpreter != selection.interpreter:
             raise RelationViolationError("Selected source owner interpreter changed")
-        raw = Path(f"/proc/{process.pid}/environ").read_bytes()
-        values = dict(item.split(b"=", 1) for item in raw.split(b"\0") if b"=" in item)
-        environment = {os.fsdecode(key): os.fsdecode(value) for key, value in values.items()}
+        environment = process.environment()
         platform.require(process)
         route = launch_runtime.ThreadLaunch.current(environment).name
         if not environment.get(NativeArguments.binary_key):

src/agent_comms/owner_lifecycle.py

@@ -103,8 +103,7 @@ def capture(cls, snapshot: RegistrySnapshot, name: str):
         process = owner.require_process()
         platform = Platform.current()
         platform.require(process)
-        command = Path(f"/proc/{process.pid}/cmdline").read_bytes().split(b"\0", 1)[0]
-        interpreter = os.fsdecode(command)
+        interpreter = process.arguments()[0]
         if not interpreter:
             raise RelationViolationError("Selected source owner interpreter is unavailable")
         platform.require(process)