diff --git a/crates/tracedecay-cli/tests/refresh_memory_journey.rs b/crates/tracedecay-cli/tests/refresh_memory_journey.rs index 22f2ea026b..81bc33297a 100644 --- a/crates/tracedecay-cli/tests/refresh_memory_journey.rs +++ b/crates/tracedecay-cli/tests/refresh_memory_journey.rs @@ -17,6 +17,7 @@ use std::time::{Duration, Instant}; use serde_json::{Value, json}; use tempfile::TempDir; +use tracedecay_runtime_core::logging::log_daemon_event; use common::fixture::{git_capture, git_run}; use common::{ @@ -32,6 +33,17 @@ const FUNCTIONS_PER_FILE: usize = 160; const INDEX_TIMEOUT: Duration = Duration::from_secs(900); const STATUS_POLL: Duration = Duration::from_secs(5); +fn report_phase(started: Instant, phase: &str, edit: usize) { + log_daemon_event( + "refresh_memory_journey_phase", + &[ + ("phase", phase.to_owned()), + ("edit", edit.to_string()), + ("elapsed_ms", started.elapsed().as_millis().to_string()), + ], + ); +} + /// Every function of every file changes with `edit`, so each refresh /// reparses the whole corpus. fn write_corpus(project: &Path, edit: usize) { @@ -53,12 +65,25 @@ fn write_corpus(project: &Path, edit: usize) { } fn run_cli(home: &Path, project: &Path, args: &[&str]) -> Vec { + let started = Instant::now(); + log_daemon_event( + "refresh_memory_journey_command_started", + &[("args", format!("{args:?}"))], + ); let output = tracedecay_command_with_home(home) .args(args) .current_dir(project) .stdin(Stdio::null()) .output() .unwrap_or_else(|error| panic!("tracedecay {args:?}: {error}")); + log_daemon_event( + "refresh_memory_journey_command_finished", + &[ + ("args", format!("{args:?}")), + ("elapsed_ms", started.elapsed().as_millis().to_string()), + ("status", output.status.to_string()), + ], + ); assert!( output.status.success(), "tracedecay {args:?} failed\nstdout:\n{}\nstderr:\n{}", @@ -104,9 +129,19 @@ fn status(home: &Path, project: &Path) -> Value { /// Waits until the index serves a complete generation of `revision` and /// returns that generation's id. fn wait_for_current(home: &Path, project: &Path, revision: &str) -> String { - let deadline = Instant::now() + INDEX_TIMEOUT; + let started = Instant::now(); + let deadline = started + INDEX_TIMEOUT; loop { let status = status(home, project); + log_daemon_event( + "refresh_memory_journey_status", + &[ + ("expected_revision", revision.to_owned()), + ("elapsed_ms", started.elapsed().as_millis().to_string()), + ("freshness", status["code_index_freshness"].to_string()), + ("memory", status["memory"].to_string()), + ], + ); let worktree = &status["code_index_freshness"]["worktree"]; if status["code_index_freshness"]["status"] == "current" && worktree["source_revision"] == revision @@ -156,6 +191,8 @@ fn assert_within_limit(home: &Path, project: &Path, pid: u32) { #[test] fn repeated_heavy_refreshes_under_the_limit_are_admitted() { + let started = Instant::now(); + report_phase(started, "prepare_corpus", 0); let home = TempDir::new().expect("isolated home"); let home = canonical_existing_path(home.path()); let workspace = TempDir::new().expect("project root"); @@ -168,23 +205,43 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() { let cold_revision = git_capture(&project, &["rev-parse", "HEAD"]); let log = home.join("daemon.log"); + report_phase(started, "start_daemon", 0); let daemon = spawn_tracedecay_daemon_logged(&home, &log, |command| { command .env( "TRACEDECAY_RESIDENT_MEMORY_LIMIT_BYTES", LIMIT_BYTES.to_string(), ) - .env("RUST_LOG", "info"); + .env("TRACEDECAY_TEST_DAEMON_LOG_STDERR", "1") + .env("TRACEDECAY_SPAN_TIMINGS", "1") + .env( + "RUST_LOG", + concat!( + "info,", + "tracedecay_code_index_runtime::code_index_scheduler::reconcile=trace,", + "tracedecay_code_index_runtime::code_index_scheduler::registry=trace,", + // Keep registry submodules at info: only its phase-level spans + // should be timed, not per-file/page or serving-read work. + "tracedecay_code_index_runtime::code_index_scheduler::registry::=info,", + "tracedecay::daemon::project_open_admission=trace", + ), + ); }); + report_phase(started, "initialize_project", 0); initialize_tracedecay_cli_project(&home, &project); + report_phase(started, "wait_for_current", 0); let mut generation = wait_for_current(&home, &project, &cold_revision); assert_within_limit(&home, &project, daemon.id()); + report_phase(started, "generation_current", 0); for edit in 1..=REFRESHES { + report_phase(started, "prepare_corpus", edit); write_corpus(&project, edit); git_run(&project, &["commit", "--quiet", "-am", "edit every file"]); let revision = git_capture(&project, &["rev-parse", "HEAD"]); + report_phase(started, "sync", edit); run_cli(&home, &project, &["sync"]); + report_phase(started, "wait_for_current", edit); let refreshed = wait_for_current(&home, &project, &revision); assert_ne!( refreshed, generation, @@ -192,6 +249,7 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() { ); generation = refreshed; assert_within_limit(&home, &project, daemon.id()); + report_phase(started, "generation_current", edit); } let log = fs::read_to_string(&log).expect("daemon log"); @@ -206,4 +264,5 @@ fn repeated_heavy_refreshes_under_the_limit_are_admitted() { "the daemon refused work it had room for:\n{log}" ); drop(daemon); + report_phase(started, "complete", REFRESHES); } diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs index 827b125218..735d40db1b 100644 --- a/crates/tracedecay/tests/common/mod.rs +++ b/crates/tracedecay/tests/common/mod.rs @@ -693,6 +693,7 @@ fn drain_daemon_stderr( startup: String, mut stderr: BufReader, log: Option, + mirror_stderr: bool, ) { std::thread::spawn(move || { if let Some(path) = @@ -703,6 +704,18 @@ fn drain_daemon_stderr( .open(path) { let _ = file.write_all(startup.as_bytes()); + if mirror_stderr { + let mut line = Vec::new(); + while stderr + .read_until(b'\n', &mut line) + .is_ok_and(|read| read != 0) + { + let _ = file.write_all(&line); + let _ = std::io::stderr().write_all(&line); + line.clear(); + } + return; + } let _ = std::io::copy(&mut stderr, &mut file); return; } @@ -1165,6 +1178,8 @@ pub fn spawn_tracedecay_daemon_with( /// [`spawn_tracedecay_daemon_with`], streaming the daemon's stderr to `log` /// for assertions over its diagnostic events. +/// Set `TRACEDECAY_TEST_DAEMON_LOG_STDERR=1` in `configure` to also stream the +/// diagnostics to test stderr, preserving evidence when a runner kills a test. pub fn spawn_tracedecay_daemon_logged( home: &Path, log: &Path, @@ -1276,6 +1291,9 @@ fn spawn_tracedecay_daemon_process( .stdout(Stdio::null()) .stderr(Stdio::piped()); configure(&mut command); + let mirror_stderr = command.get_envs().any(|(key, value)| { + key == "TRACEDECAY_TEST_DAEMON_LOG_STDERR" && value == Some(OsStr::new("1")) + }); bind_to_test_process(&mut command); let child = command.spawn().expect("tracedecay daemon should start"); let mut daemon = DaemonProcess::new(child); @@ -1315,6 +1333,9 @@ fn spawn_tracedecay_daemon_process( let read = stderr .read_line(&mut startup) .expect("daemon stderr should be readable"); + if mirror_stderr { + let _ = std::io::stderr().write_all(startup[line_start..].as_bytes()); + } if read == 0 { let status = daemon .child @@ -1329,7 +1350,7 @@ fn spawn_tracedecay_daemon_process( break; } } - drain_daemon_stderr(startup, stderr, log); + drain_daemon_stderr(startup, stderr, log, mirror_stderr); daemon } diff --git a/evals/memory/BUILD.bazel b/evals/memory/BUILD.bazel index 2b78708c10..5dd316b191 100644 --- a/evals/memory/BUILD.bazel +++ b/evals/memory/BUILD.bazel @@ -7,6 +7,7 @@ py_test( "//benchmark_data:run_benchmarks.py", "//evals/hermetic:score.py", "//scripts:build-eval-binary.sh", + "//scripts:lib/portable_process.py", ], imports = ["."], legacy_create_init = 0, diff --git a/evals/memory/README.md b/evals/memory/README.md index cc8d813f44..a5d70af0ad 100644 --- a/evals/memory/README.md +++ b/evals/memory/README.md @@ -30,6 +30,12 @@ python3 evals/memory/run_real_model.py --scenario memory-no-pollution \ The real-model runner creates a unique temporary user home for each run and installs the normal user-level Hermes integration there. It never selects a named Hermes profile or uses `HERMES_HOME` as TraceDecay storage. +Each scenario also owns an isolated `tracedecay daemon run` (private socket, +started via `scripts/lib/portable_process.py`) that must be ready before +`init`. Setup, daemon, agent, and cancellation failures are written to the +report as error entries. Cleanup is attempted on every exit path. If owned +work cannot be stopped or deletion fails, retained paths and cleanup errors +are recorded in the report. Run reports land under `evals/memory/runs//` (gitignored). diff --git a/evals/memory/run_real_model.py b/evals/memory/run_real_model.py index 0136f31486..cbb039793e 100755 --- a/evals/memory/run_real_model.py +++ b/evals/memory/run_real_model.py @@ -25,6 +25,7 @@ import os import re import shutil +import signal import sqlite3 import subprocess import sys @@ -34,8 +35,15 @@ EVAL_DIR = Path(__file__).resolve().parent REPO_ROOT = EVAL_DIR.parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts/lib")) + +from portable_process import dies_with_this_process, stop_owned_process + SCENARIO_DIR = EVAL_DIR / "scenarios" RUNS_DIR = EVAL_DIR / "runs" +PROCESS_HELPER = REPO_ROOT / "scripts/lib/portable_process.py" +DAEMON_READY_TIMEOUT_SECONDS = 60 +PROCESS_STOP_GRACE_SECONDS = 5 DEFAULT_HERMES_DIR = Path.home() / "hermes-agent" DEFAULT_MODEL = "gpt-5.4-mini" @@ -122,44 +130,162 @@ class EvalEnvironment: data_dir: Path global_db: Path env: dict[str, str] - _temp_dir: tempfile.TemporaryDirectory + socket_dir: Path | None = None + daemon: subprocess.Popen | None = None + daemon_log: Path | None = None + + def start_daemon(self, tracedecay_bin, ready_timeout=DAEMON_READY_TIMEOUT_SECONDS): + """Launches the selected binary's daemon on a socket private to this profile.""" + if self.daemon is not None: + raise EvalSetupError("isolated daemon already started") + # Unix socket paths are length-limited, so keep the socket in a short directory. + self.socket_dir = Path(tempfile.mkdtemp(prefix="td-eval-")) + socket_path = self.socket_dir / "daemon.sock" + self.env["TRACEDECAY_DAEMON_SOCKET"] = str(socket_path) + self.daemon_log = self.root / "daemon.log" + with open(self.daemon_log, "wb") as log: + self.daemon = subprocess.Popen( + [ + tracedecay_bin, + "daemon", + "run", + "--socket", + str(socket_path), + ], + env=self.env, + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + preexec_fn=dies_with_this_process(), + ) + ready = run( + [ + sys.executable, + "-S", + str(PROCESS_HELPER), + "wait-unix-socket", + "--path", + str(socket_path), + "--pid", + str(self.daemon.pid), + "--timeout", + str(ready_timeout), + ], + check=False, + ) + if ready.returncode != 0: + raise EvalSetupError( + f"isolated daemon did not become ready ({ready.returncode}): " + f"{ready.stdout.strip()}\n{self.daemon_log_tail()}" + ) + + def daemon_log_tail(self): + if self.daemon_log is None or not self.daemon_log.exists(): + return "(no daemon output captured)" + return self.daemon_log.read_text(errors="replace")[-4096:] or "(no daemon output captured)" + + def stop_daemon(self): + if self.daemon is None: + return + # Teardown must not depend on launching another Python executable. + # Retain ownership until the canonical group stop and reap complete. + stop_owned_process(self.daemon, PROCESS_STOP_GRACE_SECONDS) + self.daemon = None def cleanup(self): - self._temp_dir.cleanup() + self.stop_daemon() + errors = [] + for path in (self.socket_dir, self.root): + if path is None: + continue + try: + shutil.rmtree(path) + except FileNotFoundError: + pass + except Exception as exc: + errors.append(exc) + if len(errors) == 1: + raise errors[0] + if errors: + raise ExceptionGroup( + "isolated environment cleanup failed: " + + "; ".join(str(error) for error in errors), + errors, + ) + + +class EvalSetupError(RuntimeError): + """An operational failure (daemon, init, install) reported instead of exiting.""" + + +class OwnedProcessCleanupError(EvalSetupError): + """Transfer an unconfirmed command stop to the enclosing resource owner.""" + + def __init__(self, process, original_error, cleanup_error): + self.process = process + self.original_error = original_error + self.cleanup_error = cleanup_error + super().__init__(f"owned process group {process.pid} cleanup failed: {cleanup_error}") + + @property + def cancelled(self): + return isinstance(self.original_error, KeyboardInterrupt) or isinstance( + self.cleanup_error, KeyboardInterrupt + ) + def primary_detail(self): + error = self.original_error if self.original_error is not None else self.cleanup_error + return str(error) or ("interrupted" if self.cancelled else type(error).__name__) -def cleanup_eval_artifacts(fixture, eval_env): - if fixture is not None: - shutil.rmtree(fixture, ignore_errors=True) - eval_env.cleanup() + +class OwnedProfileCleanupError(EvalSetupError): + """Retain allocation failures before an environment can own its profile.""" + + def __init__(self, root, original_error, cleanup_error): + self.root = root + self.original_error = original_error + self.cleanup_error = cleanup_error + super().__init__(f"profile allocation cleanup failed for {root}: {cleanup_error}") def create_eval_environment(scenario_id): safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_id) - temp_dir = tempfile.TemporaryDirectory(prefix=f"tracedecay-eval-env-{safe_id}-") - root = Path(temp_dir.name) - home = root / "home" - data_dir = root / ".tracedecay" - global_db = data_dir / "global.db" - home.mkdir(parents=True, exist_ok=True) - data_dir.mkdir(parents=True, exist_ok=True) - - env = dict(os.environ) - env["HOME"] = str(home) - env["USERPROFILE"] = str(home) - env["XDG_CONFIG_HOME"] = str(home / ".config") - env["TRACEDECAY_DATA_DIR"] = str(data_dir) - env["TRACEDECAY_GLOBAL_DB"] = str(global_db) - env.pop("HERMES_HOME", None) - env.pop("HERMES_PROFILE", None) - return EvalEnvironment( - root=root, - home=home, - data_dir=data_dir, - global_db=global_db, - env=env, - _temp_dir=temp_dir, - ) + # Explicit ownership prevents an implicit temporary-directory finalizer + # from deleting the profile if stopping an owned daemon fails. + root = tempfile.mkdtemp(prefix=f"tracedecay-eval-env-{safe_id}-") + try: + root = Path(root) + home = root / "home" + data_dir = root / ".tracedecay" + global_db = data_dir / "global.db" + home.mkdir(parents=True, exist_ok=True) + data_dir.mkdir(parents=True, exist_ok=True) + + env = dict(os.environ) + env["HOME"] = str(home) + env["USERPROFILE"] = str(home) + env["XDG_CONFIG_HOME"] = str(home / ".config") + env["TRACEDECAY_DATA_DIR"] = str(data_dir) + env["TRACEDECAY_GLOBAL_DB"] = str(global_db) + env.pop("HERMES_HOME", None) + env.pop("HERMES_PROFILE", None) + return EvalEnvironment( + root=root, + home=home, + data_dir=data_dir, + global_db=global_db, + env=env, + ) + except BaseException as original_error: + # Ownership reaches the caller only after the whole environment has + # been constructed; cancellation during setup still belongs here. + try: + shutil.rmtree(root) + except FileNotFoundError: + pass + except BaseException as cleanup_error: + raise OwnedProfileCleanupError(root, original_error, cleanup_error) from original_error + raise def resolve_tracedecay_bin(explicit): @@ -171,15 +297,16 @@ def resolve_tracedecay_bin(explicit): env = dict(os.environ) if explicit: env["TRACEDECAY_BIN"] = str(explicit) - result = subprocess.run( + result = run( [str(REPO_ROOT / "scripts/build-eval-binary.sh")], env=env, - stdout=subprocess.PIPE, - text=True, check=False, + stderr=None, ) if result.returncode != 0: - sys.exit(f"no tracedecay binary: scripts/build-eval-binary.sh exited {result.returncode}") + raise EvalSetupError( + f"no tracedecay binary: scripts/build-eval-binary.sh exited {result.returncode}" + ) return result.stdout.strip() @@ -204,18 +331,45 @@ def load_scenarios(ids): return scenarios -def run(cmd, cwd=None, env=None, timeout=None, check=True): - result = subprocess.run( +def run(cmd, cwd=None, env=None, timeout=None, check=True, stderr=subprocess.STDOUT): + process = subprocess.Popen( cmd, cwd=cwd, env=env, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, + stderr=stderr, text=True, - timeout=timeout, + start_new_session=True, + preexec_fn=dies_with_this_process(), ) - if check and result.returncode != 0: - sys.exit(f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}") + original_error = None + try: + output, captured_stderr = process.communicate(timeout=timeout) + result = subprocess.CompletedProcess( + cmd, process.returncode, stdout=output, stderr=captured_stderr + ) + if check and result.returncode != 0: + raise EvalSetupError( + f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n{result.stdout}" + ) + except BaseException as exc: + original_error = exc + try: + # Popen's context manager waits during an exception without stopping + # the child. Explicit ownership makes cancellation and timeouts stop + # the entire turn, including its MCP children, before fixture cleanup. + stop_owned_process(process, PROCESS_STOP_GRACE_SECONDS) + except BaseException as exc: + # A signal can arrive for the first time during the stop itself. Keep + # the Popen reachable so the caller retries before deleting its files. + raise OwnedProcessCleanupError(process, original_error, exc) from original_error + finally: + if process.stdout is not None: + process.stdout.close() + if process.stderr is not None: + process.stderr.close() + if original_error is not None: + raise original_error return result @@ -254,8 +408,7 @@ def require_fact_record(response, operation): return fact -def build_fixture(scenario, tracedecay_bin, env): - fixture = Path(tempfile.mkdtemp(prefix=f"tracedecay-eval-{scenario['id']}-")) +def seed_fixture(scenario, tracedecay_bin, env, fixture): (fixture / "src").mkdir() (fixture / "src/lib.rs").write_text("pub fn eval_fixture_marker() {}\n") for name, contents in scenario["setup"].get("files", {}).items(): @@ -299,7 +452,6 @@ def build_fixture(scenario, tracedecay_bin, env): f"fixture preload search `{preload_query}` did not return " f"FactId {seeded_fact['fact_id']}" ) - return fixture def provider_env_passthrough(env): @@ -697,6 +849,143 @@ def evaluate_assertions(scenario, tracedecay_bin, fixture, env): return outcomes +def run_scenario(args, scenario, run_dir, report): + """Runs one scenario in an owned profile/daemon/fixture scope. + + Returns (ok, cancelled). Operational failures and cancellation are recorded + in the report. Cleanup stops owned processes before removing their files; + a cleanup failure preserves both its error and the original result. + """ + eval_env = None + fixture = None + entry = error_entry(scenario, "error (setup or agent failure)", "scenario did not finish") + report["scenarios"].append(entry) + ok = False + cancelled = False + pending_command = None + unclaimed_profile = None + cleanup_error = None + + def cleanup(): + nonlocal pending_command + if pending_command is not None: + stop_owned_process(pending_command, PROCESS_STOP_GRACE_SECONDS) + pending_command = None + if eval_env is not None: + eval_env.cleanup() + + try: + eval_env = create_eval_environment(scenario["id"]) + safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario["id"]) + # This scope owns the fixture before any daemon or seeding command can + # use it. Setup errors therefore follow the same stop-before-delete + # order as completed turns and cancellation. + fixture = Path( + tempfile.mkdtemp(prefix=f"tracedecay-eval-{safe_id}-", dir=eval_env.root) + ) + eval_env.start_daemon(args.tracedecay_bin) + seed_fixture(scenario, args.tracedecay_bin, eval_env.env, fixture) + if args.driver == "hermes": + transcripts = drive_hermes(args, scenario, fixture, run_dir, eval_env.env) + else: + transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env) + outcomes = evaluate_assertions(scenario, args.tracedecay_bin, fixture, eval_env.env) + failed = [o for o in outcomes if not o["passed"]] + status = "pass" if not failed else "fail" + if failed and scenario.get("contract") == "pending-sibling": + status = "fail (note: scenario contract is pending-sibling, see contract_notes)" + if not all(t["turn_valid"] for t in transcripts): + status = "error (agent turn invalid, see transcript logs)" + failed = failed or [{"name": "agent-turn", "passed": False}] + entry.pop("error") + entry.update(status=status, assertions=outcomes, transcripts=transcripts) + ok = not failed + except OwnedProfileCleanupError as exc: + unclaimed_profile = exc.root + cleanup_error = exc.cleanup_error + cancelled = isinstance(exc.original_error, KeyboardInterrupt) + entry["error"] = str(exc.original_error) or ( + "interrupted" if cancelled else type(exc.original_error).__name__ + ) + except OwnedProcessCleanupError as exc: + pending_command = exc.process + cancelled = exc.cancelled + if cancelled: + entry["status"] = "cancelled" + entry["error"] = exc.primary_detail() + except KeyboardInterrupt as exc: + entry.update(status="cancelled", error=str(exc) or "interrupted") + cancelled = True + except Exception as exc: + entry["error"] = str(exc) + finally: + try: + try: + cleanup() + except KeyboardInterrupt as exc: + # The first signal may arrive during teardown. main keeps + # subsequent signals from interrupting this final cleanup. + cancelled = True + ok = False + entry["status"] = "cancelled" + entry.setdefault("error", str(exc) or "interrupted") + cleanup() + except BaseException as exc: + cleanup_error = exc + if cleanup_error is not None: + ok = False + cancelled |= isinstance(cleanup_error, KeyboardInterrupt) + entry["cleanup_error"] = str(cleanup_error) or type(cleanup_error).__name__ + if pending_command is not None: + entry["command_pid"] = pending_command.pid + for label, path in ( + ("retained_fixture", fixture), + ("retained_profile", eval_env.root if eval_env is not None else unclaimed_profile), + ("retained_socket_dir", eval_env.socket_dir if eval_env is not None else None), + ): + if path is None: + continue + try: + os.stat(path) + except FileNotFoundError: + continue + except OSError: + # An inaccessible owned path may still exist. Preserve + # its location without replacing the cleanup failure. + pass + entry[label] = str(path) + if eval_env is not None and eval_env.daemon is not None: + entry["daemon_pid"] = eval_env.daemon.pid + entry["status"] = "cancelled" if cancelled else "error (cleanup failure)" + + print(f"[{scenario['id']}] {entry['status']}") + if entry.get("error"): + print(f" {entry['error']}", file=sys.stderr) + if entry.get("cleanup_error"): + print(f" cleanup: {entry['cleanup_error']}", file=sys.stderr) + for outcome in entry["assertions"]: + marker = "pass" if outcome["passed"] else "FAIL" + if outcome.get("error"): + print(f" [{marker}] {outcome['name']}, {outcome['error']}") + else: + print( + f" [{marker}] {outcome['name']}, actual {outcome['actual']} " + f"{outcome['op']} expected {outcome['expected']}" + ) + return ok, cancelled + + +def error_entry(scenario, status, error): + return { + "id": scenario["id"], + "contract": scenario.get("contract", "stable"), + "status": status, + "error": error, + "assertions": [], + "transcripts": [], + } + + def main(argv): args = parse_args(argv) scenarios = load_scenarios(args.scenario) @@ -728,55 +1017,72 @@ def main(argv): print(f"\nblocked report written to {report_path}", file=sys.stderr) return 2 - args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin) - report["tracedecay_bin"] = args.tracedecay_bin + cancel_requested = False + def raise_cancelled(signum, _frame): + nonlocal cancel_requested + if not cancel_requested: + cancel_requested = True + raise KeyboardInterrupt(f"cancelled by signal {signum}") + previous_handlers = { + sig: signal.signal(sig, raise_cancelled) + for sig in (signal.SIGINT, signal.SIGTERM) + } overall_ok = True - for scenario in scenarios: - eval_env = create_eval_environment(scenario["id"]) - fixture = None + cancelled = False + phase = "binary resolution" + report_path = run_dir / "report.json" + try: try: - fixture = build_fixture(scenario, args.tracedecay_bin, eval_env.env) - if args.driver == "hermes": - transcripts = drive_hermes(args, scenario, fixture, run_dir, eval_env.env) - else: - transcripts = drive_cursor_agent(args, scenario, fixture, run_dir, eval_env.env) - outcomes = evaluate_assertions( - scenario, args.tracedecay_bin, fixture, eval_env.env - ) - failed = [o for o in outcomes if not o["passed"]] - status = "pass" if not failed else "fail" - if failed and scenario.get("contract") == "pending-sibling": - status = "fail (note: scenario contract is pending-sibling, see contract_notes)" - if not all(t["turn_valid"] for t in transcripts): - status = "error (agent turn invalid, see transcript logs)" - failed = failed or [{"name": "agent-turn", "passed": False}] - report["scenarios"].append( - { - "id": scenario["id"], - "contract": scenario.get("contract", "stable"), - "status": status, - "assertions": outcomes, - "transcripts": transcripts, - } - ) - overall_ok &= not failed - print(f"[{scenario['id']}] {status}") - for outcome in outcomes: - marker = "pass" if outcome["passed"] else "FAIL" - if outcome.get("error"): - print(f" [{marker}] {outcome['name']}, {outcome['error']}") - else: - print( - f" [{marker}] {outcome['name']}, actual {outcome['actual']} " - f"{outcome['op']} expected {outcome['expected']}" - ) - finally: - cleanup_eval_artifacts(fixture, eval_env) + args.tracedecay_bin = resolve_tracedecay_bin(args.tracedecay_bin) + report["tracedecay_bin"] = args.tracedecay_bin + phase = "scenario execution" + for scenario in scenarios: + ok, cancelled = run_scenario(args, scenario, run_dir, report) + overall_ok &= ok + cancelled |= cancel_requested + if cancelled: + break + except OwnedProcessCleanupError as exc: + overall_ok = False + cancelled = exc.cancelled + report["reason"] = exc.primary_detail() + report["phase"] = phase + # Binary resolution runs before any environment exists, so main + # is the owner of a command whose initial stop did not complete. + try: + try: + stop_owned_process(exc.process, PROCESS_STOP_GRACE_SECONDS) + except KeyboardInterrupt: + cancelled = True + stop_owned_process(exc.process, PROCESS_STOP_GRACE_SECONDS) + except BaseException as cleanup_error: + cancelled |= isinstance(cleanup_error, KeyboardInterrupt) + report["cleanup_error"] = str(cleanup_error) or type(cleanup_error).__name__ + report["command_pid"] = exc.process.pid + except Exception as exc: + overall_ok = False + report["reason"] = str(exc) + report["phase"] = phase + + cancelled |= cancel_requested + report["status"] = "cancelled" if cancelled else ("pass" if overall_ok else "fail") + phase = "report publication" + report_path.write_text(json.dumps(report, indent=2) + "\n") + except KeyboardInterrupt as exc: + # This scope covers resolution, gaps between scenarios, and report + # publication. Finish the report before restoring signal handlers. + cancelled = True + report.update(status="cancelled", reason=str(exc) or "interrupted", phase=phase) + report_path.write_text(json.dumps(report, indent=2) + "\n") + finally: + for sig, handler in previous_handlers.items(): + signal.signal(sig, handler) + + if cancelled: + print(f"\ncancelled; report written to {report_path}", file=sys.stderr) + return 130 - report["status"] = "pass" if overall_ok else "fail" - report_path = run_dir / "report.json" - report_path.write_text(json.dumps(report, indent=2) + "\n") print(f"\nreport written to {report_path}") return 0 if overall_ok else 1 diff --git a/evals/memory/test_run_real_model.py b/evals/memory/test_run_real_model.py index 179e2383fb..da0ced847f 100644 --- a/evals/memory/test_run_real_model.py +++ b/evals/memory/test_run_real_model.py @@ -1,10 +1,16 @@ import argparse import contextlib +import errno import importlib.util import io import json +import os +import signal import subprocess +import sys import tempfile +import threading +import time import unittest from unittest import mock from pathlib import Path @@ -15,6 +21,8 @@ run_real_model = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(run_real_model) +import portable_process + BENCHMARK_PATH = MODULE_PATH.parent.parent.parent / "benchmark_data" / "run_benchmarks.py" BENCHMARK_SPEC = importlib.util.spec_from_file_location("run_benchmarks", BENCHMARK_PATH) run_benchmarks = importlib.util.module_from_spec(BENCHMARK_SPEC) @@ -107,7 +115,7 @@ def test_explicit_binary_is_used_and_must_exist(self): self.assertEqual( run_real_model.resolve_tracedecay_bin(binary), str(binary) ) - with self.assertRaises(SystemExit): + with self.assertRaises(run_real_model.EvalSetupError): run_real_model.resolve_tracedecay_bin(Path(tmp) / "missing") def test_default_never_falls_back_to_a_prebuilt_or_path_binary(self): @@ -118,12 +126,29 @@ def test_default_never_falls_back_to_a_prebuilt_or_path_binary(self): path_binary.chmod(0o755) with ( mock.patch.dict("os.environ", {"PATH": tmp}), - mock.patch.object(run_real_model.subprocess, "run", return_value=failed), - self.assertRaises(SystemExit), + mock.patch.object(run_real_model, "run", return_value=failed), + self.assertRaises(run_real_model.EvalSetupError), ): run_real_model.resolve_tracedecay_bin(None) + def test_build_failure_writes_a_failed_report_before_any_scenario_starts(self): + with tempfile.TemporaryDirectory() as tmp: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object(run_real_model, "load_scenarios", return_value=[{"id": "test"}]), + mock.patch.object(run_real_model, "run", return_value=subprocess.CompletedProcess([], 2, stdout="")), + mock.patch.object(run_real_model, "run_scenario") as scenario, + contextlib.redirect_stdout(io.StringIO()), + ): + self.assertEqual(run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 1) + report = json.loads(next(Path(tmp).glob("*/report.json")).read_text()) + scenario.assert_not_called() + self.assertEqual(report["status"], "fail") + self.assertIn("build-eval-binary.sh exited 2", report["reason"]) + self.assertIsNone(report["tracedecay_bin"]) + + class CursorComposerSelectionTest(unittest.TestCase): def test_cursor_driver_loads_the_installed_plugin_in_composer(self): env = run_real_model.create_eval_environment("cursor-composer") @@ -166,16 +191,1053 @@ def test_eval_environment_pins_home_and_profile_storage_to_tempdir(self): self.assertTrue(env.data_dir.is_relative_to(env.root)) self.assertNotEqual(env.global_db, Path.home() / ".tracedecay/global.db") + def test_cancellation_before_ownership_removes_the_profile(self): + for boundary in ("environment copy", "owner construction"): + with self.subTest(boundary=boundary), tempfile.TemporaryDirectory() as tmp: + cancelled = KeyboardInterrupt(f"cancelled during {boundary}") + interrupt = ( + mock.patch.object(os.environ, "keys", side_effect=cancelled) + if boundary == "environment copy" + else mock.patch.object( + run_real_model, "EvalEnvironment", side_effect=cancelled + ) + ) + with mock.patch.object(tempfile, "tempdir", tmp), interrupt as abort: + with self.assertRaises(KeyboardInterrupt) as raised: + run_real_model.create_eval_environment("cancelled-setup") + self.assertIs(raised.exception, cancelled) + abort.assert_called_once() + self.assertEqual(list(Path(tmp).iterdir()), []) + def test_cleanup_removes_fixture_and_isolated_eval_store(self): + eval_env = run_real_model.create_eval_environment("cleanup-store") + self.addCleanup(eval_env.cleanup) + fixture = eval_env.root / "fixture" + fixture.mkdir() + + eval_env.cleanup() + + self.assertFalse(fixture.exists()) + self.assertFalse(eval_env.root.exists()) + + +FAKE_BIN = """#!/usr/bin/env python3 +import os, socket, sys, time +args = sys.argv[1:] +if args[:2] == ["daemon", "run"]: + if os.environ.get("FAKE_DAEMON_FAIL"): + sys.exit(3) + path = args[3] + srv = socket.socket(socket.AF_UNIX) + srv.bind(path) + srv.listen(8) + open(os.environ["TRACEDECAY_DATA_DIR"] + "/daemon.pid", "w").write(str(os.getpid())) + time.sleep(600) +elif args[:1] == ["init"]: + if os.environ.get("FAKE_INIT_FAIL"): + print("code_index_scheduler_unavailable") + sys.exit(1) + c = socket.socket(socket.AF_UNIX) + c.connect(os.environ["TRACEDECAY_DAEMON_SOCKET"]) +elif args[:1] == ["tool"]: + print('{"fact": {"fact_id": "fact.1"}}') +""" + + +class IsolatedLifecycleTest(unittest.TestCase): + def make_bin(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + path = Path(tmp.name) / "tracedecay" + path.write_text(FAKE_BIN) + path.chmod(0o755) + return str(path) + + def scenario(self): + return { + "id": "lifecycle", + "setup": { + "facts": [ + {"content": "c", "category": "project", "source": "s", "trust": 0.5} + ] + }, + } + + def test_profile_allocation_failure_writes_error_report(self): with tempfile.TemporaryDirectory() as tmp: - fixture = Path(tmp) / "fixture" + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object( + run_real_model, "load_scenarios", return_value=[self.scenario()] + ), + mock.patch.object( + run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay" + ), + mock.patch.object( + run_real_model.tempfile, "mkdtemp", + side_effect=OSError("No space left on device"), + ), + mock.patch.object(run_real_model.EvalEnvironment, "start_daemon") as daemon, + mock.patch.object(run_real_model, "drive_hermes") as hermes, + mock.patch.object(run_real_model, "drive_cursor_agent") as cursor, + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual( + run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 1 + ) + daemon.assert_not_called() + hermes.assert_not_called() + cursor.assert_not_called() + reports = list(Path(tmp).glob("*/report.json")) + self.assertEqual(len(reports), 1) + report = json.loads(reports[0].read_text()) + self.assertEqual(report["status"], "fail") + self.assertEqual(report["scenarios"], [{ + "id": "lifecycle", + "contract": "stable", + "status": "error (setup or agent failure)", + "error": "No space left on device", + "assertions": [], + "transcripts": [], + }]) + + def test_environment_construction_cancellation_removes_unowned_profile(self): + original_mkdtemp = tempfile.mkdtemp + for operation in ("dict", "EvalEnvironment"): + with self.subTest(operation=operation), tempfile.TemporaryDirectory() as tmp: + allocated = [] + + def allocate(*args, **kwargs): + root = Path(original_mkdtemp(*args, **kwargs)) + allocated.append(root) + self.addCleanup(run_real_model.shutil.rmtree, root, ignore_errors=True) + return str(root) + + cancellation = KeyboardInterrupt(f"cancelled during {operation}") + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object( + run_real_model, "load_scenarios", return_value=[ + self.scenario(), {**self.scenario(), "id": "must-not-start"} + ] + ), + mock.patch.object( + run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay" + ), + mock.patch.object(run_real_model.tempfile, "mkdtemp", side_effect=allocate), + mock.patch.object(run_real_model, operation, side_effect=cancellation, create=True), + mock.patch.object(run_real_model.subprocess, "Popen") as spawn, + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual( + run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130 + ) + spawn.assert_not_called() + reports = list(Path(tmp).glob("*/report.json")) + self.assertEqual(len(reports), 1) + report = json.loads(reports[0].read_text()) + self.assertEqual(report["status"], "cancelled") + self.assertEqual(report["scenarios"], [{ + "id": "lifecycle", + "contract": "stable", + "status": "cancelled", + "error": str(cancellation), + "assertions": [], + "transcripts": [], + }]) + self.assertEqual(len(allocated), 1) + self.assertFalse(allocated[0].exists(), "unclaimed profile must be removed") + + def test_allocation_cleanup_failure_preserves_cancellation_and_inaccessible_profile(self): + original_mkdtemp = tempfile.mkdtemp + original_remove = run_real_model.shutil.rmtree + original_stat = os.stat + for operation in ("Path", "EvalEnvironment"): + with self.subTest(operation=operation), tempfile.TemporaryDirectory() as tmp: + allocated = [] + + def allocate(*args, **kwargs): + root = Path(original_mkdtemp(*args, **kwargs)) + allocated.append(root) + self.addCleanup(original_remove, root, ignore_errors=True) + return str(root) + + def remove(path, *args, **kwargs): + if allocated and Path(path) == allocated[0]: + raise PermissionError("profile removal denied") + return original_remove(path, *args, **kwargs) + + def stat(path, *args, **kwargs): + if ( + allocated + and isinstance(path, (str, os.PathLike)) + and Path(path) == allocated[0] + ): + raise PermissionError("profile stat denied") + return original_stat(path, *args, **kwargs) + + cancellation = KeyboardInterrupt(f"cancelled during {operation}") + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object( + run_real_model, "load_scenarios", return_value=[ + self.scenario(), {**self.scenario(), "id": "must-not-start"} + ] + ), + mock.patch.object( + run_real_model, "resolve_tracedecay_bin", return_value="/unused/tracedecay" + ), + mock.patch.object(run_real_model.tempfile, "mkdtemp", side_effect=allocate), + mock.patch.object(run_real_model, operation, side_effect=cancellation), + mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove), + mock.patch.object(os, "stat", side_effect=stat), + mock.patch.object(run_real_model.subprocess, "Popen") as spawn, + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual( + run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130 + ) + spawn.assert_not_called() + self.assertEqual(len(allocated), 1) + self.assertTrue(allocated[0].exists()) + reports = list(Path(tmp).glob("*/report.json")) + self.assertEqual(len(reports), 1) + report = json.loads(reports[0].read_text()) + self.assertEqual(report["status"], "cancelled") + self.assertEqual(report["scenarios"], [{ + "id": "lifecycle", + "contract": "stable", + "status": "cancelled", + "error": str(cancellation), + "cleanup_error": "profile removal denied", + "retained_profile": str(allocated[0]), + "assertions": [], + "transcripts": [], + }]) + + def test_fresh_environment_inits_and_seeds_against_isolated_daemon(self): + binary = self.make_bin() + eval_env = run_real_model.create_eval_environment("lifecycle") + try: + fixture = eval_env.root / "fixture" fixture.mkdir() - eval_env = run_real_model.create_eval_environment("cleanup-store") + eval_env.start_daemon(binary, ready_timeout=15) + run_real_model.seed_fixture(self.scenario(), binary, eval_env.env, fixture) + self.assertTrue(fixture.is_relative_to(eval_env.root)) + pid = int((eval_env.data_dir / "daemon.pid").read_text()) + finally: + eval_env.cleanup() + self.assertFalse(eval_env.root.exists()) + self.assertFalse(Path(eval_env.env["TRACEDECAY_DAEMON_SOCKET"]).parent.exists()) + with self.assertRaises(ProcessLookupError): + os.kill(pid, 0) + + def test_init_failure_stops_daemon_before_removing_fixture(self): + binary = self.make_bin() + eval_env = run_real_model.create_eval_environment("lifecycle") + self.addCleanup(eval_env.cleanup) + eval_env.env["FAKE_INIT_FAIL"] = "1" + args = argparse.Namespace(tracedecay_bin=binary, driver="hermes") + report = {"scenarios": []} + removed = [] + original_remove = run_real_model.shutil.rmtree + + def remove(path, *args, **kwargs): + path = Path(path) + if path.is_relative_to(eval_env.root) and path.exists(): + self.assertIsNone(eval_env.daemon, "owned work must stop before deleting inputs") + pid = int((eval_env.data_dir / "daemon.pid").read_text()) + with self.assertRaises(ProcessLookupError): + os.kill(pid, 0) + removed.append(path) + return original_remove(path, *args, **kwargs) + + with ( + tempfile.TemporaryDirectory() as run_dir, + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove), + mock.patch.object(run_real_model, "drive_hermes") as driver, + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + result = run_real_model.run_scenario( + args, self.scenario(), Path(run_dir), report + ) + self.assertEqual(result, (False, False)) + driver.assert_not_called() + self.assertTrue(removed) + self.assertFalse(eval_env.root.exists()) + self.assertIn("command failed (1)", report["scenarios"][0]["error"]) + self.assertNotIn("cleanup_error", report["scenarios"][0]) + + def test_daemon_startup_failure_is_reported_and_cleaned(self): + binary = self.make_bin() + eval_env = run_real_model.create_eval_environment("lifecycle") + eval_env.env["FAKE_DAEMON_FAIL"] = "1" + try: + with self.assertRaises(run_real_model.EvalSetupError): + eval_env.start_daemon(binary, ready_timeout=15) + finally: + eval_env.cleanup() + self.assertFalse(eval_env.root.exists()) + + def run_scenario_with(self, binary, driver): + args = argparse.Namespace(tracedecay_bin=binary, driver="hermes") + report = {"scenarios": []} + with tempfile.TemporaryDirectory() as run_dir, mock.patch.object( + run_real_model, "drive_hermes", side_effect=driver + ): + result = run_real_model.run_scenario( + args, self.scenario(), Path(run_dir), report + ) + return result, report + + def test_agent_failure_and_cancellation_write_error_entries(self): + binary = self.make_bin() + for exc, status, cancelled in ( + (RuntimeError("agent exploded"), "error", False), + (KeyboardInterrupt("stop"), "cancelled", True), + ): + with self.subTest(status=status): + result, report = self.run_scenario_with(binary, exc) + self.assertEqual(result, (False, cancelled)) + entry = report["scenarios"][0] + self.assertTrue(entry["status"].startswith(status)) + self.assertEqual( + entry["error"], str(exc) or "interrupted" + ) + + def test_daemon_cleanup_does_not_need_to_launch_a_helper(self): + eval_env = run_real_model.create_eval_environment("cleanup-no-helper") + self.addCleanup(eval_env.cleanup) + eval_env.start_daemon(self.make_bin(), ready_timeout=15) + daemon = eval_env.daemon + with mock.patch.object( + run_real_model.subprocess, "Popen", side_effect=OSError("cannot spawn helper") + ): + eval_env.cleanup() + self.assertIsNotNone(daemon.returncode) + self.assertIsNone(eval_env.daemon) + self.assertFalse(eval_env.root.exists()) + + def test_cleanup_failure_retains_original_error_when_artifact_stat_is_denied(self): + eval_env = run_real_model.create_eval_environment("cleanup-error") + self.addCleanup(eval_env.cleanup) + args = argparse.Namespace(tracedecay_bin=self.make_bin(), driver="hermes") + report = {"scenarios": []} + # Start before injecting stop failure so setup still uses real commands. + eval_env.start_daemon(args.tracedecay_bin, ready_timeout=15) + daemon = eval_env.daemon + original_stat = os.stat + inaccessible = [] + + def stat(path, *args, **kwargs): + if isinstance(path, (str, os.PathLike)): + artifact = Path(path) + if ( + artifact == eval_env.root + or artifact == eval_env.socket_dir + or artifact.parent == eval_env.root + ): + inaccessible.append(artifact) + raise PermissionError(errno.EACCES, "owned artifact stat denied", str(path)) + return original_stat(path, *args, **kwargs) + + with ( + tempfile.TemporaryDirectory() as run_dir, + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(eval_env, "start_daemon"), + mock.patch.object(run_real_model, "seed_fixture"), + mock.patch.object(run_real_model, "drive_hermes", side_effect=RuntimeError("agent failed")), + mock.patch.object(run_real_model, "stop_owned_process", side_effect=OSError("stop failed")), + mock.patch.object(os, "stat", side_effect=stat), + ): + result = run_real_model.run_scenario(args, self.scenario(), Path(run_dir), report) + self.assertEqual(result, (False, False)) + self.assertEqual(report["scenarios"][0]["error"], "agent failed") + self.assertEqual(report["scenarios"][0]["cleanup_error"], "stop failed") + entry = report["scenarios"][0] + fixture = Path(entry["retained_fixture"]) + self.assertEqual(entry["retained_profile"], str(eval_env.root)) + self.assertEqual(entry["retained_socket_dir"], str(eval_env.socket_dir)) + self.assertEqual(set(inaccessible), {fixture, eval_env.root, eval_env.socket_dir}) + self.assertTrue(fixture.exists()) + self.assertIs(eval_env.daemon, daemon) + self.assertIsNone(daemon.poll()) + self.assertTrue(eval_env.root.exists()) + eval_env.cleanup() + self.assertFalse(eval_env.root.exists()) + + def test_socket_cleanup_failure_reports_retained_directory_and_cleans_profile(self): + eval_env = run_real_model.create_eval_environment("socket-cleanup-error") + self.addCleanup(eval_env.cleanup) + binary = self.make_bin() + original_remove = run_real_model.shutil.rmtree + original_unlink = os.unlink + removed = [] + fixture = None + daemon = None + + def drive(_args, _scenario, path, _run_dir, _env): + nonlocal fixture, daemon + fixture = path + daemon = eval_env.daemon + return [{"turn_valid": True}] + + def remove(path, *args, **kwargs): + path = Path(path) + if path in (eval_env.socket_dir, eval_env.root): + self.assertIsNone(eval_env.daemon) + self.assertIsNotNone(daemon.returncode) + removed.append(path) + return original_remove(path, *args, **kwargs) + + def unlink(path, *args, **kwargs): + if Path(path).name == "daemon.sock": + raise PermissionError(errno.EACCES, "socket directory deletion failed", path) + return original_unlink(path, *args, **kwargs) + + with tempfile.TemporaryDirectory() as tmp: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=binary), + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(run_real_model, "drive_hermes", side_effect=drive), + mock.patch.object(run_real_model, "evaluate_assertions", return_value=[]), + mock.patch.object(run_real_model.shutil, "rmtree", side_effect=remove), + mock.patch.object(os, "unlink", side_effect=unlink), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"]) + report_path, = Path(tmp).glob("*/report.json") + report = json.loads(report_path.read_text()) + + self.assertEqual(code, 1) + self.assertEqual(report["status"], "fail") + self.assertEqual(removed, [eval_env.socket_dir, eval_env.root]) + self.assertFalse(eval_env.root.exists()) + self.assertFalse(fixture.exists()) + self.assertTrue(eval_env.socket_dir.exists()) + entry = report["scenarios"][0] + self.assertEqual(entry["status"], "error (cleanup failure)") + self.assertIn("socket directory deletion failed", entry["cleanup_error"]) + self.assertEqual(entry["retained_socket_dir"], str(eval_env.socket_dir)) + self.assertNotIn("retained_profile", entry) + self.assertNotIn("retained_fixture", entry) + self.assertNotIn("daemon_pid", entry) + eval_env.cleanup() + self.assertFalse(eval_env.socket_dir.exists()) + + def test_seed_failure_retains_fixture_until_daemon_stop_is_confirmed(self): + binary = self.make_bin() + eval_env = run_real_model.create_eval_environment("seed-cleanup-error") + self.addCleanup(eval_env.cleanup) + args = argparse.Namespace(tracedecay_bin=binary, driver="hermes") + scenario = self.scenario() + # Initialization and the fact write succeed before preload validation + # fails, so the daemon has already used the fixture being retained. + scenario["setup"]["facts"][0]["preload_searches"] = 1 + report = {"scenarios": []} + fixture = None + original_seed = run_real_model.seed_fixture + original_stop = run_real_model.stop_owned_process + + def seed(scenario, binary, env, path): + nonlocal fixture + fixture = path + return original_seed(scenario, binary, env, path) - run_real_model.cleanup_eval_artifacts(fixture, eval_env) + def stop(process, grace): + if process is eval_env.daemon: + raise OSError("daemon stop failed") + return original_stop(process, grace) + + with ( + tempfile.TemporaryDirectory() as run_dir, + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(run_real_model, "seed_fixture", side_effect=seed), + mock.patch.object(run_real_model, "stop_owned_process", side_effect=stop), + mock.patch.object(run_real_model, "drive_hermes") as driver, + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + result = run_real_model.run_scenario(args, scenario, Path(run_dir), report) + + self.assertEqual(result, (False, False)) + driver.assert_not_called() + entry = report["scenarios"][0] + self.assertIn("sets preload_searches without preload_query", entry["error"]) + self.assertEqual(entry["cleanup_error"], "daemon stop failed") + self.assertIsNotNone(fixture) + self.assertTrue(fixture.exists(), "a live daemon must retain its seeded fixture") + self.assertEqual(entry["retained_fixture"], str(fixture)) + self.assertEqual(entry["retained_profile"], str(eval_env.root)) + self.assertEqual(entry["daemon_pid"], eval_env.daemon.pid) + self.assertIsNone(eval_env.daemon.poll()) + self.assertEqual( + (fixture / "src/lib.rs").read_text(), "pub fn eval_fixture_marker() {}\n" + ) + eval_env.cleanup() + self.assertFalse(eval_env.root.exists()) + self.assertFalse(fixture.exists()) + + def test_signals_stop_agent_descendants_and_persist_cancelled_report(self): + # A real child and grandchild that ignore SIGTERM exercise forced + # process-group teardown rather than a mocked KeyboardInterrupt. + script = """ +import json, os, pathlib, signal, subprocess, sys, time +signal.signal(signal.SIGTERM, signal.SIG_IGN) +ready = pathlib.Path(sys.argv[1] + '.ready') +child = subprocess.Popen([sys.executable, '-c', + 'import pathlib,signal,sys,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); ' + 'pathlib.Path(sys.argv[1]).touch(); time.sleep(600)', str(ready)]) +while not ready.exists(): time.sleep(0.01) +pathlib.Path(sys.argv[1]).write_text(json.dumps([os.getpid(), child.pid])) +time.sleep(600) +""" + for sig in (signal.SIGINT, signal.SIGTERM): + with self.subTest(signal=sig), tempfile.TemporaryDirectory() as tmp: + pidfile = Path(tmp) / "agent-pids.json" + eval_env = run_real_model.create_eval_environment("signal-cleanup") + self.addCleanup(eval_env.cleanup) + notified = threading.Event() + + def interrupt_agent(): + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if pidfile.exists() and pidfile.stat().st_size: + notified.set() + os.kill(os.getpid(), sig) + return + time.sleep(0.01) + + def drive(*_args): + run_real_model.run([sys.executable, "-c", script, str(pidfile)], timeout=20) + self.fail("the interrupted agent must not complete") + + thread = threading.Thread(target=interrupt_agent, daemon=True) + previous_handlers = {s: signal.getsignal(s) for s in (signal.SIGINT, signal.SIGTERM)} + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"), + mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=self.make_bin()), + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(run_real_model, "drive_hermes", side_effect=drive), + mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + thread.start() + code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"]) + thread.join(timeout=1) + self.assertTrue(notified.is_set()) + self.assertEqual(code, 130) + report_path, = (Path(tmp) / "runs").glob("*/report.json") + report = json.loads(report_path.read_text()) + self.assertEqual(report["status"], "cancelled") + self.assertEqual(report["scenarios"][0]["status"], "cancelled") + leader, child = json.loads(pidfile.read_text()) + with self.assertRaises(ProcessLookupError): + os.kill(leader, 0) + # Linux can retain an orphaned zombie until PID 1 reaps it; + # use the canonical runnable-member check for the descendant. + self.assertFalse(portable_process._group_alive(leader), child) + self.assertIsNone(eval_env.daemon) + self.assertFalse(eval_env.root.exists()) + self.assertEqual(previous_handlers, { + s: signal.getsignal(s) for s in (signal.SIGINT, signal.SIGTERM) + }) + + def test_cancelled_command_stop_failure_retains_its_files_and_primary_error(self): + script = """ +import os, pathlib, signal, sys, time +signal.signal(signal.SIGTERM, signal.SIG_IGN) +pathlib.Path(sys.argv[1]).write_text(str(os.getpid())) +time.sleep(30) +""" + # Exercise main before scenario allocation, fixture seeding, and an + # established agent turn with their respective process owners. + for phase in ("binary resolution", "fixture setup", "agent turn"): + with self.subTest(phase=phase), tempfile.TemporaryDirectory() as tmp: + pidfile = Path(tmp) / "command.pid" + eval_env = ( + None if phase == "binary resolution" + else run_real_model.create_eval_environment("command-stop-failure") + ) + fixture = None + command = None + stop_attempts = 0 + notified = threading.Event() + finished = threading.Event() + original_stop = run_real_model.stop_owned_process + original_seed = run_real_model.seed_fixture - self.assertFalse(fixture.exists()) - self.assertFalse(eval_env.root.exists()) + def interrupt_command(): + deadline = time.monotonic() + 15 + while time.monotonic() < deadline and not finished.wait(0.01): + if pidfile.exists() and pidfile.stat().st_size: + notified.set() + os.kill(os.getpid(), signal.SIGTERM) + return + + def run_command(): + run_real_model.run( + [sys.executable, "-c", script, str(pidfile)], + cwd=fixture, + timeout=20, + ) + self.fail("the interrupted command must not complete") + + def seed(scenario, binary, env, path): + nonlocal fixture + fixture = path + if phase == "fixture setup": + (path / "owned-input").write_text("must survive a failed stop") + run_command() + else: + original_seed(scenario, binary, env, path) + + def fail_command_stop(process, grace): + nonlocal command, stop_attempts + if str(pidfile) in process.args: + command = process + stop_attempts += 1 + raise OSError("command stop failed") + return original_stop(process, grace) + + thread = threading.Thread(target=interrupt_command, daemon=True) + try: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"), + mock.patch.object(run_real_model, "load_scenarios", return_value=[ + self.scenario(), {"id": "must-not-run-after-cancellation"}, + ]), + mock.patch.object( + run_real_model, "resolve_tracedecay_bin", + side_effect=( + (lambda _explicit: run_command()) if phase == "binary resolution" + else (lambda _explicit: self.make_bin()) + ), + ), + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env) as create_env, + mock.patch.object(run_real_model, "seed_fixture", side_effect=seed), + mock.patch.object(run_real_model, "drive_hermes", side_effect=lambda *_args: run_command()), + mock.patch.object(run_real_model, "stop_owned_process", side_effect=fail_command_stop), + mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + thread.start() + code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"]) + finished.set() + thread.join(timeout=1) + self.assertTrue(notified.is_set()) + self.assertEqual(code, 130) + self.assertEqual(stop_attempts, 2, "the enclosing owner must retry the command stop") + self.assertIsNotNone(command) + self.assertIsNone(command.poll()) + report_path, = (Path(tmp) / "runs").glob("*/report.json") + report = json.loads(report_path.read_text()) + self.assertEqual(report["status"], "cancelled") + if phase == "binary resolution": + create_env.assert_not_called() + self.assertEqual(report["scenarios"], []) + self.assertEqual(report["phase"], phase) + entry = report + primary_error = entry["reason"] + else: + self.assertEqual(len(report["scenarios"]), 1) + entry = report["scenarios"][0] + self.assertEqual(entry["status"], "cancelled") + primary_error = entry["error"] + self.assertEqual(entry["retained_profile"], str(eval_env.root)) + self.assertEqual(entry["retained_fixture"], str(fixture)) + self.assertEqual(entry["daemon_pid"], eval_env.daemon.pid) + self.assertTrue(eval_env.root.exists()) + self.assertTrue(fixture.exists()) + self.assertIsNone(eval_env.daemon.poll()) + if phase == "fixture setup": + self.assertEqual( + (fixture / "owned-input").read_text(), + "must survive a failed stop", + ) + self.assertIn("cancelled by signal", primary_error) + self.assertEqual(entry["cleanup_error"], "command stop failed") + self.assertEqual(entry["command_pid"], command.pid) + finally: + finished.set() + thread.join(timeout=1) + if command is not None: + original_stop(command, 0.2) + if eval_env is not None: + eval_env.cleanup() + + def test_first_signal_during_command_stop_finishes_owned_group_cleanup(self): + script = """ +import json, os, pathlib, subprocess, sys, time +ready = pathlib.Path(sys.argv[1] + '.ready') +child = subprocess.Popen([sys.executable, '-c', + 'import pathlib,signal,sys,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); ' + 'pathlib.Path(sys.argv[1]).touch(); time.sleep(30)', str(ready)], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) +while not ready.exists(): time.sleep(0.01) +pathlib.Path(sys.argv[1]).write_text(json.dumps([os.getpid(), child.pid])) +""" + with tempfile.TemporaryDirectory() as tmp: + pidfile = Path(tmp) / "agent-pids.json" + eval_env = run_real_model.create_eval_environment("signal-during-stop") + self.addCleanup(eval_env.cleanup) + interrupted = False + original_signal_group = portable_process._signal_group + + def interrupt_stop(pid, sig): + nonlocal interrupted + original_signal_group(pid, sig) + if not interrupted and sig == signal.SIGTERM and pidfile.exists(): + leader, _child = json.loads(pidfile.read_text()) + if pid == leader: + interrupted = True + os.kill(os.getpid(), signal.SIGTERM) + + def drive(*_args): + run_real_model.run([sys.executable, "-c", script, str(pidfile)], timeout=20) + self.fail("the interrupted command stop must cancel the scenario") + + try: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp) / "runs"), + mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=self.make_bin()), + mock.patch.object(run_real_model, "create_eval_environment", return_value=eval_env), + mock.patch.object(run_real_model, "drive_hermes", side_effect=drive), + mock.patch.object(portable_process, "_signal_group", side_effect=interrupt_stop), + mock.patch.object(run_real_model, "PROCESS_STOP_GRACE_SECONDS", 0.2), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"]) + self.assertTrue(interrupted) + self.assertEqual(code, 130) + leader, child = json.loads(pidfile.read_text()) + self.assertFalse(portable_process._group_alive(leader), child) + self.assertIsNone(eval_env.daemon) + self.assertFalse(eval_env.root.exists()) + report_path, = (Path(tmp) / "runs").glob("*/report.json") + entry = json.loads(report_path.read_text())["scenarios"][0] + self.assertEqual(entry["status"], "cancelled") + self.assertIn("cancelled by signal", entry["error"]) + self.assertNotIn("cleanup_error", entry) + finally: + if pidfile.exists(): + leader, _child = json.loads(pidfile.read_text()) + portable_process._stop_group(leader, 0.2) + + + def test_daemon_cleanup_failure_retains_completed_results_and_owned_artifacts(self): + binary = self.make_bin() + for agent_failure, cleanup_failure in ( + (RuntimeError("agent failed"), OSError("cleanup failed")), + (None, OSError("cleanup failed")), + (None, KeyboardInterrupt("cancelled during cleanup")), + ): + with self.subTest(agent=agent_failure, cleanup=cleanup_failure): + environment = run_real_model.create_eval_environment("cleanup-failure") + self.addCleanup(environment.cleanup) + daemon_pid = None + outcomes = [{ + "name": "completed assertion", + "passed": True, + "actual": 1, + "op": "eq", + "expected": 1, + }] + transcripts = [{"turn_valid": True}] + + def stop_then_fail(env): + nonlocal daemon_pid + daemon_pid = env.daemon.pid + run_real_model.stop_owned_process(env.daemon, 5) + raise cleanup_failure + + with tempfile.TemporaryDirectory() as tmp: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object(run_real_model, "load_scenarios", return_value=[self.scenario()]), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value=binary), + mock.patch.object(run_real_model, "create_eval_environment", return_value=environment), + mock.patch.object(run_real_model, "drive_hermes", side_effect=agent_failure, return_value=transcripts), + mock.patch.object(run_real_model, "evaluate_assertions", return_value=outcomes), + mock.patch.object(run_real_model.EvalEnvironment, "stop_daemon", autospec=True, side_effect=stop_then_fail), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + cancelled = isinstance(cleanup_failure, KeyboardInterrupt) + self.assertEqual( + run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), + 130 if cancelled else 1, + ) + report = json.loads(next(Path(tmp).glob("*/report.json")).read_text()) + self.assertEqual(report["status"], "cancelled" if cancelled else "fail") + self.assertNotIn("reason", report) + entry = report["scenarios"][0] + if agent_failure is not None: + self.assertEqual(entry["error"], str(agent_failure)) + else: + if cancelled: + self.assertEqual(entry["error"], str(cleanup_failure)) + else: + self.assertNotIn("error", entry) + self.assertEqual(entry["assertions"], outcomes) + self.assertEqual(entry["transcripts"], transcripts) + self.assertIn(str(cleanup_failure), entry["cleanup_error"]) + # An interrupted/failed stop has not returned confirmation to + # its owner, even when this injected failure follows a reap. + self.assertTrue(environment.root.exists()) + self.assertTrue(environment.socket_dir.exists()) + self.assertEqual(entry["retained_profile"], str(environment.root)) + self.assertTrue(Path(entry["retained_fixture"]).exists()) + self.assertIsNotNone(environment.daemon) + with self.assertRaises(ProcessLookupError): + os.kill(daemon_pid, 0) + environment.cleanup() + self.assertFalse(environment.root.exists()) + self.assertFalse(environment.socket_dir.exists()) + + +class ProcessCancellationTest(unittest.TestCase): + def test_stop_failure_still_kills_and_reaps_the_owned_process(self): + process = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(600)"], + start_new_session=True, + ) + try: + with mock.patch.object(portable_process, "_stop_group", side_effect=OSError("stop failed")): + with self.assertRaisesRegex(OSError, "stop failed"): + portable_process.stop_owned_process(process, 1) + self.assertIsNotNone(process.returncode) + with self.assertRaises(ProcessLookupError): + os.kill(process.pid, 0) + finally: + portable_process.stop_owned_process(process, 1) + + def test_failed_forced_stop_preserves_the_stop_error_and_live_process_owner(self): + process = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(600)"], + start_new_session=True, + ) + stop_error = OSError("graceful stop failed") + force_error = PermissionError("forced stop denied") + try: + with ( + mock.patch.object(portable_process, "_stop_group", side_effect=stop_error), + mock.patch.object(portable_process, "_signal_group", side_effect=force_error), + self.assertRaises(OSError) as raised, + ): + portable_process.stop_owned_process(process, 1) + self.assertIs(raised.exception, stop_error) + self.assertIs(raised.exception.__cause__, force_error) + self.assertIsNone(process.poll()) + os.kill(process.pid, 0) + finally: + portable_process.stop_owned_process(process, 1) + + def test_signals_during_binary_resolution_and_turns_stop_owned_groups_and_write_reports(self): + workload = """ +import os, subprocess, sys, time +from pathlib import Path +child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(600)']) +Path(sys.argv[1]).write_text(str(os.getpid())) +time.sleep(600) +""" + entrypoint = """ +import sys +from pathlib import Path +from unittest import mock +sys.path.insert(0, sys.argv[1]) +import run_real_model as runner +root, phase = Path(sys.argv[2]), sys.argv[3] +runner.RUNS_DIR = root / 'runs' +scenario = {'id': 'cancel', 'setup': {}, 'assertions': []} +def work(*args, **kwargs): + return runner.run([sys.executable, '-c', sys.argv[4], str(root / 'work.pid')]) +with mock.patch.object(runner, 'load_scenarios', return_value=[scenario]): + if phase == 'build': + runner.resolve_tracedecay_bin = work + else: + runner.resolve_tracedecay_bin = lambda explicit: sys.argv[5] + runner.drive_hermes = work + sys.exit(runner.main(['--agent-turn', '--i-understand-model-cost'])) +""" + for phase in ("build", "turn"): + for sig in (signal.SIGINT, signal.SIGTERM): + with self.subTest(phase=phase, signal=sig), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + binary = root / "tracedecay" + binary.write_text(FAKE_BIN) + binary.chmod(0o755) + process = subprocess.Popen( + [sys.executable, "-c", entrypoint, str(MODULE_PATH.parent), tmp, + phase, workload, str(binary)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, + start_new_session=True, + ) + group = None + try: + deadline = time.monotonic() + 15 + while not (root / "work.pid").exists() and process.poll() is None: + if time.monotonic() >= deadline: + self.fail("owned workload did not start") + time.sleep(0.02) + self.assertTrue((root / "work.pid").exists()) + group = int((root / "work.pid").read_text()) + process.send_signal(sig) + output, _ = process.communicate(timeout=15) + self.assertEqual(process.returncode, 130, output) + report = json.loads(next((root / "runs").glob("*/report.json")).read_text()) + self.assertEqual(report["status"], "cancelled", report) + if phase == "turn": + self.assertEqual(report["scenarios"][0]["status"], "cancelled") + self.assertEqual( + subprocess.run( + [sys.executable, "-S", str(run_real_model.PROCESS_HELPER), + "group-alive", "--pid", str(group)], + check=False, + ).returncode, + 1, + "a cancelled workload left runnable descendants", + ) + with self.assertRaises(ProcessLookupError): + os.kill(group, 0) + finally: + run_real_model.stop_owned_process(process, 1) + if group is not None: + try: + os.killpg(group, signal.SIGKILL) + except ProcessLookupError: + pass + if process.stdout is not None: + process.stdout.close() + + def test_timeout_stops_descendants_and_reaps_the_leader(self): + with tempfile.TemporaryDirectory() as tmp: + pid_path = Path(tmp) / "pid" + with self.assertRaises(subprocess.TimeoutExpired): + run_real_model.run( + [sys.executable, "-c", + "import os, subprocess, sys, time; " + "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(600)']); " + "open(sys.argv[1], 'w').write(str(os.getpid())); time.sleep(600)", + str(pid_path)], timeout=0.5, + ) + pid = int(pid_path.read_text()) + with self.assertRaises(ProcessLookupError): + os.kill(pid, 0) + self.assertEqual(subprocess.run( + [sys.executable, "-S", str(run_real_model.PROCESS_HELPER), + "group-alive", "--pid", str(pid)], check=False, + ).returncode, 1) + + +class ReportLifecycleTest(unittest.TestCase): + def test_command_preserves_captured_stderr_and_closes_both_pipes(self): + processes = [] + original_popen = subprocess.Popen + + def spawn(*args, **kwargs): + process = original_popen(*args, **kwargs) + processes.append(process) + return process + + with mock.patch.object(run_real_model.subprocess, "Popen", side_effect=spawn): + result = run_real_model.run( + [sys.executable, "-c", "import sys; print('output'); print('detail', file=sys.stderr)"], + stderr=subprocess.PIPE, + ) + self.assertEqual(result.stdout, "output\n") + self.assertEqual(result.stderr, "detail\n") + process, = processes + self.assertTrue(process.stdout.closed) + self.assertTrue(process.stderr.closed) + + def test_command_cleanup_failure_keeps_the_original_command_error(self): + with mock.patch.object( + run_real_model, "stop_owned_process", side_effect=OSError("stop failed") + ): + with self.assertRaises(run_real_model.OwnedProcessCleanupError) as raised: + run_real_model.run([sys.executable, "-c", "print('original output'); exit(7)"]) + error = raised.exception + self.assertIsInstance(error.original_error, run_real_model.EvalSetupError) + self.assertIn("command failed (7)", error.primary_detail()) + self.assertIn("original output", error.primary_detail()) + self.assertEqual(str(error.cleanup_error), "stop failed") + self.assertFalse(error.cancelled) + self.assertEqual(error.process.returncode, 7) + self.assertTrue(error.process.stdout.closed) + + def test_resolution_failure_and_cancellation_always_write_report(self): + for error, expected_status, expected_code in ( + (run_real_model.EvalSetupError("build failed"), "fail", 1), + (KeyboardInterrupt("build interrupted"), "cancelled", 130), + ): + with self.subTest(status=expected_status), tempfile.TemporaryDirectory() as tmp: + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object(run_real_model, "load_scenarios", return_value=[{"id": "test"}]), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", side_effect=error), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + code = run_real_model.main(["--agent-turn", "--i-understand-model-cost"]) + self.assertEqual(code, expected_code) + report_path, = Path(tmp).glob("*/report.json") + report = json.loads(report_path.read_text()) + self.assertEqual(report["status"], expected_status) + self.assertEqual(report["reason"], str(error)) + self.assertEqual(report["phase"], "binary resolution") + self.assertEqual(report["scenarios"], []) + + def test_cancellation_between_scenarios_and_during_report_write(self): + class InterruptedScenarios(list): + def __iter__(self): + yield self[0] + raise KeyboardInterrupt("between scenarios") + + for during_write in (False, True): + with self.subTest(during_write=during_write), tempfile.TemporaryDirectory() as tmp: + scenarios = [{"id": "first"}] if during_write else InterruptedScenarios([{"id": "first"}]) + original_write = Path.write_text + attempted = False + + def write(path, contents, *args, **kwargs): + nonlocal attempted + if during_write and path.name == "report.json" and not attempted: + attempted = True + os.kill(os.getpid(), signal.SIGTERM) + return original_write(path, contents, *args, **kwargs) + + def run_scenario(_args, scenario, _run_dir, report): + report["scenarios"].append({"id": scenario["id"], "status": "pass"}) + return True, False + + with ( + mock.patch.object(run_real_model, "RUNS_DIR", Path(tmp)), + mock.patch.object(run_real_model, "load_scenarios", return_value=scenarios), + mock.patch.object(run_real_model, "resolve_tracedecay_bin", return_value="/unused"), + mock.patch.object(run_real_model, "run_scenario", side_effect=run_scenario), + mock.patch.object(Path, "write_text", new=write), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual(run_real_model.main(["--agent-turn", "--i-understand-model-cost"]), 130) + report_path, = Path(tmp).glob("*/report.json") + report = json.loads(report_path.read_text()) + self.assertEqual(report["status"], "cancelled") + self.assertEqual(report["scenarios"], [{"id": "first", "status": "pass"}]) class ExactFactStoreEvaluationTest(unittest.TestCase): @@ -206,15 +1268,18 @@ def test_fixture_preload_requires_the_seeded_fact_in_search_results(self): ), ] ) - with mock.patch.object( - run_real_model, - "run", - side_effect=lambda *args, **kwargs: next(responses), + with ( + tempfile.TemporaryDirectory() as fixture, + mock.patch.object( + run_real_model, + "run", + side_effect=lambda *args, **kwargs: next(responses), + ), ): with self.assertRaisesRegex( run_real_model.ExactFactStoreError, "did not return FactId" ): - run_real_model.build_fixture(scenario, "tracedecay", {}) + run_real_model.seed_fixture(scenario, "tracedecay", {}, Path(fixture)) def test_non_real_model_assertions_are_skipped_without_a_store_read(self): scenario = { diff --git a/scripts/BUILD.bazel b/scripts/BUILD.bazel index 5f568bcca2..fa8d73f87a 100644 --- a/scripts/BUILD.bazel +++ b/scripts/BUILD.bazel @@ -101,6 +101,7 @@ exports_files( [ "bazel/rust_test.bzl", "build-eval-binary.sh", + "lib/portable_process.py", "test-rust-cache-lineage.py", "test-check-packaged-lsp-bridge.py", "test-check-sdk-publish-workflow.py", diff --git a/scripts/lib/portable_process.py b/scripts/lib/portable_process.py index 718a9f0f8e..ff06cadf4c 100755 --- a/scripts/lib/portable_process.py +++ b/scripts/lib/portable_process.py @@ -169,6 +169,44 @@ def _return_code(status: int) -> int: return status if status >= 0 else 128 + abs(status) +def stop_owned_process(process: subprocess.Popen, grace_seconds: float) -> None: + """Stop an owned session, including descendants, then reap its leader. + + Callers must create the child with ``start_new_session=True``. This also + cleans children left behind by an already-exited leader, so a completed + command cannot leave work running against a fixture being removed. + """ + stop_error = None + try: + _stop_group(process.pid, grace_seconds) + except Exception as exc: + stop_error = exc + # A failed graceful stop must still attempt to stop owned work. Keep + # the initial error if forced cleanup also fails; an interrupt remains + # a cancellation for the enclosing resource owner to finish. + try: + _signal_group(process.pid, signal.SIGKILL) + except Exception as cleanup_error: + raise stop_error from cleanup_error + try: + try: + process.wait(timeout=grace_seconds) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + deadline = time.monotonic() + grace_seconds + while _group_alive(process.pid) and time.monotonic() < deadline: + time.sleep(min(0.05, max(0.0, deadline - time.monotonic()))) + if _group_alive(process.pid): + raise RuntimeError(f"owned process group {process.pid} did not stop") + except Exception as cleanup_error: + if stop_error is not None: + raise stop_error from cleanup_error + raise + if stop_error is not None: + raise stop_error + + def command_exec_session(args: argparse.Namespace) -> int: command = _command_after_separator(args.command) die_with_parent(args.parent_pid)