fix(simulation): stabilize PX4 readiness supervision

src/k1link/simulation/orchestrator.py

@@ -419,7 +419,7 @@ class SimulationApplicationService:
             return self._terminal_failure(
                 starting,
                 operation="start",
-                detail=type(exc).__name__,
+                detail=_exception_detail(exc),
                 observed_at_utc=observed_at_utc,
                 host_monotonic_ns=host_monotonic_ns,
                 sim_time_ns=0,
@@ -471,7 +471,7 @@ class SimulationApplicationService:
         return self._terminal_failure(
             run,
             operation=operation,
-            detail=type(exc).__name__,
+            detail=_exception_detail(exc),
             observed_at_utc=observed_at_utc,
             host_monotonic_ns=host_monotonic_ns,
             sim_time_ns=sim_time_ns,
@@ -537,6 +537,11 @@ class SimulationApplicationService:
         )
 
 
+def _exception_detail(exc: Exception) -> str:
+    message = " ".join(str(exc).split())
+    return type(exc).__name__ if not message else f"{type(exc).__name__}: {message}"
+
+
 def _idempotency_key(value: str) -> str:
     normalized = value.strip()
     if not 1 <= len(normalized) <= 160:

src/k1link/simulation/process_supervisor.py

@@ -30,6 +30,7 @@ class ProcessSpec:
     ready_log_patterns: tuple[str, ...] = ()
     health_timeout_seconds: float = 0.0
     health_poll_interval_seconds: float = 0.05
+    keep_stdin_open: bool = False
     interrupt_timeout_seconds: float = 2.0
     terminate_timeout_seconds: float = 1.0
 
@@ -225,7 +226,7 @@ class PosixProcessSupervisor:
                 spec.argv,
                 cwd=run_root,
                 env=environment,
-                stdin=subprocess.DEVNULL,
+                stdin=subprocess.PIPE if spec.keep_stdin_open else subprocess.DEVNULL,
                 stdout=stdout,
                 stderr=stderr,
                 start_new_session=True,
@@ -274,31 +275,40 @@ class PosixProcessSupervisor:
         stderr_path: Path,
     ) -> None:
         pending = {pattern: pattern.encode("utf-8") for pattern in spec.ready_log_patterns}
+        maximum_tail_bytes = max(len(pattern) for pattern in pending.values()) - 1
         deadline = time.monotonic() + spec.health_timeout_seconds
-        while pending:
-            return_code = process.poll()
-            if return_code is not None:
-                raise ProcessSupervisorError(
-                    f"provider {spec.process_id} exited before readiness with {return_code}; "
-                    f"missing markers: {', '.join(pending)}"
-                )
-            for path in (stdout_path, stderr_path):
-                content = path.read_bytes()
-                pending = {
-                    pattern: encoded
-                    for pattern, encoded in pending.items()
-                    if encoded not in content
-                }
-                if not pending:
-                    return
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                raise ProcessSupervisorError(
-                    f"provider {spec.process_id} did not reach readiness within "
-                    f"{spec.health_timeout_seconds:g} seconds; "
-                    f"missing markers: {', '.join(pending)}"
-                )
-            time.sleep(min(spec.health_poll_interval_seconds, remaining))
+        with (
+            stdout_path.open("rb", buffering=0) as stdout,
+            stderr_path.open("rb", buffering=0) as stderr,
+        ):
+            streams = (stdout, stderr)
+            tails = [b"", b""]
+            while pending:
+                return_code = process.poll()
+                if return_code is not None:
+                    raise ProcessSupervisorError(
+                        f"provider {spec.process_id} exited before readiness with {return_code}; "
+                        f"missing markers: {', '.join(pending)}"
+                    )
+                for index, stream in enumerate(streams):
+                    content = tails[index] + stream.read()
+                    if content:
+                        pending = {
+                            pattern: encoded
+                            for pattern, encoded in pending.items()
+                            if encoded not in content
+                        }
+                        tails[index] = content[-maximum_tail_bytes:] if maximum_tail_bytes else b""
+                    if not pending:
+                        return
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    raise ProcessSupervisorError(
+                        f"provider {spec.process_id} did not reach readiness within "
+                        f"{spec.health_timeout_seconds:g} seconds; "
+                        f"missing markers: {', '.join(pending)}"
+                    )
+                time.sleep(min(spec.health_poll_interval_seconds, remaining))
 
     def _stop_group(
         self,
@@ -314,6 +324,7 @@ class PosixProcessSupervisor:
         ):
             if process.poll() is not None and not _group_exists(pgid):
                 process.wait(timeout=0)
+                _close_stdin(process)
                 return True
             try:
                 if group_signal_supported:
@@ -321,6 +332,7 @@ class PosixProcessSupervisor:
                 else:
                     process.send_signal(sent_signal)
             except ProcessLookupError:
+                _close_stdin(process)
                 return True
             except PermissionError:
                 # Some macOS application sandboxes deny killpg for an owned
@@ -334,11 +346,15 @@ class PosixProcessSupervisor:
                 if process.poll() is not None:
                     process.wait(timeout=0)
                     if not group_signal_supported or not _group_exists(pgid):
+                        _close_stdin(process)
                         return True
                 time.sleep(0.01)
-        return process.poll() is not None and (
+        stopped = process.poll() is not None and (
             not group_signal_supported or not _group_exists(pgid)
         )
+        if stopped:
+            _close_stdin(process)
+        return stopped
 
     def _persist_registry(
         self,
@@ -368,3 +384,8 @@ def _group_exists(pgid: int) -> bool:
     except PermissionError:
         return True
     return True
+
+
+def _close_stdin(process: subprocess.Popen[bytes]) -> None:
+    if process.stdin is not None and not process.stdin.closed:
+        process.stdin.close()

src/k1link/simulation/stock_rover.py

@@ -171,6 +171,7 @@ def stock_rover_process_specs(
             ready_log_patterns=profile.px4_ready_log_patterns,
             health_timeout_seconds=120,
             health_poll_interval_seconds=0.25,
+            keep_stdin_open=True,
             interrupt_timeout_seconds=10,
             terminate_timeout_seconds=5,
         ),