Let go of a process that will not die

src/bazaar_compute_node/contrib/claude/process.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import logging
 from collections import deque
 from collections.abc import Mapping
 from dataclasses import dataclass
@@ -22,6 +23,8 @@
 MAX_JSONL_BYTES = 1024 * 1024
 _EXIT_PIPE_DRAIN_SECONDS = 1
 _CLOSED = object()
+_REAP_REPORT_SECONDS = 30.0
+_LOGGER = logging.getLogger("bazaar_compute_node.runtime.claude.process")
 
 
 class ProcessState(StrEnum):
@@ -120,6 +123,8 @@ def __init__(self, spec: ProcessSpec, *, stderr_tail_limit: int = 64) -> None:
         self._write_lock = asyncio.Lock()
         self._lifecycle_lock = asyncio.Lock()
         self._exit_event = asyncio.Event()
+        self._abandoned: set[asyncio.Task[None]] = set()
+        self._started = False
 
     @property
     def state(self) -> ProcessState:
@@ -147,8 +152,9 @@ def result_error_tail(self) -> tuple[str, ...]:
 
     async def start(self, *, timeout: float) -> None:
         async with self._lifecycle_lock:
-            if self.is_running:
-                return
+            if self._started:
+                raise RuntimeError("a stopped supervisor cannot start again")
+            self._started = True
             self._state = ProcessState.RUNNING
             try:
                 async with asyncio.timeout(timeout):
@@ -227,6 +233,7 @@ async def stop(self, *, timeout: float) -> None:
             now = asyncio.get_running_loop().time()
             graceful_deadline = now + timeout * 0.6
             terminate_deadline = now + timeout * 0.9
+            kill_deadline = now + timeout
             if process.stdin is not None:
                 process.stdin.close()
             try:
@@ -237,12 +244,74 @@ async def stop(self, *, timeout: float) -> None:
                     await self._wait_until(process, terminate_deadline)
                 except TimeoutError:
                     process.kill()
-                    await process.wait()
+                    try:
+                        await self._wait_until(process, kill_deadline)
+                    except TimeoutError:
+                        # a process that survived a kill will not be waited out,
+                        # so let go of it here instead of holding the lifecycle
+                        # lock
+                        self._abandon(process)
+                        self._state = ProcessState.STOPPED
+                        self._exit_event.set()
+                        return
             self._returncode = process.returncode
             await self._join_tasks(cancel=True)
             self._state = ProcessState.STOPPED
             self._exit_event.set()
 
+    def _abandon(self, process: asyncio.subprocess.Process) -> None:
+        """Give up on a process without losing track of it.
+
+        Closing this side of the pipes is what unblocks everything: the readers
+        stop waiting for an EOF that a grandchild holding the write end will
+        never deliver, and the exit waiters can complete once the pipes count as
+        disconnected. Whoever still holds that write end is free to keep it.
+        """
+
+        transport = getattr(process, "_transport", None)
+        if transport is not None:
+            for fd in (1, 2):
+                pipe = transport.get_pipe_transport(fd)
+                if pipe is not None:
+                    pipe.close()
+        # the watcher normally closes the queue, and it is about to be
+        # cancelled, so release whoever is already waiting in receive()
+        self._incoming.put_nowait(_CLOSED)
+        tasks = tuple(
+            task
+            for task in (self._stdout_task, self._stderr_task, self._watch_task)
+            if task is not None and task is not asyncio.current_task()
+        )
+        for task in tasks:
+            task.cancel()
+        self._stdout_task = None
+        self._stderr_task = None
+        self._watch_task = None
+        self._process = None
+        reaper = asyncio.create_task(
+            self._reap(process, tasks),
+            name=f"bcn-claude-reap-{process.pid}",
+        )
+        self._abandoned.add(reaper)
+        reaper.add_done_callback(self._abandoned.discard)
+
+    async def _reap(
+        self,
+        process: asyncio.subprocess.Process,
+        tasks: tuple[asyncio.Task[None], ...],
+    ) -> None:
+        """Outlive the caller waiting for a process nobody else is waiting for."""
+
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        try:
+            async with asyncio.timeout(_REAP_REPORT_SECONDS):
+                returncode = await process.wait()
+        except TimeoutError:
+            _LOGGER.warning("process %s outlived its kill; still waiting", process.pid)
+            returncode = await process.wait()
+        self._returncode = returncode
+
     async def _read_stdout(self, process: asyncio.subprocess.Process) -> None:
         stdout = process.stdout
         if stdout is None:

src/bazaar_compute_node/contrib/codex/process.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import logging
 import os
 import signal
 from collections import deque
@@ -60,6 +61,8 @@ def command(self) -> tuple[str, ...]:
 
 
 _QUEUE_CLOSED = object()
+_REAP_REPORT_SECONDS = 30.0
+_LOGGER = logging.getLogger("bazaar_compute_node.runtime.codex.process")
 
 
 class JsonlProcessSupervisor:
@@ -93,6 +96,8 @@ def __init__(
         self._pending: dict[JsonlRequestId, asyncio.Future[JsonlMessage]] = {}
         self._next_request_id = 0
         self._closed_message_sent = False
+        self._abandoned: set[asyncio.Task[None]] = set()
+        self._started = False
 
     @property
     def state(self) -> JsonlProcessState:
@@ -123,10 +128,9 @@ def stderr_tail(self) -> tuple[str, ...]:
     async def start(self, *, timeout: float) -> None:
         _validate_timeout(timeout)
         async with self._lifecycle_lock:
-            if self.is_running:
-                return
-            await self._join_tasks()
-            self._reset_runtime_state()
+            if self._started:
+                raise RuntimeError("a stopped supervisor cannot start again")
+            self._started = True
             self._state = JsonlProcessState.STARTING
             try:
                 async with asyncio.timeout(timeout):
@@ -171,19 +175,42 @@ async def stop(self, *, timeout: float) -> None:
                 self._send_closed_message()
                 return
             self._state = JsonlProcessState.STOPPING
-            deadline = asyncio.get_running_loop().time() + timeout
+            now = asyncio.get_running_loop().time()
+            graceful_deadline = now + timeout * 0.6
+            terminate_deadline = now + timeout * 0.8
+            kill_deadline = now + timeout * 0.9
+            deadline = now + timeout
             if process.stdin is not None:
                 process.stdin.close()
             try:
-                await self._wait_for_process(process, deadline)
+                await self._wait_for_process(process, graceful_deadline)
             except TimeoutError:
                 _terminate_process(process)
                 try:
-                    await self._wait_for_process(process, deadline)
+                    await self._wait_for_process(process, terminate_deadline)
                 except TimeoutError:
                     _kill_process(process)
-                    await process.wait()
-            await self._join_tasks()
+                    try:
+                        await self._wait_for_process(process, kill_deadline)
+                    except TimeoutError:
+                        # a process that survived SIGKILL will not be waited out,
+                        # so let go of it here instead of holding the lifecycle
+                        # lock
+                        self._abandon(process)
+                        self._state = JsonlProcessState.STOPPED
+                        self._send_closed_message()
+                        return
+            # an exited process is no guarantee of EOF: a background command it
+            # left behind can hold the pipes open, and then the readers never
+            # finish on their own either
+            remaining = deadline - asyncio.get_running_loop().time()
+            try:
+                if remaining <= 0:
+                    raise TimeoutError
+                async with asyncio.timeout(remaining):
+                    await self._join_tasks()
+            except TimeoutError:
+                self._abandon(process)
             self._state = JsonlProcessState.STOPPED
             self._send_closed_message()
 
@@ -493,6 +520,70 @@ async def _wait_for_process(
         async with asyncio.timeout(remaining):
             await asyncio.shield(process.wait())
 
+    def _abandon(self, process: asyncio.subprocess.Process) -> None:
+        """Give up on a process without losing track of it.
+
+        Closing this side of the pipes is what unblocks everything: the readers
+        stop waiting for an EOF that a grandchild holding the write end will
+        never deliver, and the exit waiters can complete once the pipes count as
+        disconnected. Whoever still holds that write end is free to keep it.
+        """
+
+        transport = getattr(process, "_transport", None)
+        if transport is not None:
+            for fd in (1, 2):
+                pipe = transport.get_pipe_transport(fd)
+                if pipe is not None:
+                    pipe.close()
+        # the watcher normally publishes the exit, and it is about to be
+        # cancelled, so release whoever is already waiting on it
+        self._exit_event.set()
+        pending = tuple(self._pending.values())
+        self._pending.clear()
+        # the process may already have exited here, and then its status is the
+        # honest answer; the reaper publishes the rest once it arrives
+        exited = JsonlProcessExited(
+            returncode=process.returncode,
+            stderr_tail=self.stderr_tail,
+        )
+        for future in pending:
+            if not future.done():
+                future.set_exception(exited)
+        tasks = tuple(
+            task
+            for task in (self._stdout_task, self._stderr_task, self._watch_task)
+            if task is not None and task is not asyncio.current_task()
+        )
+        for task in tasks:
+            task.cancel()
+        self._stdout_task = None
+        self._stderr_task = None
+        self._watch_task = None
+        self._process = None
+        reaper = asyncio.create_task(
+            self._reap(process, tasks),
+            name=f"bcn-codex-reap-{process.pid}",
+        )
+        self._abandoned.add(reaper)
+        reaper.add_done_callback(self._abandoned.discard)
+
+    async def _reap(
+        self,
+        process: asyncio.subprocess.Process,
+        tasks: tuple[asyncio.Task[None], ...],
+    ) -> None:
+        """Outlive the caller waiting for a process nobody else is waiting for."""
+
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        try:
+            async with asyncio.timeout(_REAP_REPORT_SECONDS):
+                returncode = await process.wait()
+        except TimeoutError:
+            _LOGGER.warning("process %s outlived its kill; still waiting", process.pid)
+            returncode = await process.wait()
+        self._returncode = returncode
+
     async def _join_tasks(self) -> None:
         tasks = tuple(
             task
@@ -505,20 +596,6 @@ async def _join_tasks(self) -> None:
         self._stderr_task = None
         self._watch_task = None
 
-    def _reset_runtime_state(self) -> None:
-        while True:
-            try:
-                self._incoming.get_nowait()
-            except asyncio.QueueEmpty:
-                break
-        self._stderr_tail.clear()
-        self._returncode = None
-        self._fatal_error = None
-        self._next_request_id = 0
-        self._closed_message_sent = False
-        self._exit_event.clear()
-        self._pending.clear()
-
     def _send_closed_message(self) -> None:
         if self._closed_message_sent:
             return

tests/app/test_attachments.py

@@ -60,7 +60,10 @@ async def no_references() -> set[str]:
 
     def recording_open(path: Path, flags: int, mode: int) -> int:
         opened_flags.append(flags)
-        return real_open(path, flags & ~binary_flag, mode)
+        # the flag is real on Windows, where dropping it would translate
+        # newlines and corrupt the bytes this test is checking
+        native = flags if os.name == "nt" else flags & ~binary_flag
+        return real_open(path, native, mode)
 
     monkeypatch.setattr(attachment_module.os, "O_BINARY", binary_flag, raising=False)
     monkeypatch.setattr(attachment_module.os, "open", recording_open)

tests/app/test_daemon_process.py

@@ -18,9 +18,17 @@
 from bazaar_compute_node.core.paths import resolve_data_dir
 
 
-async def wait_for_runtime_endpoint(endpoint_path: Path) -> str:
+async def wait_for_runtime_endpoint(
+    endpoint_path: Path,
+    process: subprocess.Popen[str] | None = None,
+    *,
+    timeout: float = 30,
+) -> str:
+    # a node that is merely slow to start looks the same as one that never
+    # will, so wait on the clock rather than on a fixed number of attempts
     endpoint = local_endpoint_for_path(endpoint_path)
-    for _ in range(200):
+    deadline = asyncio.get_running_loop().time() + timeout
+    while asyncio.get_running_loop().time() < deadline:
         response: Mapping[str, object] | None = None
         try:
             response = await LocalCommandClient.request(
@@ -32,8 +40,16 @@ async def wait_for_runtime_endpoint(endpoint_path: Path) -> str:
             response = None
         if response is not None and response.get("ok") is True:
             return endpoint
+        # a node that already died will never publish, so say so now rather
+        # than sitting out the whole allowance
+        if process is not None and process.poll() is not None:
+            break
         await asyncio.sleep(0.01)
-    raise AssertionError("test node did not publish its local endpoint")
+    detail = ""
+    if process is not None and process.poll() is not None:
+        stderr = process.stderr.read() if process.stderr is not None else ""
+        detail = f" (exited {process.returncode}): {stderr.strip()}"
+    raise AssertionError(f"test node did not publish its local endpoint{detail}")
 
 
 async def request_with_retry(
@@ -139,7 +155,7 @@ async def test_real_process_reports_health_and_keeps_agent_configuration(
 ) -> None:
     process, endpoint_path, data_dir, _ = start_test_process(tmp_path)
     try:
-        endpoint = await wait_for_runtime_endpoint(endpoint_path)
+        endpoint = await wait_for_runtime_endpoint(endpoint_path, process)
         assert endpoint.startswith("pipe://" if os.name == "nt" else "unix://")
         health = await wait_for_health(endpoint)
         assert health["started"] is True
@@ -180,14 +196,14 @@ async def test_foreground_process_restarts_with_persisted_configuration(
 ) -> None:
     process, endpoint_path, data_dir, _ = start_test_process(tmp_path)
     try:
-        endpoint = await wait_for_runtime_endpoint(endpoint_path)
+        endpoint = await wait_for_runtime_endpoint(endpoint_path, process)
         await wait_for_health(endpoint)
         await stop_test_process(endpoint_path)
         await asyncio.to_thread(process.wait, 5)
         assert process.returncode == 0
 
         process, endpoint_path, data_dir, _ = start_test_process(tmp_path)
-        response_endpoint = await wait_for_runtime_endpoint(endpoint_path)
+        response_endpoint = await wait_for_runtime_endpoint(endpoint_path, process)
         assert response_endpoint == endpoint
         health = await wait_for_health(response_endpoint)
         assert health["ready"] is True

tests/app/test_system_service.py

@@ -98,7 +98,7 @@ def test_native_command_uses_system_encoding_without_decode_failures() -> None:
         text=True,
         encoding=locale.getencoding(),
         errors="replace",
-        creationflags=0,
+        creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
     )