From 67b779f984a1ab1d41fa6c647d0e406875ad0b03 Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:38:16 -0400 Subject: [PATCH 1/3] adversarial: keep executor close bounded when a child holds stdout ExecutorClient.close() closed the buffered stdout stream after the wrapper had exited. If a child process it started still held stdout, the protocol reader was blocked in readline and holds the buffered stream's lock, so close waited until that child exited. The reader join with its timeout ran only after the close. Join the reader first. If it is still blocked, do not close stdout; report that a surviving process holds the executor output. The reader is a daemon thread and ends when that process exits. Adds a real-process regression test in which the wrapper exits and leaves a child holding stdout. Before the change, close blocked for the child's full lifetime and returned without error. Fixes #503 (executor close blocks on inherited stdout) Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/adversarial.py | 15 ++++++++- scripts/test_adversarial.py | 54 ++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/scripts/nvx_tools/adversarial.py b/scripts/nvx_tools/adversarial.py index 9439e8822..7d4e550c9 100644 --- a/scripts/nvx_tools/adversarial.py +++ b/scripts/nvx_tools/adversarial.py @@ -563,13 +563,26 @@ def close(self) -> None: cleanup_error = error finally: self._closed = True - for stream in (self._stdin, self._stdout, self._stderr_stream): + for stream in (self._stdin, self._stderr_stream): try: stream.close() except OSError as error: if cleanup_error is None: cleanup_error = error self._reader.join(timeout=2.0) + if self._reader.is_alive(): + # A process that outlived the wrapper still holds stdout. Closing the buffered + # stream would wait for the blocked reader, so leave it to the daemon thread. + if cleanup_error is None: + cleanup_error = ScriptError( + "executor output is still held open by a surviving process" + ) + else: + try: + self._stdout.close() + except OSError as error: + if cleanup_error is None: + cleanup_error = error if cleanup_error is not None: raise cleanup_error diff --git a/scripts/test_adversarial.py b/scripts/test_adversarial.py index 996eba46b..97fc2e601 100644 --- a/scripts/test_adversarial.py +++ b/scripts/test_adversarial.py @@ -23,6 +23,7 @@ from nvx_tools.adversarial import ( LOCAL_EXECUTOR_STATE_ROOT, + SCHEMA_VERSION, CampaignConfig, CopilotController, ExecutorClient, @@ -2764,5 +2765,58 @@ def fake_run( self.assertIn(case_id, repair_prompt) +class ExecutorCloseTests(unittest.TestCase): + def test_close_is_bounded_when_a_surviving_child_holds_executor_stdout(self): + # The wrapper exits after starting a child that inherits its stdout, so the protocol + # reader stays blocked. Closing must not wait for that child to exit. + with tempfile.TemporaryDirectory(prefix="nvx-executor-close-") as temporary: + root = Path(temporary) + run_dir = root / "run" + run_dir.mkdir() + pid_file = root / "child.pid" + child = ( + f'"{sys.executable}" -c "import os, sys, time; ' + "open(sys.argv[1], 'w').write(str(os.getpid())); time.sleep(20)\"" + f' "{pid_file}"' + ) + if os.name == "nt": + wrapper = root / "wrapper.cmd" + wrapper.write_bytes(f'@echo off\r\nstart "" /B {child}\r\n'.encode()) + else: + wrapper = root / "wrapper.sh" + wrapper.write_bytes(f"#!/bin/sh\n{child} &\n".encode()) + wrapper.chmod(0o755) + client = ExecutorClient( + run_dir=run_dir, + command=str(wrapper), + label="surviving-child", + ) + try: + with self.assertRaises(ExecutorUnavailableError): + client.request( + {"schema_version": SCHEMA_VERSION, "operation": "shutdown"}, + timeout=0.5, + ) + started = time.monotonic() + with self.assertRaises(ScriptError): + client.close() + self.assertLess(time.monotonic() - started, 8.0) + finally: + # The surviving child holds the log and must not outlive the test. + deadline = time.monotonic() + 5.0 + while not pid_file.exists() and time.monotonic() < deadline: + time.sleep(0.05) + if pid_file.exists(): + pid = pid_file.read_text(encoding="utf-8") + if os.name == "nt": + subprocess.run( + ["taskkill.exe", "/PID", pid, "/T", "/F"], + check=False, + capture_output=True, + ) + else: + os.kill(int(pid), signal.SIGKILL) + + if __name__ == "__main__": unittest.main() From c685042c2ee811240df55107caedc92e06c339e2 Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:19:51 -0400 Subject: [PATCH 2/3] adversarial: drop surviving output and close stdout from the reader When close gives up on a surviving process, the reader kept queueing its output for a client that no longer reads, and stdout was closed only on the synchronous path, so a deferred close never released the pipe. Once close abandons the reader, the reader drains and discards what the surviving process writes, and the reader closes stdout when its output ends. Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/nvx_tools/adversarial.py | 31 ++++++++++++++++++------------- scripts/test_adversarial.py | 23 +++++++++++++++++++++++ 2 files changed, 41 insertions(+), 13 deletions(-) diff --git a/scripts/nvx_tools/adversarial.py b/scripts/nvx_tools/adversarial.py index 7d4e550c9..45f178a1a 100644 --- a/scripts/nvx_tools/adversarial.py +++ b/scripts/nvx_tools/adversarial.py @@ -485,6 +485,8 @@ def __init__( self._stdin = self._process.stdin self._stdout = self._process.stdout self._responses: queue.Queue[bytes | None] = queue.Queue() + # Set when close gives up on a reader that a surviving process keeps blocked. + self._abandoned = False self._reader = threading.Thread( target=self._read_responses, name=f"nvx-executor-reader-{label}", @@ -572,17 +574,13 @@ def close(self) -> None: self._reader.join(timeout=2.0) if self._reader.is_alive(): # A process that outlived the wrapper still holds stdout. Closing the buffered - # stream would wait for the blocked reader, so leave it to the daemon thread. + # stream would wait for the blocked reader, so the reader closes it when it + # ends, and it retains nothing further. + self._abandoned = True if cleanup_error is None: cleanup_error = ScriptError( "executor output is still held open by a surviving process" ) - else: - try: - self._stdout.close() - except OSError as error: - if cleanup_error is None: - cleanup_error = error if cleanup_error is not None: raise cleanup_error @@ -590,12 +588,19 @@ def abort(self) -> None: self._terminate() def _read_responses(self) -> None: - while line := self._stdout.readline(MAX_PROTOCOL_LINE_BYTES + 1): - if len(line) > MAX_PROTOCOL_LINE_BYTES: - self._responses.put(None) - return - self._responses.put(line) - self._responses.put(None) + try: + while line := self._stdout.readline(MAX_PROTOCOL_LINE_BYTES + 1): + if self._abandoned: + # Keep draining so a surviving process is not blocked on a full pipe, + # but retain nothing once the client has been closed. + continue + if len(line) > MAX_PROTOCOL_LINE_BYTES: + self._responses.put(None) + return + self._responses.put(line) + self._responses.put(None) + finally: + self._stdout.close() def _terminate(self) -> None: if self._process.poll() is None: diff --git a/scripts/test_adversarial.py b/scripts/test_adversarial.py index 97fc2e601..bf1e26eef 100644 --- a/scripts/test_adversarial.py +++ b/scripts/test_adversarial.py @@ -8,6 +8,7 @@ import hashlib import json import os +import queue import signal import socket import subprocess @@ -2817,6 +2818,28 @@ def test_close_is_bounded_when_a_surviving_child_holds_executor_stdout(self): else: os.kill(int(pid), signal.SIGKILL) + def test_reader_closes_its_stream_when_the_output_ends(self): + client = ExecutorClient.__new__(ExecutorClient) + client._stdout = MagicMock() + client._stdout.readline.side_effect = [b'{"ok": true}\n', b""] + client._responses = queue.Queue() + client._abandoned = False + client._read_responses() + client._stdout.close.assert_called_once_with() + self.assertEqual(client._responses.get_nowait(), b'{"ok": true}\n') + self.assertIsNone(client._responses.get_nowait()) + + def test_abandoned_reader_drains_output_without_retaining_it(self): + client = ExecutorClient.__new__(ExecutorClient) + client._stdout = MagicMock() + client._stdout.readline.side_effect = [b"late\n", b"later\n", b""] + client._responses = queue.Queue() + client._abandoned = True + client._read_responses() + client._stdout.close.assert_called_once_with() + self.assertIsNone(client._responses.get_nowait()) + self.assertTrue(client._responses.empty()) + if __name__ == "__main__": unittest.main() From 756f215b2587ebecd3cb0b44d861853f98f21d71 Mon Sep 17 00:00:00 2001 From: jay23606 <4602056+jay23606@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:21:27 -0400 Subject: [PATCH 3/3] test(adversarial): accept BrokenPipeError when close races an unflushed request Co-Authored-By: Claude Haiku 5.5 Signed-off-by: jay23606 <4602056+jay23606@users.noreply.github.com> --- scripts/test_adversarial.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/test_adversarial.py b/scripts/test_adversarial.py index bf1e26eef..9af8734e0 100644 --- a/scripts/test_adversarial.py +++ b/scripts/test_adversarial.py @@ -2799,7 +2799,9 @@ def test_close_is_bounded_when_a_surviving_child_holds_executor_stdout(self): timeout=0.5, ) started = time.monotonic() - with self.assertRaises(ScriptError): + # If the wrapper exited before request() flushed stdin, close() can see the + # buffered stdin data fail with BrokenPipeError instead of the stdout ScriptError. + with self.assertRaises((ScriptError, BrokenPipeError)): client.close() self.assertLess(time.monotonic() - started, 8.0) finally: