[codex] bound deadline process cleanup

packages/headless/README.md

@@ -222,6 +222,8 @@ The manual `.github/workflows/oracle-evidence-audit.yml` workflow is the only CI
 
 For an unattended run, invoke `node packages/headless/harbor/run-harness-ab-detached.mjs` with the same environment. It detaches the worker from the terminal and atomically journals `running`, `completed`, or `failed` in `background-run.json`; stdout and stderr go to `background-run.log`.
 
+Pinned CLI adapters isolate each shell command in a process scope so a task-native timeout can terminate the active process tree before Harbor invokes the verifier. Command output is buffered away from background descendants, then stdout and stderr are replayed concurrently with a 60-second hard bound and the same scope markers. If the caller has already abandoned its transport, replay is killed instead of holding the Docker exec pipe open. Deadline cleanup gives both TERM and KILL at most 10 seconds; a TERM transport failure followed by a successful KILL cannot replace Claude Code's original budget-exhausted cancellation, so Harbor can still grade artifacts written before the deadline.
+
 Binary outputs are `harness-ab-report.json`, `.csv`, and `.md`; multi-arm outputs are `harness-cohort-report.json`, `.csv`, and `.md`. The cohort JSON and Markdown retain aggregate Pass@1, cache-aware token/cost, duration, native protocol, and Oracle evidence, while its CSV is the canonical task × arm outcome projection. Every pairwise comparison is derived from the same cohort denominator rather than rerunning Maka as separate baselines. Report schema v4 records scheduled, attempted, model-scored, unscored (including the infrastructure-failed subset), and missing-final-usage cell coverage while keeping paired Pass@1 and token economy on separate denominators. Effectiveness pass rates are arm-local (each arm's passes over its own valid cells); the shared paired-sample delta is `pairedCandidateMinusBaseline`, and the `nonBudgetConditional` section discloses the subset where neither arm exhausted its budget. Runs with unilateral infra gaps therefore read differently from v3 reports, whose rates all shared the paired denominator. Account-plan runs record zero cost and use real token totals as the economy measure. Evidence gaps finish as `completed_with_gaps` and fail the completion assertion; an unattempted suffix remains `incomplete`. Reports do not claim fixed-plan spend or publish results.
 
 ## Attention semantic-compaction A/B

packages/headless/harbor/process_scope.py

@@ -10,6 +10,8 @@
 COMMAND_SCOPE_ENV = "MAKA_HARBOR_COMMAND_SCOPE"
 COMMAND_ID_ENV = "MAKA_HARBOR_COMMAND_ID"
 COMMAND_SCOPE_ROOT = "/tmp/maka-harbor-command-scopes"
+OUTPUT_REPLAY_TIMEOUT_SEC = 60
+PROCESS_SCOPE_CLEANUP_TIMEOUT_SEC = 10
 
 
 async def cleanup_process_scope(
@@ -20,6 +22,7 @@ async def cleanup_process_scope(
         await agent.exec_as_agent(
             environment,
             command=scoped_process_cleanup_command(scope, "TERM"),
+            timeout_sec=PROCESS_SCOPE_CLEANUP_TIMEOUT_SEC,
         )
     except BaseException as error:
         first_error = error
@@ -28,20 +31,33 @@ async def cleanup_process_scope(
         await agent.exec_as_agent(
             environment,
             command=scoped_process_cleanup_command(scope, "KILL"),
+            timeout_sec=PROCESS_SCOPE_CLEANUP_TIMEOUT_SEC,
         )
     except BaseException as error:
-        if first_error is None:
-            first_error = error
-    if first_error is not None:
-        raise first_error
+        if first_error is not None:
+            raise error from first_error
+        raise
 
 
-def scoped_command(command: str, scope: str, command_id: str) -> str:
+def scoped_command(
+    command: str,
+    scope: str,
+    command_id: str,
+    output_replay_timeout_sec: int = OUTPUT_REPLAY_TIMEOUT_SEC,
+) -> str:
+    if isinstance(output_replay_timeout_sec, bool) or not isinstance(
+        output_replay_timeout_sec, int
+    ) or output_replay_timeout_sec < 1:
+        raise ValueError("output replay timeout must be positive")
     scope_dir = shlex.quote(f"{COMMAND_SCOPE_ROOT}/{scope}")
     pgid_path = shlex.quote(f"{COMMAND_SCOPE_ROOT}/{scope}/{command_id}.pgid")
     wrapper_path = shlex.quote(f"{COMMAND_SCOPE_ROOT}/{scope}/{command_id}.wrapper")
     stdout_path = shlex.quote(f"{COMMAND_SCOPE_ROOT}/{scope}/{command_id}.stdout")
     stderr_path = shlex.quote(f"{COMMAND_SCOPE_ROOT}/{scope}/{command_id}.stderr")
+    replay_env = (
+        f"env {COMMAND_SCOPE_ENV}={shlex.quote(scope)} "
+        f"{COMMAND_ID_ENV}={shlex.quote(command_id)}"
+    )
     return (
         f"mkdir -p -- {scope_dir}; printf '%s\\n' \"$$\" > {wrapper_path}; set -m; "
         f"env {COMMAND_SCOPE_ENV}={shlex.quote(scope)} "
@@ -51,7 +67,12 @@ def scoped_command(command: str, scope: str, command_id: str) -> str:
         "set +m; "
         f"printf '%s\\n' \"$command_pid\" > {pgid_path}; "
         "wait \"$command_pid\" 2>/dev/null; command_status=$?; "
-        f"cat -- {stdout_path}; cat -- {stderr_path} >&2; "
+        f"{replay_env} timeout -s KILL {output_replay_timeout_sec} cat -- {stdout_path} & "
+        "stdout_replay_pid=$!; "
+        f"{replay_env} timeout -s KILL {output_replay_timeout_sec} cat -- {stderr_path} >&2 & "
+        "stderr_replay_pid=$!; "
+        "wait \"$stdout_replay_pid\" 2>/dev/null || true; "
+        "wait \"$stderr_replay_pid\" 2>/dev/null || true; "
         f"rm -f -- {stdout_path} {stderr_path}; "
         f"kill -0 -- \"-$command_pid\" 2>/dev/null || rm -f -- {pgid_path}; "
         f"rm -f -- {wrapper_path}; "

packages/headless/src/__tests__/harbor-adapter.test.ts

@@ -1208,7 +1208,9 @@ with tempfile.TemporaryDirectory() as tmp:
 
 function pythonBufferedWrapperCleanupSmokeScript(root: string): string {
   return String.raw`
+import contextlib
 import os
+import signal
 import subprocess
 import sys
 import time
@@ -1221,28 +1223,63 @@ from process_scope import COMMAND_SCOPE_ROOT, scoped_command, scoped_process_cle
 
 scope = f"buffered-wrapper-cleanup-{os.getpid()}"
 scope_dir = Path(COMMAND_SCOPE_ROOT) / scope
+stdout_path = scope_dir / "command.stdout"
+pgid_path = scope_dir / "command.pgid"
+
+def replay_pids():
+    matches = []
+    for cmdline_path in Path("/proc").glob("[0-9]*/cmdline"):
+        try:
+            cmdline = cmdline_path.read_bytes()
+        except (FileNotFoundError, PermissionError, ProcessLookupError):
+            continue
+        if cmdline.split(bytes([0]))[:2] == [b"cat", b"--"] and str(stdout_path).encode() in cmdline:
+            matches.append(int(cmdline_path.parent.name))
+    return matches
+
 process = subprocess.Popen(
-    ["bash", "-lc", scoped_command("head -c 1048576 /dev/zero; sleep 30", scope, "command")],
+    [
+        "bash",
+        "-lc",
+        scoped_command(
+            "head -c 1048576 /dev/zero; sleep 30",
+            scope,
+            "command",
+            output_replay_timeout_sec=1,
+        ),
+    ],
     stdout=subprocess.PIPE,
     stderr=subprocess.PIPE,
 )
 try:
-    stdout_path = scope_dir / "command.stdout"
     deadline = time.time() + 2
     while (not stdout_path.exists() or stdout_path.stat().st_size < 1048576) and time.time() < deadline:
         time.sleep(0.01)
     assert stdout_path.exists() and stdout_path.stat().st_size == 1048576
+    command_pgid = int(pgid_path.read_text(encoding="utf-8").strip())
+    os.killpg(command_pgid, signal.SIGKILL)
+    deadline = time.time() + 2
+    while not replay_pids() and time.time() < deadline:
+        time.sleep(0.01)
+    assert replay_pids(), "wrapper never entered buffered stdout replay"
+    process.wait(timeout=2)
+    deadline = time.time() + 1
+    while replay_pids() and time.time() < deadline:
+        time.sleep(0.01)
+    assert not replay_pids(), "bounded stdout replay did not settle"
     subprocess.run(
         ["bash", "-lc", scoped_process_cleanup_command(scope, "KILL")],
         check=True,
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
     )
-    process.wait(timeout=1)
 finally:
     if process.poll() is None:
         process.kill()
         process.wait()
+    for pid in replay_pids():
+        with contextlib.suppress(ProcessLookupError):
+            os.kill(pid, signal.SIGKILL)
 `;
 }
 
@@ -3807,6 +3844,9 @@ class ClaudeCode:
             "auth_token": self._get_env("ANTHROPIC_AUTH_TOKEN"),
             "oauth_token": self._get_env("CLAUDE_CODE_OAUTH_TOKEN"),
         })
+        if instruction == "cancel":
+            environment.started.set()
+            await asyncio.Event().wait()
         if instruction == "pipeline-fail":
             result = await self.exec_as_agent(
                 environment,
@@ -3856,9 +3896,16 @@ from claude_code_agent import MakaClaudeCodeAgent
 class Environment:
     def __init__(self):
         self.commands = []
+        self.cleanup_timeouts = []
+        self.started = asyncio.Event()
 
-    async def exec(self, command, env=None, **kwargs):
+    async def exec(self, command, env=None, timeout_sec=None, **kwargs):
         self.commands.append(command)
+        if "/proc/[0-9]*/environ" in command:
+            self.cleanup_timeouts.append(timeout_sec)
+            if "kill -TERM" in command:
+                raise RuntimeError("simulated stuck TERM cleanup transport")
+            return types.SimpleNamespace(return_code=0, stdout="", stderr="")
         process = await asyncio.create_subprocess_exec(
             "bash",
             "-lc",
@@ -3934,6 +3981,23 @@ with tempfile.TemporaryDirectory() as tmp:
     assert pipeline_cell["status"] == "failed", pipeline_cell
     assert pipeline_cell["errorClass"] == "infra_failed", pipeline_cell
 
+    environment.cleanup_timeouts.clear()
+    cancelled = agent(logs)
+
+    async def cancel_claude():
+        task = asyncio.create_task(cancelled.run("cancel", environment, context))
+        await asyncio.wait_for(environment.started.wait(), timeout=1)
+        task.cancel()
+        try:
+            await asyncio.wait_for(task, timeout=1)
+        except asyncio.CancelledError:
+            return
+        raise AssertionError("expected Claude Code cancellation")
+
+    asyncio.run(cancel_claude())
+    assert environment.cleanup_timeouts == [10, 10], environment.cleanup_timeouts
+    assert cancelled._failure_class == "budget_exhausted", cancelled._failure_class
+
 print("claude-code adapter ok")
 `;
 }