diff --git a/bridge/agentenv/capture.py b/bridge/agentenv/capture.py index 732efe5b70..c4491ec0c0 100644 --- a/bridge/agentenv/capture.py +++ b/bridge/agentenv/capture.py @@ -291,14 +291,30 @@ def run(arguments: dict) -> int: # The controller is sent with python -c, so a detached supervisor needs its -# own source. It alone holds the child process handle and signals its group. +# own source. It alone holds the child process handle. One SIGINT is not +# enough: a leader can ignore it, and a child in the same session can outlive +# the leader and keep stderr open. Signals go only to processes still in that +# session. A descendant that called setsid is reported, not signaled, and is +# not treated as exited. LONG_RUN = r''' import json, os, signal, subprocess, sys, threading, time job = sys.argv[1] result = {} child = None +pgid = None drainer = None +request = {} stderr_tail, stderr_errors = bytearray(), [] +deadline = None +timed_out = False +interrupted = False +force_stop = False +output_incomplete = False +leader_unsettled = False +stop_step = 0 +stop_next = None +GRACE_S = 2.0 +STOP_SIGNALS = (signal.SIGINT, signal.SIGTERM, signal.SIGKILL) def drain(stream): try: @@ -313,12 +329,215 @@ def drain(stream): except OSError as error: stderr_errors.append(str(error)) -def timed_out_marker(): +def mark_timed_out(): + global timed_out + if timed_out: + return + timed_out = True try: open(os.path.join(job, "timed_out"), "x").close() except FileExistsError: pass +def stop_requested(): + global interrupted + if deadline is not None and time.monotonic() >= deadline: + mark_timed_out() + cancelled = os.path.exists(os.path.join(job, "cancel")) + if cancelled: + interrupted = True + return force_stop or timed_out or cancelled + +def proc_identity(pid): + """Return state, process group, and session from /proc, or None. + + The command name is inside parentheses and can itself contain parentheses, + so the fields after the last closing parenthesis are the ones that count. + """ + try: + with open(f"/proc/{pid}/stat", "rb") as handle: + data = handle.read() + except OSError: + return None + end = data.rfind(b")") + if end < 0: + return None + fields = data[end + 2:].split() + if len(fields) < 4: + return None + try: + return fields[0].decode("ascii", "replace"), int(fields[2]), int(fields[3]) + except ValueError: + return None + +def live_owned_members(): + """Pids whose process group and session are still the leader's session. + + The leader's pid is that session id because it was started in a new + session. A zombie is not a member: it cannot hold stderr. After the leader + is reaped, the numeric id can be reused, so this never uses killpg. + """ + if pgid is None or child is None: + return [] + found = [] + try: + names = os.listdir("/proc") + except OSError: + names = [] + for name in names: + if not name.isdigit(): + continue + pid = int(name) + identity = proc_identity(pid) + if identity is None: + continue + state, group, session = identity + if group == pgid and session == pgid and state not in ("Z", "X"): + found.append(pid) + if child.poll() is None and child.pid not in found: + found.append(child.pid) + return found + +def signal_owned(sig): + """Signal current session members. + + An unreaped leader is still the process this supervisor started. Every + other pid is read from /proc again immediately before the signal. + """ + leader = child.pid if child is not None else None + for pid in live_owned_members(): + if leader is not None and pid == leader and child.poll() is None: + try: + os.kill(pid, sig) + except OSError: + pass + continue + identity = proc_identity(pid) + if ( + identity is None + or identity[0] in ("Z", "X") + or identity[1] != pgid + or identity[2] != pgid + ): + continue + try: + os.kill(pid, sig) + except OSError: + pass + +def escalate(): + """Send SIGINT, then SIGTERM, then SIGKILL, each once, with GRACE_S between them. + + Returns true only after the last grace. The leader exiting does not end + the sequence while a same-session child is still alive. + """ + global interrupted, stop_step, stop_next + forced = force_stop + if not stop_requested() or child is None: + return False + if child.poll() is not None and not live_owned_members(): + return False + if not forced: + interrupted = True + now = time.monotonic() + if stop_step < len(STOP_SIGNALS) and (stop_next is None or now >= stop_next): + signal_owned(STOP_SIGNALS[stop_step]) + stop_step += 1 + stop_next = now + GRACE_S + return stop_step >= len(STOP_SIGNALS) and stop_next is not None and now >= stop_next + +def stop_owned(): + """Finish the signal sequence for the leader and same-session children.""" + global leader_unsettled + while child is not None and child.poll() is None: + if escalate(): + if child.poll() is None or live_owned_members(): + leader_unsettled = True + return + time.sleep(0.2) + while stop_requested() and live_owned_members(): + if escalate(): + if live_owned_members(): + leader_unsettled = True + return + time.sleep(0.2) + +def abandon_drain(): + """Wait out stderr, and keep stopping same-session children after it closes. + + A descendant outside the session is not signaled. If it still holds stderr + after one grace, the result says the output is incomplete. + """ + global output_incomplete, leader_unsettled + if drainer is None: + return + grace_end = None + while drainer.is_alive() or (stop_requested() and live_owned_members()): + if stop_requested() and live_owned_members(): + grace_end = None + if escalate() and live_owned_members(): + leader_unsettled = True + if drainer.is_alive(): + output_incomplete = True + return + elif stop_requested() and drainer.is_alive(): + if grace_end is None: + grace_end = time.monotonic() + GRACE_S + elif time.monotonic() >= grace_end: + output_incomplete = True + return + elif not drainer.is_alive(): + return + else: + grace_end = None + if drainer.is_alive(): + drainer.join(0.2) + else: + time.sleep(0.2) + +def finish_stop(): + stop_owned() + if leader_unsettled and drainer is not None and drainer.is_alive(): + output_incomplete = True + elif not leader_unsettled: + abandon_drain() + +def write_result(): + if result.get("absent") is True: + result["settled"] = True + else: + if child is not None: + result.setdefault("exit_code", child.returncode) + result.setdefault("stderr", bytes(stderr_tail).decode("utf-8", "replace")) + stdout = request.get("stdout") + if isinstance(stdout, str): + try: + result.setdefault("stdout_bytes", os.path.getsize(stdout)) + except OSError: + pass + result.setdefault("timed_out", timed_out) + result.setdefault("cancelled", interrupted and not timed_out) + if stderr_errors and "error" not in result: + result["error"] = f"The REA command's stderr could not be read: {stderr_errors[0]}" + if leader_unsettled: + result["settled"] = False + result.setdefault("error", "The REA command did not exit after SIGINT, SIGTERM, and SIGKILL.") + elif output_incomplete: + result["settled"] = False + result.setdefault( + "error", + "A process outside the supervised group still holds the command's stderr, so its output is incomplete.", + ) + elif child is None or child.poll() is not None: + result["settled"] = True + else: + result["settled"] = False + result.setdefault("error", "The REA command supervisor lost the command before it exited.") + temporary = os.path.join(job, "result.tmp") + with open(temporary, "w", encoding="utf-8") as out: + json.dump(result, out) + os.replace(temporary, os.path.join(job, "result.json")) + try: request = json.loads("".join(sys.argv[2:])) if os.path.exists(os.path.join(job, "cancel")): @@ -327,29 +546,11 @@ try: with open(request["stdout"], "wb") as out: child = subprocess.Popen(request["argv"], cwd=request["home"], stdin=subprocess.DEVNULL, stdout=out, stderr=subprocess.PIPE, start_new_session=True) + pgid = child.pid drainer = threading.Thread(target=drain, args=(child.stderr,), daemon=True) drainer.start() deadline = time.monotonic() + request["timeout_s"] - interrupted = False - timed_out = False - while child.poll() is None: - timed_out = timed_out or time.monotonic() >= deadline - if not interrupted and (timed_out or os.path.exists(os.path.join(job, "cancel"))): - interrupted = True - if timed_out: - timed_out_marker() - try: - os.killpg(child.pid, signal.SIGINT) - except OSError: - pass - time.sleep(0.2) - # A descendant can still hold stderr after the leader exits. Do - # not publish completion until that stream has actually closed. - while drainer.is_alive(): - drainer.join(0.2) - if time.monotonic() >= deadline: - timed_out = True - timed_out_marker() + finish_stop() result = {"exit_code": child.returncode, "stdout_bytes": os.path.getsize(request["stdout"]), "timed_out": timed_out, "cancelled": interrupted and not timed_out, "stderr": bytes(stderr_tail).decode("utf-8", "replace")} @@ -357,29 +558,10 @@ try: result["error"] = f"The REA command's stderr could not be read: {stderr_errors[0]}" except Exception as error: result["error"] = f"The REA command supervisor failed: {error}" - # Keep the child handle until it exits, including when cancellation arrives - # after an unexpected supervisor error. - while child is not None and child.poll() is None: - if os.path.exists(os.path.join(job, "cancel")): - try: - os.killpg(child.pid, signal.SIGINT) - except OSError: - pass - time.sleep(0.2) - if drainer is not None and drainer.ident is not None: - drainer.join() - if child is not None: - result.setdefault("exit_code", child.returncode) - result.setdefault("stderr", bytes(stderr_tail).decode("utf-8", "replace")) - try: - result.setdefault("stdout_bytes", os.path.getsize(request["stdout"])) - except OSError: - pass -if child is None or child.poll() is not None: - temporary = os.path.join(job, "result.tmp") - with open(temporary, "w", encoding="utf-8") as out: - json.dump(result, out) - os.replace(temporary, os.path.join(job, "result.json")) + # The supervisor can no longer report a normal exit, so stop processes it still owns. + force_stop = True + finish_stop() +write_result() ''' diff --git a/bridge/agentenv/test_capture.py b/bridge/agentenv/test_capture.py index 7fed0437e0..bfc0041e89 100644 --- a/bridge/agentenv/test_capture.py +++ b/bridge/agentenv/test_capture.py @@ -5,9 +5,12 @@ import io from contextlib import redirect_stdout import os from pathlib import Path +import signal import subprocess import sys import tempfile +import threading +import time import unittest from unittest.mock import patch @@ -88,8 +91,31 @@ class LongRunTest(unittest.TestCase): "if len(sys.argv) > 1 and sys.argv[1] == 'stderr':\n" " sys.stderr.write('x' * 100000 + 'TAIL')\n" "if len(sys.argv) > 1 and sys.argv[1] == 'hold':\n" - " subprocess.Popen([sys.executable, '-c', 'import os,sys,time\\nwhile not os.path.exists(sys.argv[1]): time.sleep(0.02)', sys.argv[2]], stdout=subprocess.DEVNULL, stderr=sys.stderr, start_new_session=True)\n" + " held = subprocess.Popen([sys.executable, '-c', 'import os,sys,time\\nwhile not os.path.exists(sys.argv[1]): time.sleep(0.02)', sys.argv[2]], stdout=subprocess.DEVNULL, stderr=sys.stderr, start_new_session=True)\n" + " open(sys.argv[2] + '.pid.tmp', 'w').write(str(held.pid))\n" + " os.replace(sys.argv[2] + '.pid.tmp', sys.argv[2] + '.pid')\n" + " open(sys.argv[2] + '.leader.tmp', 'w').write(str(os.getpid()))\n" + " os.replace(sys.argv[2] + '.leader.tmp', sys.argv[2] + '.leader')\n" " sys.exit(0)\n" + "if len(sys.argv) > 1 and sys.argv[1] == 'group':\n" + " code = 'import os,signal,sys,time\\nsignal.signal(signal.SIGINT, signal.SIG_IGN)\\nsignal.signal(signal.SIGTERM, signal.SIG_IGN)\\nsignal.signal(signal.SIGHUP, signal.SIG_IGN)\\ntemporary = sys.argv[1] + \".tmp\"\\nopen(temporary, \"w\").write(str(os.getpid()))\\nos.replace(temporary, sys.argv[1])\\nwhile not os.path.exists(sys.argv[2]): time.sleep(0.02)\\n'\n" + " held = subprocess.Popen([sys.executable, '-c', code, sys.argv[2] + '.pid', sys.argv[2]], stdout=subprocess.DEVNULL, stderr=sys.stderr)\n" + " open(sys.argv[2] + '.leader.tmp', 'w').write(str(os.getpid()))\n" + " os.replace(sys.argv[2] + '.leader.tmp', sys.argv[2] + '.leader')\n" + " while not os.path.exists(sys.argv[2] + '.pid'): time.sleep(0.02)\n" + " sys.exit(0)\n" + "if len(sys.argv) > 1 and sys.argv[1] == 'drop':\n" + " code = 'import os,signal,sys,time\\nsignal.signal(signal.SIGINT, signal.SIG_IGN)\\nsignal.signal(signal.SIGTERM, signal.SIG_IGN)\\nsignal.signal(signal.SIGHUP, signal.SIG_IGN)\\ntemporary = sys.argv[1] + \".tmp\"\\nopen(temporary, \"w\").write(str(os.getpid()))\\nos.replace(temporary, sys.argv[1])\\nwhile not os.path.exists(sys.argv[2]): time.sleep(0.02)\\nnull = os.open(\"/dev/null\", os.O_WRONLY)\\nos.dup2(null, 2)\\nos.close(null)\\nopen(sys.argv[1] + \".closed\", \"w\").close()\\nwhile not os.path.exists(sys.argv[3]): time.sleep(0.02)\\n'\n" + " held = subprocess.Popen([sys.executable, '-c', code, sys.argv[3] + '.pid', sys.argv[2], sys.argv[3]], stdout=subprocess.DEVNULL, stderr=sys.stderr)\n" + " open(sys.argv[3] + '.leader.tmp', 'w').write(str(os.getpid()))\n" + " os.replace(sys.argv[3] + '.leader.tmp', sys.argv[3] + '.leader')\n" + " while not os.path.exists(sys.argv[3] + '.pid'): time.sleep(0.02)\n" + " sys.exit(0)\n" + "if len(sys.argv) > 1 and sys.argv[1] == 'resist':\n" + " signal.signal(signal.SIGINT, signal.SIG_IGN)\n" + " signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + " open(sys.argv[2], 'w').close()\n" + " while True: time.sleep(0.05)\n" "print(os.environ.get('AGENTENV_CONTROL_TOKEN', 'no token'))\n" ) rea.chmod(0o755) @@ -98,15 +124,45 @@ class LongRunTest(unittest.TestCase): self.job = self.root / "job" self.output = self.root / "out.json" - def command(self, name, **values): + def command(self, name, command_timeout=10, **values): result = subprocess.run([sys.executable, str(CONTROLLER), name, json.dumps(values)], - env=self.environment, text=True, capture_output=True, timeout=10, check=True) + env=self.environment, text=True, capture_output=True, timeout=command_timeout, check=True) return json.loads(result.stdout) def begin(self, argv, timeout=60): return self.command("run-begin", job=str(self.job), work=str(self.root), stdout=str(self.output), token="owner-token", argv=argv, files={}, timeout_s=timeout) + def wait_until(self, ready, seconds=5): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if ready(): + return True + time.sleep(0.02) + return False + + def leader_has_exited(self, pid): + try: + data = Path(f"/proc/{pid}/stat").read_bytes() + except FileNotFoundError: + return True + end = data.rfind(b")") + if end < 0: + return False + fields = data[end + 2:].split() + return bool(fields) and fields[0] == b"Z" + + def assert_process_gone(self, pid): + deadline = time.monotonic() + 2 + while True: + try: + os.kill(pid, 0) + except ProcessLookupError: + return + if time.monotonic() >= deadline: + self.fail(f"pid {pid} still exists") + time.sleep(0.05) + def test_runs_and_reports_its_output(self): self.assertEqual(self.begin([]), {"started": True}) result = self.command("run-poll", job=str(self.job), token="owner-token", timeout_s=5) @@ -128,7 +184,6 @@ class LongRunTest(unittest.TestCase): def test_cancels_a_running_child_and_confirms_exit(self): ready = self.root / "ready" self.assertEqual(self.begin(["wait", str(ready)]), {"started": True}) - import time deadline = time.monotonic() + 5 while not ready.exists() and time.monotonic() < deadline: time.sleep(0.02) @@ -169,17 +224,132 @@ class LongRunTest(unittest.TestCase): release.touch() self.command("run-poll", job=str(self.job), token="owner-token", timeout_s=5) - def test_waits_for_an_inherited_stderr_pipe_after_the_child_exits(self): + def test_reports_incomplete_stderr_while_a_new_session_descendant_holds_the_pipe(self): release = self.root / "release" - self.assertEqual(self.begin(["hold", str(release)], timeout=3), {"started": True}) + self.assertEqual(self.begin(["hold", str(release)], timeout=1), {"started": True}) try: - result = self.command("run-poll", job=str(self.job), token="owner-token", timeout_s=4) - self.assertTrue(result["pending"]) + result = self.command( + "run-poll", job=str(self.job), token="owner-token", timeout_s=8, command_timeout=15, + ) + self.assertTrue(result["finished"]) self.assertTrue(result["timed_out"]) + self.assertFalse(result["settled"]) + self.assertEqual(result["exit_code"], 0) + self.assertIn("outside the supervised group", result["error"]) + pid = int((self.root / "release.pid").read_text()) + self.assertEqual(os.getsid(pid), pid) + os.kill(pid, 0) finally: release.touch() - result = self.command("run-poll", job=str(self.job), token="owner-token", timeout_s=5) + + def test_cancel_during_inherited_stderr_drain_stays_unsettled(self): + release = self.root / "release" + self.assertEqual(self.begin(["hold", str(release)], timeout=30), {"started": True}) + pid_path = self.root / "release.pid" + leader_path = self.root / "release.leader" + try: + self.assertTrue(self.wait_until(lambda: pid_path.exists() and leader_path.exists())) + leader = int(leader_path.read_text()) + self.assertTrue(self.wait_until(lambda: self.leader_has_exited(leader))) + result = self.command( + "run-cancel", job=str(self.job), token="owner-token", timeout_s=8, command_timeout=15, + ) self.assertTrue(result["finished"]) + self.assertTrue(result["cancelled"]) + self.assertFalse(result["settled"]) + self.assertEqual(result["exit_code"], 0) + pid = int(pid_path.read_text()) + self.assertEqual(os.getsid(pid), pid) + os.kill(pid, 0) + finally: + release.touch() + + def test_cancel_after_leader_exit_kills_a_same_session_child_holding_stderr(self): + release = self.root / "release" + self.assertEqual(self.begin(["group", str(release)], timeout=30), {"started": True}) + pid_path = self.root / "release.pid" + leader_path = self.root / "release.leader" + try: + self.assertTrue(self.wait_until(lambda: pid_path.exists() and leader_path.exists())) + leader = int(leader_path.read_text()) + child = int(pid_path.read_text()) + self.assertTrue(self.wait_until(lambda: self.leader_has_exited(leader))) + self.assertEqual(os.getpgid(child), leader) + self.assertEqual(os.getsid(child), leader) + result = self.command( + "run-cancel", job=str(self.job), token="owner-token", timeout_s=12, command_timeout=20, + ) + self.assertTrue(result["finished"]) + self.assertTrue(result["cancelled"]) + self.assertTrue(result["settled"]) + self.assertEqual(result["exit_code"], 0) + self.assert_process_gone(child) + finally: + release.touch() + + def test_cancel_kills_a_same_session_child_that_closes_stderr(self): + release = self.root / "release" + stderr_release = self.root / "release.stderr" + self.assertEqual( + self.begin(["drop", str(stderr_release), str(release)], timeout=30), + {"started": True}, + ) + pid_path = self.root / "release.pid" + leader_path = self.root / "release.leader" + closed_path = self.root / "release.pid.closed" + observed = {} + + def close_stderr(): + if not self.wait_until(lambda: (self.job / "cancel").exists(), seconds=10): + observed["error"] = "cancel file did not appear" + return + time.sleep(0.3) + stderr_release.touch() + if not self.wait_until(lambda: closed_path.exists(), seconds=5): + observed["error"] = "child did not close stderr" + return + try: + os.kill(int(pid_path.read_text()), 0) + except ProcessLookupError: + observed["alive_after_close"] = False + else: + observed["alive_after_close"] = True + + closer = threading.Thread(target=close_stderr, daemon=True) + closer.start() + try: + self.assertTrue(self.wait_until(lambda: pid_path.exists() and leader_path.exists())) + leader = int(leader_path.read_text()) + child = int(pid_path.read_text()) + self.assertTrue(self.wait_until(lambda: self.leader_has_exited(leader))) + self.assertEqual(os.getpgid(child), leader) + self.assertEqual(os.getsid(child), leader) + result = self.command( + "run-cancel", job=str(self.job), token="owner-token", timeout_s=12, command_timeout=20, + ) + self.assertNotIn("error", observed, observed.get("error")) + self.assertTrue(observed.get("alive_after_close")) + self.assertTrue(result["finished"]) + self.assertTrue(result["cancelled"]) + self.assertTrue(result["settled"]) + self.assertEqual(result["exit_code"], 0) + self.assert_process_gone(child) + finally: + stderr_release.touch() + release.touch() + closer.join(timeout=2) + + def test_kills_a_leader_that_ignores_sigint_and_sigterm(self): + ready = self.root / "ready" + self.assertEqual(self.begin(["resist", str(ready)], timeout=1), {"started": True}) + result = self.command( + "run-poll", job=str(self.job), token="owner-token", timeout_s=12, command_timeout=20, + ) + self.assertTrue(result["finished"]) + self.assertTrue(result["timed_out"]) + self.assertTrue(result["settled"]) + self.assertFalse(result.get("pending", False)) + self.assertEqual(result["exit_code"], -signal.SIGKILL) def test_failure_closing_supervisor_log_after_launch_does_not_claim_exit(self): ready = self.root / "ready" diff --git a/docs/desktop-build-computer.md b/docs/desktop-build-computer.md index 43a6d6db48..177e843a9a 100644 --- a/docs/desktop-build-computer.md +++ b/docs/desktop-build-computer.md @@ -353,6 +353,18 @@ it, and tears down once that call has ended. Closing the MCP server closes every computer the same way, and a helper whose REA server exits sees its stdin close and tears down. +A timed-out or cancelled export command is stopped by that command's +supervisor. It sends SIGINT, then SIGTERM, then SIGKILL, with two seconds +between each, to processes that are still in the command's session. A child +that stays in that session after the leader exits receives the rest of the +sequence. A process that has left the session is not signaled. If that +process still holds the command's stderr, the result keeps the leader's exit +status and says the stderr is incomplete. That result does not mean the +outside process is gone. `close_build_computer` +still tears the computer down in that case. The teardown is what clears the +retained command. A reader that has not confirmed its own stop still keeps +the computer up so that reader can be retried. + The control token that guards the computer's framework-only routes stays in REA: no result, error or log carries it. Code running in the computer as its user can still read it, which is acceptable for a local build. diff --git a/src/application/desktop/DesktopBuildComputers.cleanup.test.ts b/src/application/desktop/DesktopBuildComputers.cleanup.test.ts index f1c4497fe7..434545f0f2 100644 --- a/src/application/desktop/DesktopBuildComputers.cleanup.test.ts +++ b/src/application/desktop/DesktopBuildComputers.cleanup.test.ts @@ -228,6 +228,70 @@ it.each([false, true])( }, ); +it("tears the computer down when a long run is unsettled, and resolves that run only after teardown", async () => { + const computerClose = vi + .fn() + .mockResolvedValueOnce({ + verified: false, + reason: "container busy", + resources: ["sandbox"], + }) + .mockResolvedValue({ verified: true, teardown: "agentenv" }); + const resolve = vi.spyOn( + BuildComputerLongRuns.prototype, + "resolveAfterVerifiedComputerTeardown", + ); + const commands = vi + .spyOn(BuildComputerLongRuns.prototype, "release") + .mockResolvedValue({ + verified: false, + reason: "command output is incomplete", + resources: ["job"], + }); + const spawn = vi + .spyOn(ComputerHelper, "spawn") + .mockResolvedValue({} as never); + const connect = vi.spyOn(AgentEnvComputer, "connect").mockResolvedValue({ + computer: { close: computerClose, sandboxId: "sandbox" }, + frame: { structured: null, content: [] }, + } as never); + const computers = new DesktopBuildComputers({ + REA_AGENT_ENV_COMMAND: "/runner", + REA_DOCKER_COMMAND: "/docker", + }); + try { + const started = await computers.start( + { source: { kind: "bundle", path: "/bundle" } }, + new AbortController().signal, + ); + if (!started.ok) throw started.error; + const id = started.value.value.computer_id; + const first = await computers.close(id); + if (first.ok) throw new Error("expected incomplete cleanup"); + expect(computerClose).toHaveBeenCalledTimes(1); + expect(resolve).not.toHaveBeenCalled(); + expect(first.error.cleanupResources).toEqual( + expect.arrayContaining(["job", "sandbox"]), + ); + expect(first.error.diagnostics?.reason).toContain( + "command output is incomplete", + ); + expect(first.error.diagnostics?.reason).toContain("container busy"); + const second = await computers.close(id); + expect(second.ok).toBe(true); + if (!second.ok) throw new Error("expected teardown"); + expect(computerClose).toHaveBeenCalledTimes(2); + expect(resolve).toHaveBeenCalledTimes(1); + expect(first.error.cleanupIncomplete).toBe(false); + } finally { + await computers.closeAll().catch(() => computers.closeAll()); + resolve.mockRestore(); + commands.mockRestore(); + spawn.mockRestore(); + connect.mockRestore(); + } +}); + it("retains the computer while command and reader cleanup fail, but still releases the judge", async () => { const computerClose = vi .fn() diff --git a/src/application/desktop/DesktopBuildComputers.ts b/src/application/desktop/DesktopBuildComputers.ts index 9091a8dc80..579592c463 100644 --- a/src/application/desktop/DesktopBuildComputers.ts +++ b/src/application/desktop/DesktopBuildComputers.ts @@ -283,7 +283,10 @@ const liveComputer = ( attempt(() => witness.readers.release()), attempt(() => judge.close()), ]); - if (!commands.verified || !readers.verified) { + // Reader cleanup can be retried against the live computer. An unsettled + // long run cannot: its stop already ended, and the computer teardown is + // what removes a descendant the supervisor does not own. + if (!readers.verified) { const failed = [commands, readers, judgeResult].filter( (result) => !result.verified, ); @@ -297,10 +300,16 @@ const liveComputer = ( closedComputer === undefined ? await attempt(() => computer.close()) : closedComputer; - if (computerResult.verified) closedComputer = computerResult; - const failed = [judgeResult, computerResult].filter( - (result) => !result.verified, - ); + if (computerResult.verified) { + closedComputer = computerResult; + if (!commands.verified) + capture.longRuns.resolveAfterVerifiedComputerTeardown(); + } + const failed = [ + judgeResult, + computerResult, + ...(!computerResult.verified && !commands.verified ? [commands] : []), + ].filter((result) => !result.verified); if (failed.length === 0 && computerResult.verified) return computerResult; return { verified: false, diff --git a/src/desktop/BuildComputerController.ts b/src/desktop/BuildComputerController.ts index ca538ba2fe..148ab4849c 100644 --- a/src/desktop/BuildComputerController.ts +++ b/src/desktop/BuildComputerController.ts @@ -164,6 +164,8 @@ export const answerSchema = z.looseObject({ output_bytes: z.int().nonnegative().optional(), stdout_bytes: z.int().nonnegative().optional(), stderr: z.string().optional(), + /** False when the supervisor published a result but the process group is not confirmed gone. */ + settled: z.boolean().optional(), apt: z .looseObject({ lists: z.int().nonnegative(), error: z.string().optional() }) .optional(), diff --git a/src/desktop/BuildComputerLongRuns.test.ts b/src/desktop/BuildComputerLongRuns.test.ts index 9803f025ea..cd6df8faf9 100644 --- a/src/desktop/BuildComputerLongRuns.test.ts +++ b/src/desktop/BuildComputerLongRuns.test.ts @@ -282,6 +282,161 @@ it("preserves a completed command's receipt if the supervisor reports a later fa expect(owner.cleanupResources).toEqual([]); }); +it("retains an unsettled finished command until computer teardown clears it", async () => { + const computer = { + exec: async (argv: readonly string[]) => { + if (argv[4] === "run-begin") return response({ started: true }); + if (argv[4] === "run-poll" || argv[4] === "run-cancel") + return response({ + finished: true, + settled: false, + exit_code: 0, + stdout_bytes: 4, + stderr: "partial", + timed_out: true, + error: + "A process outside the supervised group still holds the command's stderr, so its output is incomplete.", + }); + throw new Error("unexpected command"); + }, + homeFiles: async () => new Map(), + redacted: (text: string) => text, + } as ControlledComputer; + const owner = new BuildComputerLongRuns(computer); + const failure = await owner + .run(["export-desktop-application"], options(new AbortController().signal)) + .catch((error: unknown) => error); + if (!(failure instanceof ProviderCleanupError)) + throw new Error("expected retained cleanup"); + expect(failure.cleanupPrimary?.userMessage).toContain( + "outside the supervised group", + ); + expect(failure.diagnostics).toMatchObject({ + completed_command: { stdout_bytes: 4, exit_code: 0, timed_out: true }, + }); + expect(await owner.release()).toMatchObject({ verified: false }); + expect(owner.cleanupResources).toHaveLength(1); + expect(failure.cleanupIncomplete).toBe(true); + owner.resolveAfterVerifiedComputerTeardown(); + expect(owner.cleanupResources).toEqual([]); + expect(failure.cleanupIncomplete).toBe(false); +}); + +it.each([ + { cancelOutcome: "pending", exitCode: 0 }, + { cancelOutcome: "transport failure", exitCode: 0 }, + { cancelOutcome: "transport failure", exitCode: null }, +])( + "keeps a polled command receipt after $cancelOutcome (exit: $exitCode)", + async ({ cancelOutcome, exitCode }) => { + const computer = { + exec: async (argv: readonly string[]) => { + if (argv[4] === "run-begin") return response({ started: true }); + if (argv[4] === "run-poll") + return response({ + finished: true, + settled: false, + exit_code: exitCode, + stdout_bytes: 9, + stderr: "observed output", + timed_out: true, + cancelled: false, + error: "inherited stderr remains open", + }); + if (argv[4] === "run-cancel") { + if (cancelOutcome === "transport failure") + throw new Error("transport lost"); + return response({ pending: true }); + } + throw new Error("unexpected command"); + }, + homeFiles: async () => new Map(), + redacted: (text: string) => text, + } as ControlledComputer; + const owner = new BuildComputerLongRuns(computer); + const failure = await owner + .run( + ["export-desktop-application"], + options(new AbortController().signal), + ) + .catch((error: unknown) => error); + if (!(failure instanceof ProviderCleanupError)) + throw new Error("expected retained cleanup"); + const expected = { + completed_command: { + exit_code: exitCode, + stdout_bytes: 9, + stderr: "observed output", + settled: false, + timed_out: true, + cancelled: false, + }, + }; + expect(failure.diagnostics).toMatchObject(expected); + expect(failure.cleanupPrimary?.diagnostics).toMatchObject(expected); + expect(owner.cleanupResources).toHaveLength(1); + owner.resolveAfterVerifiedComputerTeardown(); + expect(failure.cleanupIncomplete).toBe(false); + expect(failure.diagnostics).toMatchObject(expected); + }, +); + +it("keeps command facts when a later release first sees an unsettled result", async () => { + const controller = new AbortController(); + let stops = 0; + const computer = { + exec: async (argv: readonly string[]) => { + if (argv[4] === "run-begin") return response({ started: true }); + if (argv[4] === "run-poll") { + controller.abort(new Error("cancelled")); + throw controller.signal.reason; + } + if (argv[4] === "run-cancel") { + stops += 1; + if (stops === 1) return response({ pending: true }); + return response({ + finished: true, + settled: false, + exit_code: 0, + stdout_bytes: 9, + stderr: "partial", + timed_out: true, + cancelled: false, + error: + "A process outside the supervised group still holds the command's stderr, so its output is incomplete.", + }); + } + throw new Error("unexpected command"); + }, + homeFiles: async () => new Map(), + redacted: (text: string) => text, + } as ControlledComputer; + const owner = new BuildComputerLongRuns(computer); + const failure = await owner + .run(["export-desktop-application"], options(controller.signal)) + .catch((error: unknown) => error); + if (!(failure instanceof ProviderCleanupError)) + throw new Error("expected retained cleanup"); + expect(failure.diagnostics).not.toHaveProperty("completed_command"); + expect(await owner.release()).toMatchObject({ verified: false }); + expect(stops).toBe(2); + expect(failure.diagnostics).toMatchObject({ + completed_command: { + stdout_bytes: 9, + exit_code: 0, + stderr: "partial", + timed_out: true, + cancelled: false, + }, + }); + owner.resolveAfterVerifiedComputerTeardown(); + expect(owner.cleanupResources).toEqual([]); + expect(failure.cleanupIncomplete).toBe(false); + expect(failure.diagnostics).toMatchObject({ + completed_command: { stdout_bytes: 9, exit_code: 0, stderr: "partial" }, + }); +}); + it("preserves a new cancellation while retrying a previous job's cleanup", async () => { const first = new AbortController(); const second = new AbortController(); diff --git a/src/desktop/BuildComputerLongRuns.ts b/src/desktop/BuildComputerLongRuns.ts index 42fa24c95d..10313e7548 100644 --- a/src/desktop/BuildComputerLongRuns.ts +++ b/src/desktop/BuildComputerLongRuns.ts @@ -73,6 +73,7 @@ export class BuildComputerLongRuns { }; this.#jobs.add(job); const deadline = performance.now() + options.timeoutS * 1000; + let observedCommand: JsonValue | undefined; try { const started = answerSchema.parse( await runBuildController( @@ -128,6 +129,14 @@ export class BuildComputerLongRuns { ? answer.error : "The REA command did not report its outcome.", ); + if (answer.settled === false) { + observedCommand = this.#receipt(job, answer); + throw new Error( + typeof answer.error === "string" + ? answer.error + : "The REA command did not confirm that its process had exited.", + ); + } this.#jobs.delete(job); if (answer.timed_out === true || typeof answer.error === "string") { const failure = new ProviderAdapterError( @@ -167,19 +176,25 @@ export class BuildComputerLongRuns { }); if (signal.aborted && cause instanceof AnalysisError && cause !== primary) primary.retainSecondaryFailure(cause); + if (observedCommand !== undefined) + primary.retainDiagnostics({ completed_command: observedCommand }); if (!this.#jobs.has(job)) throw primary; const released = await this.#cancel(job); + const completed = released.completed ?? observedCommand; if (signal.aborted && !(primary instanceof AnalysisCancelledError)) { const cancelled = cancelledBy(OPERATION, signal); cancelled.retainSecondaryFailure(primary); primary = cancelled; } if (released.verified) { - if (released.completed !== undefined) - primary.retainDiagnostics({ completed_command: released.completed }); + if (completed !== undefined) + primary.retainDiagnostics({ completed_command: completed }); throw primary; } - throw this.#cleanupFailure(released.reason, primary, job); + const failure = this.#cleanupFailure(released.reason, primary, job); + if (completed !== undefined) + failure.retainDiagnostics({ completed_command: completed }); + throw failure; } } @@ -199,11 +214,21 @@ export class BuildComputerLongRuns { }; } - async #cancel( - job: Job, - ): Promise< + /** Clear retained jobs after a verified computer teardown removed their processes. */ + resolveAfterVerifiedComputerTeardown(): void { + for (const job of this.#jobs) { + for (const reference of job.failures) reference.deref()?.resolveCleanup(); + } + this.#jobs.clear(); + } + + async #cancel(job: Job): Promise< | { readonly verified: true; readonly completed?: JsonValue } - | { readonly verified: false; readonly reason: string } + | { + readonly verified: false; + readonly reason: string; + readonly completed?: JsonValue; + } > { try { const answer = answerSchema.parse( @@ -218,6 +243,21 @@ export class BuildComputerLongRuns { { timeoutS: CANCEL_S }, ), ); + if (answer.finished === true && answer.settled === false) { + const completed = this.#receipt(job, answer); + for (const reference of job.failures) + reference + .deref() + ?.retainDiagnostics({ completed_command: completed }); + return { + verified: false, + reason: + typeof answer.error === "string" + ? answer.error + : "The REA command did not confirm that its process had exited.", + completed, + }; + } if (answer.absent !== true && answer.finished !== true) return { verified: false, @@ -252,9 +292,9 @@ export class BuildComputerLongRuns { ...(typeof answer.stdout_bytes === "number" ? { stdout_bytes: answer.stdout_bytes } : {}), - ...(typeof answer.exit_code === "number" - ? { exit_code: answer.exit_code } - : {}), + ...(answer.exit_code === undefined + ? {} + : { exit_code: answer.exit_code }), ...(typeof answer.stderr === "string" ? { stderr: answer.stderr } : {}), ...(typeof answer.error === "string" ? { error: answer.error } : {}), ...(answer.timed_out === undefined @@ -263,6 +303,7 @@ export class BuildComputerLongRuns { ...(answer.cancelled === undefined ? {} : { cancelled: answer.cancelled }), + ...(answer.settled === undefined ? {} : { settled: answer.settled }), }; }