diff --git a/src/bazaar_compute_node/contrib/claude/process.py b/src/bazaar_compute_node/contrib/claude/process.py index 67fd3d52..4c9488a8 100644 --- a/src/bazaar_compute_node/contrib/claude/process.py +++ b/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: diff --git a/src/bazaar_compute_node/contrib/codex/process.py b/src/bazaar_compute_node/contrib/codex/process.py index 9f747475..92e6d355 100644 --- a/src/bazaar_compute_node/contrib/codex/process.py +++ b/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 diff --git a/tests/app/test_attachments.py b/tests/app/test_attachments.py index 72e7e307..0be3e81d 100644 --- a/tests/app/test_attachments.py +++ b/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) diff --git a/tests/app/test_daemon_process.py b/tests/app/test_daemon_process.py index 6ccd0c33..21206f9c 100644 --- a/tests/app/test_daemon_process.py +++ b/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 diff --git a/tests/app/test_system_service.py b/tests/app/test_system_service.py index 2245ee0f..e173a31c 100644 --- a/tests/app/test_system_service.py +++ b/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, ) diff --git a/tests/contrib/test_claude.py b/tests/contrib/test_claude.py index e37b21eb..7f4bd962 100644 --- a/tests/contrib/test_claude.py +++ b/tests/contrib/test_claude.py @@ -582,6 +582,26 @@ async def test_claude_parent_exit_is_not_blocked_by_inherited_pipes( pass +@pytest.mark.skipif(os.name == "nt", reason="requires POSIX process signals") +@pytest.mark.asyncio +async def test_claude_stop_reports_the_exit_of_a_process_it_had_to_kill( + tmp_path: Path, +) -> None: + # the last slice of the stop budget is reserved to watch the kill land, so a + # process that merely ignores SIGTERM still answers with a real exit + script = ( + "import signal,time;signal.signal(signal.SIGTERM,signal.SIG_IGN);time.sleep(60)" + ) + supervisor = ProcessSupervisor( + ProcessSpec(sys.executable, ("-c", script), tmp_path, os.environ) + ) + await supervisor.start(timeout=10) + await supervisor.stop(timeout=1) + + assert supervisor.returncode is not None + assert await supervisor.wait(timeout=1) is not None + + def test_claude_runtime_factory_preserves_runtime_options() -> None: async def run_command( command: str, arguments: Sequence[str], cwd: str | None diff --git a/tests/contrib/test_codex.py b/tests/contrib/test_codex.py index 34b15441..3af3c8db 100644 --- a/tests/contrib/test_codex.py +++ b/tests/contrib/test_codex.py @@ -3,7 +3,9 @@ import asyncio import os import shutil +import signal from collections.abc import Callable, Sequence +from contextlib import suppress from dataclasses import replace from pathlib import Path from time import time_ns @@ -126,6 +128,16 @@ def load_agent( ) +def terminate_recorded_process(pid_file: Path) -> None: + """Kill a grandchild the test spawned, so runs do not leave sleepers behind.""" + + if not pid_file.exists(): + return + for line in pid_file.read_text().split(): + with suppress(OSError): + os.kill(int(line), getattr(signal, "SIGKILL", signal.SIGTERM)) + + def python_process(script: str, *, cwd: Path | None = None) -> JsonlProcessSpec: import sys @@ -963,6 +975,115 @@ async def run_command(*_: object) -> None: assert await runtime.has_background_job(session, timeout=3) is running +@pytest.mark.asyncio +async def test_jsonl_supervisor_stop_outlives_an_exited_process_holding_pipes( + tmp_path: Path, +) -> None: + # the process exits on its own, so the kill path never runs, but the + # grandchild it left behind still holds the pipes open + pid_file = tmp_path / "grandchild.pid" + supervisor = JsonlProcessSupervisor( + python_process( + f""" +import pathlib, subprocess, sys + +child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=sys.stdout, + stderr=sys.stderr, +) +with pathlib.Path({str(pid_file)!r}).open("a") as handle: + handle.write(f"{{child.pid}}\\n") +""", + cwd=tmp_path, + ) + ) + try: + await supervisor.start(timeout=5) + started = asyncio.get_running_loop().time() + await supervisor.stop(timeout=1) + elapsed = asyncio.get_running_loop().time() - started + + assert supervisor.state is JsonlProcessState.STOPPED + assert elapsed < 10 + finally: + terminate_recorded_process(pid_file) + + +@pytest.mark.skipif(os.name == "nt", reason="requires POSIX signal semantics") +@pytest.mark.asyncio +async def test_jsonl_supervisor_stop_outlives_a_held_pipe(tmp_path: Path) -> None: + # a grandchild that inherited stdout keeps the pipe open after its parent is + # killed, so stop() has to let go instead of waiting for an EOF that is not + # coming + pid_file = tmp_path / "grandchild.pid" + supervisor = JsonlProcessSupervisor( + python_process( + f""" +import pathlib, signal, subprocess, sys, time + +child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=sys.stdout, + stderr=sys.stderr, +) +with pathlib.Path({str(pid_file)!r}).open("a") as handle: + handle.write(f"{{child.pid}}\\n") +signal.signal(signal.SIGTERM, signal.SIG_IGN) +time.sleep(60) +""", + cwd=tmp_path, + ) + ) + try: + await supervisor.start(timeout=5) + waiting = asyncio.create_task(supervisor.wait()) + await asyncio.sleep(0) + started = asyncio.get_running_loop().time() + await supervisor.stop(timeout=1) + elapsed = asyncio.get_running_loop().time() - started + + assert supervisor.state is JsonlProcessState.STOPPED + assert elapsed < 10 + assert not supervisor.is_running + # a caller that was already waiting has to be released too + async with asyncio.timeout(10): + await waiting + + # the reaper still publishes the exit once the process is finally gone + async with asyncio.timeout(10): + while supervisor.returncode is None: + await asyncio.sleep(0.05) + + finally: + terminate_recorded_process(pid_file) + + +@pytest.mark.skipif(os.name == "nt", reason="requires POSIX signal semantics") +@pytest.mark.asyncio +async def test_jsonl_supervisor_stop_reports_the_exit_of_a_process_it_had_to_kill( + tmp_path: Path, +) -> None: + # the stop budget keeps a slice back to watch the kill land, so a process + # that only ignores SIGTERM still answers with a real exit + supervisor = JsonlProcessSupervisor( + python_process( + """ +import signal, time + +signal.signal(signal.SIGTERM, signal.SIG_IGN) +time.sleep(60) +""", + cwd=tmp_path, + ) + ) + await supervisor.start(timeout=5) + await supervisor.stop(timeout=1) + + assert supervisor.state is JsonlProcessState.STOPPED + assert supervisor.returncode is not None + + @pytest.mark.asyncio async def test_jsonl_supervisor_contract(tmp_path: Path) -> None: # invalid JSON and a nonzero exit are classified @@ -1000,7 +1121,7 @@ async def test_jsonl_supervisor_contract(tmp_path: Path) -> None: assert str(exited.fatal_error).endswith(": fatal app-server detail") await exited.stop(timeout=2) - # timeout, cancellation and restart are handled + # timeout and cancellation are handled supervisor = JsonlProcessSupervisor( python_process( """ @@ -1022,9 +1143,6 @@ async def test_jsonl_supervisor_contract(tmp_path: Path) -> None: with pytest.raises(asyncio.CancelledError): await cancelled await supervisor.stop(timeout=2) - - await supervisor.start(timeout=2) - await supervisor.stop(timeout=2) assert supervisor.state is JsonlProcessState.STOPPED # only consumed notifications are routed diff --git a/tests/contrib/test_orchestration.py b/tests/contrib/test_orchestration.py index 9a0cea3a..4e79d982 100644 --- a/tests/contrib/test_orchestration.py +++ b/tests/contrib/test_orchestration.py @@ -2,6 +2,7 @@ import asyncio import json +import os from collections.abc import Callable, Mapping, Sequence from dataclasses import replace from pathlib import Path @@ -258,7 +259,10 @@ def test_inbox_notice_carries_the_upgrade_line_inside_the_bracket() -> None: assert lines[-2].startswith( "Upgrade available: bazaar-compute-node 0.2.0 (installed 0.1.31)." ) - assert "`bcc node upgrade`" in lines[-2] + # the offer names whatever the platform can actually act on + assert ( + "`bcn system-service stop`" if os.name == "nt" else "`bcc node upgrade`" + ) in lines[-2] # case: half an answer is not an offer assert "Upgrade available" not in inbox_notice(