From 612e7d49346ef2404f361cd2cbf98ec5c38a2cf4 Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Wed, 28 Oct 2026 00:17:52 +0800 Subject: [PATCH 1/5] fix: make run state truthful and enforce the metric contract Reserve the terminal run state instead of assuming it. A run that raises from a hook outside the documented adapter error contract now records "crashed" and re-raises, so status.json can no longer claim "completed" for a crashed run. Termination moved out of the finally block: a failure there records "termination_failed" with the original measurement error beside it instead of masking that error and leaving "running" behind. Move the finite, positive metric contract into validate_metric and apply it to whatever a custom metric reader returns. A reader yielding 0.0 used to reach _summary and raise ZeroDivisionError inside termination, destroying the whole evaluation; it now records a metric error and the run keeps its evaluation.json. Exclude metrics from trials that did not complete. _summary reports the count in excluded_trials and evaluate names the exclusion in reasons, so an aborted trial is never presented as a measurement. Agent-assisted-by: Claude Code (author) Signed-off-by: GUOGUO --- docs/architecture.md | 14 +- src/inferrsi/harness/evaluation.py | 12 +- src/inferrsi/harness/runner.py | 154 +++++++++++++-------- src/inferrsi/harness/scenarios/external.py | 9 +- src/inferrsi/model/protocols/metrics.py | 15 +- tests/test_runtime.py | 115 +++++++++++++++ 6 files changed, 247 insertions(+), 72 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 4055dc3e2..9974cad14 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -187,8 +187,18 @@ run_dir = run_experiment( The standard executor retains timeout, cancellation, logs, result paths and child cleanup. A custom `execute(command, cwd, trial_dir, deadline, metric_key)` must preserve that trial record contract. Prefer composing `execute_trial` over -reimplementing process lifetime. Metric readers must preserve the finite, -positive metric contract before returning a float. +reimplementing process lifetime. Metric readers convert a result document into a +float; `execute_trial` then applies the one finite, positive metric contract +(`model.protocols.metrics.validate_metric`) to whatever the reader returns, so a +reader that yields `0.0`, a `bool` or a non-finite value records a metric error +rather than entering the evaluation. + +Run state is written by the termination path, never assumed. `status.json` +reports `completed` only after the final artifacts were written; +a run stopped by a command state keeps `timeout`, `failed` or `cancelled`; an +exception outside the adapter error contract writes `crashed` and re-raises; and a +failure while writing the terminal artifacts writes `termination_failed` with the +original measurement error beside it. A new source adapter implements `snapshot(repo, excluded_run_dir)` and returns `available` plus a stable `snapshot_hash` and explanatory metadata. It must diff --git a/src/inferrsi/harness/evaluation.py b/src/inferrsi/harness/evaluation.py index 9ff4cb605..62f9cd33e 100644 --- a/src/inferrsi/harness/evaluation.py +++ b/src/inferrsi/harness/evaluation.py @@ -8,7 +8,11 @@ def _summary(trials: list[dict]) -> dict | None: - values = [trial["metric"] for trial in trials if trial.get("metric") is not None] + # Only a completed trial recorded a measurement. A timed-out or failed trial + # may still have written a result file, but its metric is not a datapoint; + # `evaluate` reports the exclusion in `reasons` instead of dropping it silently. + measured = [trial for trial in trials if trial.get("status") == "completed"] + values = [trial["metric"] for trial in measured if trial.get("metric") is not None] if not values: return None ordered = sorted(values) @@ -25,6 +29,7 @@ def _summary(trials: list[dict]) -> dict | None: "min": min(values), "max": max(values), "range_pct_of_median": spread if math.isfinite(spread) else None, + "excluded_trials": len(trials) - len(values), } @@ -43,7 +48,10 @@ def evaluate( ) for index, trial in enumerate(trials, 1): if trial.get("status") != "completed": - incomplete.append(f"{label} trial {index}: {trial.get('status')}") + incomplete.append( + f"{label} trial {index}: {trial.get('status')} trial excluded from " + "the recorded measurements" + ) if trial.get("metric") is None: incomplete.append( f"{label} trial {index}: {trial.get('metric_error') or 'metric missing'}" diff --git a/src/inferrsi/harness/runner.py b/src/inferrsi/harness/runner.py index a59707337..8ba4a212d 100644 --- a/src/inferrsi/harness/runner.py +++ b/src/inferrsi/harness/runner.py @@ -35,7 +35,8 @@ def run_experiment( _write_json(run_dir / "environment.json", environment) baseline, candidate, validation = [], [], None before, after, source_checks = {}, {}, [] - state, error = "completed", None + # No state is claimed until the work that would establish it has finished. + state, error = None, None def event(phase: str, **details: object) -> None: record = {"time": datetime.now(timezone.utc).isoformat(), "phase": phase, **details} @@ -45,6 +46,17 @@ def event(phase: str, **details: object) -> None: run_dir / "status.json", {"state": "running", "phase": phase, "run_dir": str(run_dir)} ) + def record_status(phase: str, **fields: object) -> None: + _write_json( + run_dir / "status.json", + { + "run_dir": str(run_dir), + "elapsed_seconds": time.monotonic() - started, + "phase": phase, + **fields, + }, + ) + repos = { "baseline": Path(spec.repo).resolve(), "candidate": Path(spec.candidate_repo or spec.repo).resolve(), @@ -66,6 +78,66 @@ def execute_checked( _write_json(trial_dir / "source.json", check) return record + def terminate() -> None: + """Write the terminal artifacts; a failure here is reported, not hidden.""" + nonlocal after + try: + after = {label: snapshot(repo, run_dir) for label, repo in repos.items()} + _write_json(run_dir / "repositories.final.json", after) + _write_json(run_dir / "source_checks.json", source_checks) + identities = list(before.values()) + list(after.values()) + identities.extend( + identity + for check in source_checks + for identity in (check["before"], check["after"]) + ) + known = bool(before) and all(item.get("available") for item in identities) + unchanged = ( + known + and all( + before[label]["snapshot_hash"] == after[label]["snapshot_hash"] + for label in repos + ) + and all( + check["before"]["snapshot_hash"] + == check["after"]["snapshot_hash"] + == before[check["side"]]["snapshot_hash"] + for check in source_checks + ) + ) + integrity = { + "verified": bool(unchanged), + "reason": "Git source snapshots unchanged" + if unchanged + else "Git source snapshot unavailable" + if not known + else "source changed during experiment", + "command_checks": len(source_checks), + } + result = evaluate(spec, baseline, candidate, validation, integrity) + if error: + result["reasons"].append(error) + result["verdict"] = "INCONCLUSIVE" + _write_json(run_dir / "evaluation.json", result) + event("complete", state=state, verdict=result["verdict"]) + record_status( + "complete", state=state, verdict=result["verdict"], error=error + ) + write_report(run_dir) + except (OSError, ValueError, subprocess.SubprocessError) as exc: + # The run did not reach a finished state. Keep the measurement error + # visible beside the termination failure instead of replacing it. + record_status( + "termination", + state="termination_failed", + verdict=None, + error=error, + requested_state=state, + termination_error=f"{type(exc).__name__}: {exc}", + ) + raise + + stopped = None try: event("source_snapshot") before = {label: snapshot(repo, run_dir) for label, repo in repos.items()} @@ -82,78 +154,42 @@ def execute_checked( ) (baseline if label == "baseline" else candidate).append(record) if record["status"] == "cancelled": - state = "cancelled" raise KeyboardInterrupt if record["status"] != "completed": - state = "timeout" if record["status"] == "timeout" else "failed" + stopped = record["status"] break - if state != "completed": + if stopped is not None: break - if spec.validation_command is not None and state == "completed": + if spec.validation_command is not None and stopped is None: event("quality_validation") validation = execute_checked( spec.validation_command, "candidate", run_dir / "validation", None ) if validation["status"] != "completed": - state = ( - "cancelled" - if validation["status"] == "cancelled" - else "timeout" - if validation["status"] == "timeout" - else "failed" - ) + stopped = validation["status"] elif spec.validation_command is not None: validation = {"status": "skipped", "reason": "measurement did not complete"} + if stopped is None: + state = "completed" + else: + state = ( + "cancelled" + if stopped == "cancelled" + else "timeout" + if stopped == "timeout" + else "failed" + ) except KeyboardInterrupt: state, error = "cancelled", "interrupted by user" except (OSError, ValueError, subprocess.SubprocessError) as exc: state, error = "failed", str(exc) - finally: - after = {label: snapshot(repo, run_dir) for label, repo in repos.items()} - _write_json(run_dir / "repositories.final.json", after) - _write_json(run_dir / "source_checks.json", source_checks) - identities = list(before.values()) + list(after.values()) - identities.extend( - identity for check in source_checks for identity in (check["before"], check["after"]) - ) - known = bool(before) and all(item.get("available") for item in identities) - unchanged = ( - known - and all( - before[label]["snapshot_hash"] == after[label]["snapshot_hash"] for label in repos - ) - and all( - check["before"]["snapshot_hash"] - == check["after"]["snapshot_hash"] - == before[check["side"]]["snapshot_hash"] - for check in source_checks - ) - ) - integrity = { - "verified": bool(unchanged), - "reason": "Git source snapshots unchanged" - if unchanged - else "Git source snapshot unavailable" - if not known - else "source changed during experiment", - "command_checks": len(source_checks), - } - result = evaluate(spec, baseline, candidate, validation, integrity) - if error: - result["reasons"].append(error) - result["verdict"] = "INCONCLUSIVE" - _write_json(run_dir / "evaluation.json", result) - event("complete", state=state, verdict=result["verdict"]) - _write_json( - run_dir / "status.json", - { - "state": state, - "phase": "complete", - "verdict": result["verdict"], - "run_dir": str(run_dir), - "elapsed_seconds": time.monotonic() - started, - "error": error, - }, + except BaseException as exc: + # A hook failure outside the documented error contract still left a run + # directory. Record the crash truthfully, then let the caller see it. + event("crashed") + record_status( + "crashed", state="crashed", verdict=None, error=f"{type(exc).__name__}: {exc}" ) - write_report(run_dir) + raise + terminate() return run_dir diff --git a/src/inferrsi/harness/scenarios/external.py b/src/inferrsi/harness/scenarios/external.py index 986a77184..e665293f7 100644 --- a/src/inferrsi/harness/scenarios/external.py +++ b/src/inferrsi/harness/scenarios/external.py @@ -6,7 +6,7 @@ from ...infra.artifacts import atomic_write_json as _write_json from ...infra.executors.local import execute_command -from ..evaluation import read_metric +from ...model.protocols.metrics import read_metric, validate_metric def execute_trial( @@ -22,9 +22,10 @@ def execute_trial( result_path = Path(record["result_path"]) if metric_key is not None: try: - record["metric"] = metric_reader( - json.loads(result_path.read_text(encoding="utf-8")), metric_key - ) + document = json.loads(result_path.read_text(encoding="utf-8")) + # A reader is a conversion, not an exemption: the common metric + # contract applies here, before the value enters the trial record. + record["metric"] = validate_metric(metric_key, metric_reader(document, metric_key)) except (OSError, ValueError, TypeError, OverflowError) as exc: record["metric_error"] = str(exc) _write_json(trial_dir / "execution.json", record) diff --git a/src/inferrsi/model/protocols/metrics.py b/src/inferrsi/model/protocols/metrics.py index c51e76e06..44a07f309 100644 --- a/src/inferrsi/model/protocols/metrics.py +++ b/src/inferrsi/model/protocols/metrics.py @@ -3,14 +3,19 @@ import math +def validate_metric(key: str, value) -> float: + """Apply the one numeric contract, whatever produced the value.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"metric {key} must be a number") + if not math.isfinite(value) or value <= 0: + raise ValueError(f"metric {key} must be finite and positive") + return float(value) + + def read_metric(result: dict, key: str) -> float: value = result for part in key.split("."): if not isinstance(value, dict) or part not in value: raise ValueError(f"missing metric: {key}") value = value[part] - if isinstance(value, bool) or not isinstance(value, (int, float)): - raise ValueError(f"metric {key} must be a number") - if not math.isfinite(value) or value <= 0: - raise ValueError(f"metric {key} must be finite and positive") - return float(value) + return validate_metric(key, value) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index a5781fae1..24937a8fc 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -7,11 +7,15 @@ import tempfile import unittest from dataclasses import replace +from functools import partial from pathlib import Path from unittest.mock import patch from inferrsi.contracts import RunSpec from inferrsi.execution import repository_identity, run_experiment +from inferrsi.harness.evaluation import _summary, evaluate +from inferrsi.harness.runner import run_experiment as run_with_adapters +from inferrsi.harness.scenarios.external import execute_trial from inferrsi.reporting import write_report @@ -352,6 +356,117 @@ def test_contract_rejects_invalid_inputs_before_allocating_run(self): with self.subTest(changes=changes), self.assertRaises(ValueError): run_experiment(replace(self.spec, **changes), {}) + def latest_run(self): + return max((Path(self.root / "runs")).iterdir(), key=lambda path: path.name) + + def test_hook_exception_is_recorded_as_crashed_and_reaches_the_caller(self): + # A KeyError is outside the documented adapter error contract, so it must + # not be absorbed. The run directory must not claim completion either. + def reader(document, key): + return document["metrics"]["latency"] + + with self.assertRaises(KeyError): + run_with_adapters( + self.spec, + {"backend": "cpu", "test": True}, + execute=partial(execute_trial, metric_reader=reader), + ) + run = self.latest_run() + status = json.loads((run / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "crashed") + self.assertNotEqual(status["state"], "completed") + self.assertEqual(status["error"], "KeyError: 'latency'") + self.assertIsNone(status["verdict"]) + self.assertFalse((run / "evaluation.json").exists()) + events = [ + json.loads(line) + for line in (run / "events.jsonl").read_text(encoding="utf-8").splitlines() + ] + self.assertEqual([event["phase"] for event in events][-1], "crashed") + self.assertNotIn("complete", [event["phase"] for event in events]) + + def test_reader_violating_the_metric_contract_keeps_the_evaluation(self): + # 0.0 is outside the finite-positive contract; before validation it reached + # _summary and raised ZeroDivisionError inside the termination path. + def reader(document, key): + return 0.0 + + path = run_with_adapters( + self.spec, + {"backend": "cpu", "test": True}, + execute=partial(execute_trial, metric_reader=reader), + ) + result = json.loads((path / "evaluation.json").read_text(encoding="utf-8")) + self.assertEqual(result["verdict"], "INCONCLUSIVE") + self.assertIsNone(result["baseline"]) + self.assertIsNone(result["candidate"]) + self.assertTrue(any("finite and positive" in reason for reason in result["reasons"])) + status = json.loads((path / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "completed") + trial = json.loads( + (path / "baseline/trial-001/execution.json").read_text(encoding="utf-8") + ) + self.assertIsNone(trial["metric"]) + self.assertIn("finite and positive", trial["metric_error"]) + + def test_snapshot_failure_terminates_with_a_named_state_and_retained_error(self): + # The second snapshot call fails inside the try block: the run records + # "failed". Cleanup needs another snapshot, so it fails too. The original + # error must survive that second failure rather than be replaced by it. + calls = [] + + def snapshot(path, exclude=None): + calls.append(path) + if len(calls) > 1: + raise OSError("snapshot backend unavailable") + return {"available": True, "snapshot_hash": "fixture", "path": str(path)} + + with self.assertRaises(OSError): + run_with_adapters( + replace(self.spec, repetitions=1), + {"backend": "cpu", "test": True}, + snapshot=snapshot, + ) + run = self.latest_run() + status = json.loads((run / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "termination_failed") + self.assertEqual(status["phase"], "termination") + self.assertEqual(status["requested_state"], "failed") + self.assertEqual(status["error"], "snapshot backend unavailable") + self.assertIn("snapshot backend unavailable", status["termination_error"]) + self.assertFalse((run / "evaluation.json").exists()) + self.assertFalse((run / "report.html").exists()) + + def test_failed_trial_metric_is_excluded_and_named(self): + failure = (sys.executable, "-c", "import sys; sys.exit(7)") + _, evaluation = self.run_spec(validation_command=failure) + # A completed quality command that rejects its gate fails the run, but + # both measurement trials completed, so both remain measured. + self.assertEqual(evaluation["verdict"], "FAIL") + self.assertEqual(evaluation["baseline"]["values"], [100.0, 100.0]) + self.assertEqual(evaluation["candidate"]["values"], [80.0, 80.0]) + self.assertEqual(evaluation["candidate"]["excluded_trials"], 0) + + def test_summary_ignores_a_timed_out_trial_metric(self): + summary = _summary( + [{"status": "completed", "metric": 10.0}, {"status": "timeout", "metric": 5.0}] + ) + self.assertEqual(summary["values"], [10.0]) + self.assertEqual(summary["excluded_trials"], 1) + spec = replace(self.spec, repetitions=2) + result = evaluate( + spec, + [{"status": "completed", "metric": 10.0}, {"status": "timeout", "metric": 5.0}], + [{"status": "completed", "metric": 8.0}, {"status": "completed", "metric": 8.0}], + {"status": "completed"}, + {"verified": True}, + ) + # A timed-out trial's metric must not enter the baseline median. + self.assertEqual(result["baseline"]["values"], [10.0]) + self.assertEqual(result["baseline"]["excluded_trials"], 1) + self.assertTrue(any("excluded from the recorded measurements" in reason + for reason in result["reasons"])) + if __name__ == "__main__": unittest.main() From 709066e8c549efaaf3c02a496bcc328af783d46e Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Wed, 28 Oct 2026 00:18:46 +0800 Subject: [PATCH 2/5] fix: make atomic JSON writes survive Windows replace contention A fixed .tmp name let two writers share one temporary file, and a bare os.replace raised PermissionError whenever another handle held the destination open. The temporary name is now per writer, and the replace retries a bounded number of times before surfacing the error. The failed temporary file is removed either way. Agent-assisted-by: Claude Code (author) Signed-off-by: GUOGUO --- src/inferrsi/infra/artifacts.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/src/inferrsi/infra/artifacts.py b/src/inferrsi/infra/artifacts.py index 9493259b7..774c5ea35 100644 --- a/src/inferrsi/infra/artifacts.py +++ b/src/inferrsi/infra/artifacts.py @@ -2,8 +2,16 @@ import hashlib import json +import os +import time +import uuid from pathlib import Path +# Windows refuses os.replace while another handle holds the destination open. +# A bounded retry covers the observed contention without hiding a real failure. +REPLACE_ATTEMPTS = 8 +REPLACE_DELAY_SECONDS = 0.05 + def digest(value): return hashlib.sha256( @@ -29,8 +37,21 @@ def write_json(path, value): def atomic_write_json(path: Path, value: object) -> None: - temporary = path.with_suffix(path.suffix + ".tmp") + path = Path(path) + # A per-writer temporary name keeps two writers from sharing one file. + temporary = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8" ) - temporary.replace(path) + try: + for attempt in range(REPLACE_ATTEMPTS): + try: + temporary.replace(path) + return + except PermissionError: + if attempt + 1 == REPLACE_ATTEMPTS: + raise + time.sleep(REPLACE_DELAY_SECONDS) + finally: + if temporary.exists(): + temporary.unlink() From c2267652ec5b8e38077dceeac4f1b8d58bf684c2 Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Wed, 28 Oct 2026 00:19:42 +0800 Subject: [PATCH 3/5] fix: derive comparison authority instead of trusting stated fields load_plan now re-derives what the frozen inputs determine: the schedule from the frozen tasks and repeats, the identical-bundle flag from the frozen arms, and each exam hash from the frozen file list. An edited freeze.json with a recomputed SHA is rejected, so the authorization hash no longer covers only the bytes of the file that states the semantics. report re-grades every retained PASS/FAIL with the trusted in-tree verifier, the same authority the PR path already uses. A results tree rewritten with a consistent receipt can no longer publish a grade the answers do not support; a custom exam is marked frozen_copy_only rather than silently trusted. A checkpoint can no longer recommend report_accepted while the comparison has not executed: acceptance of criteria is not execution of the work. Unknown token usage stops before the next trial when it cannot be shown to be within the authorized limit, instead of running the full schedule. The PR case mapping's skill ID is validated before an output path is built from it. Agent-assisted-by: Claude Code (author) Signed-off-by: GUOGUO --- tests/test_goal_checkpoint.py | 29 +++++++++++ tests/test_skill_eval_compare.py | 87 +++++++++++++++++++++++++++++++ tests/test_skill_eval_pr.py | 15 ++++++ tools/goal_checkpoint.py | 2 - tools/skill_eval_compare.py | 88 +++++++++++++++++++++++++++++--- tools/skill_eval_pr.py | 4 ++ 6 files changed, 216 insertions(+), 9 deletions(-) diff --git a/tests/test_goal_checkpoint.py b/tests/test_goal_checkpoint.py index f5806b425..2cff24f16 100644 --- a/tests/test_goal_checkpoint.py +++ b/tests/test_goal_checkpoint.py @@ -220,6 +220,33 @@ def test_distinct_damaged_inputs_keep_distinct_checkpoint_identities( name: sha256(compare.ROOT / name) for name in compare.CONTROLLERS} +def test_accepted_criteria_cannot_recommend_acceptance_before_any_trial_ran( + tmp_path, comparison_templates, +): + # Every criterion can be explicitly accepted while the comparison is still + # only prepared. Acceptance of criteria is not execution of the comparison. + path, goal, run = setup_goal(tmp_path, comparison_templates[0]) + # A never-executed comparison has no results to cite; the frozen plan is all + # the evidence that exists, and the owner can still accept the criteria. + criterion = goal["criteria"][0] + criterion["evidence"] = [pin(run / "plan/freeze.json", path.parent)] + record = path.parent / "acceptance.json" + write_json(record, {"goal_id": goal["id"], "criterion_id": criterion["id"], + "criterion_sha256": checkpoint.criterion_sha256(goal, criterion), + "accepted_by": goal["acceptance_owner"], "decision": "accepted"}) + criterion["acceptance"] = pin(record, path.parent) + write_json(path, goal) + result = checkpoint.build_checkpoint(path, run) + assert result["comparison"]["status"]["execution_state"] == "not_completed" + assert not result["remaining"] and not result["problems"] + assert result["recommendation"]["action"] != "report_accepted" + assert result["recommendation"]["action"] == "request_authorization" + authorize(path, goal) + result = checkpoint.build_checkpoint(path, run) + assert result["recommendation"]["action"] not in ("report_accepted", "wait_acceptance") + assert result["recommendation"]["action"] == "propose_step" + + def test_interruption_never_becomes_a_fresh_or_accepted_attempt(tmp_path, comparison_templates): path, goal, run = setup_goal(tmp_path, comparison_templates[0]) authorize(path, goal) @@ -254,6 +281,8 @@ def test_real_client_failures_preserve_evidence_and_choose_the_actual_next_step( before = files(run) result = checkpoint.build_checkpoint(path, run) assert not result["problems"] + # One case is two trials, so an unknown usage on the last trial starts nothing + # further; the schedule completed and the next step is owner acceptance. expected = "prepare_new_plan" if mode in ("nonzero", "timeout") else "wait_acceptance" assert result["recommendation"]["action"] == expected assert result["comparison"]["counts"] == report["counts"] diff --git a/tests/test_skill_eval_compare.py b/tests/test_skill_eval_compare.py index a331bd5f4..8f70d17cd 100644 --- a/tests/test_skill_eval_compare.py +++ b/tests/test_skill_eval_compare.py @@ -694,6 +694,93 @@ def test_required_execution_case_fields_are_validated_before_freeze(tmp_path): assert not output.exists() +def test_edited_freeze_semantics_cannot_pass_with_a_recomputed_hash(tmp_path): + # The authorization hash is taken over freeze.json, so anyone who edits it can + # recompute that hash. The fields it states must agree with the frozen inputs. + config, repo = configuration(tmp_path, ids=("one", "two"), repeats=1) + output = tmp_path / "run" + authorization = compare.prepare(config, repo, output) + plan = compare.load_plan(output) + assert len(plan["schedule"]) == 4 + freeze = output / "plan/freeze.json" + original = freeze.read_bytes() + for field, mutate in ( + ("schedule", lambda value: value.__setitem__("schedule", value["schedule"][:2])), + ("limits", lambda value: value["limits"].__setitem__("repeats", 2)), + ("same_bundle", lambda value: value.__setitem__( + "same_bundle", not value["same_bundle"])), + ("exam", lambda value: value["exam"].__setitem__("grader", "0" * 64)), + ): + with pytest.raises(ValueError, match="Frozen"): + value = json.loads(original) + mutate(value) + write_json(freeze, value) + forged = sha256(freeze) + assert forged != authorization + compare.execute(output, forged, client=FixedOutputClient(tmp_path)) + assert not (output / "results/start.json").exists() + with pytest.raises(ValueError, match="Frozen"): + compare.report(output) + freeze.write_bytes(original) + assert compare.load_plan(output) == plan + + +def test_rewritten_result_and_receipt_is_regraded_by_the_trusted_verifier(tmp_path): + config, repo = configuration(tmp_path, ids=("one",)) + output = tmp_path / "run" + authorization = compare.prepare(config, repo, output) + client = FixedOutputClient(tmp_path, {("one", role): "incorrect" for role in compare.ROLES}) + original = compare.execute(output, authorization, client=client) + assert original["counts"]["unchanged_incorrect"] == 1 + directory = output / "results" / "one-baseline-1" + saved = (directory / "result.json").read_bytes() + record = json.loads(saved) + record.update(verdict="PASS", checks={"passed": True, "assertions": {"fabricated": True}}) + write_json(directory / "result.json", record) + # A rewritten tree can also carry a consistent receipt; that alone proves nothing. + write_json(directory / "receipt.json", {"result_sha256": sha256(directory / "result.json")}) + entry = {"task_id": "one", "condition": "baseline", "repeat": 1, "split": "holdout"} + assert compare.retained_trial(output, entry)["verdict"] == "PASS" + with pytest.raises(ValueError, match="trusted grader"): + compare.report(output) + # Restoring the record does not restore the grade: the receipt is proof of bytes only. + (directory / "result.json").write_bytes(saved) + downgraded = compare.report(output) + trial = downgraded["pairs"][0]["baseline"] + assert trial["verdict"] == "INCONCLUSIVE" + assert trial["evaluation_error"] == "retained_trial_record_hash_mismatch" + assert downgraded["counts"]["unchanged_incorrect"] == 0 + write_json(directory / "receipt.json", {"result_sha256": sha256(directory / "result.json")}) + assert compare.report(output) == original + + +def test_partial_usage_stops_the_schedule_when_the_next_trial_would_be_unauthorized(tmp_path): + # token_stop_new_trials bounds a budget that is unknown, so no further trial + # may start once a call reports usage the controller cannot account for. + config, repo = configuration(tmp_path, ids=("one", "two"), repeats=1) + output = tmp_path / "run" + authorization = compare.prepare(config, repo, output) + client = FixedOutputClient(tmp_path, {("one", "baseline"): "unknown_usage"}) + result = compare.execute(output, authorization, client=client) + assert len(client.calls) == 1 + assert result["status"]["stop_reason"] == "unknown_usage" + assert result["status"]["execution_state"] == "stopped" + assert result["status"]["recorded_trials"] == 1 + assert result["status"]["observed_tokens"] == 0 + assert result["counts"]["missing_pair"] == 2 + + +def test_full_known_usage_schedule_still_runs_to_completion(tmp_path): + config, repo = configuration(tmp_path, ids=("one",), repeats=2) + output = tmp_path / "run" + authorization = compare.prepare(config, repo, output) + client = FixedOutputClient(tmp_path) + result = compare.execute(output, authorization, client=client) + assert len(client.calls) == 4 + assert result["status"]["stop_reason"] is None + assert result["status"]["execution_state"] == "completed" + + def test_modified_trial_verdicts_and_checks_cannot_override_retained_grades(tmp_path): config, repo = configuration(tmp_path, ids=("one",)) output = tmp_path / "run" diff --git a/tests/test_skill_eval_pr.py b/tests/test_skill_eval_pr.py index 7d923f5fd..5e24e4e6b 100644 --- a/tests/test_skill_eval_pr.py +++ b/tests/test_skill_eval_pr.py @@ -110,6 +110,21 @@ def test_deleted_and_unmapped_skill_changes_are_visible_not_passed(tmp_path): approval(plan) +@pytest.mark.parametrize("skill", ["../../escape", "selected/child", "Selected", "", "."]) +def test_mapping_skill_ids_are_validated_before_building_output_paths(tmp_path, skill): + repo, mapping, identity = setup(tmp_path) + raw = compare.read_json(mapping) + raw["skills"][0]["skill"] = skill + write_json(mapping, raw) + output = tmp_path / "plan" + with pytest.raises(ValueError): + integration.prepare(repo, mapping, identity, output) + # Nothing may be written outside --output, and the config file is not created. + assert not (tmp_path / "escape.json").exists() + assert not (tmp_path / "escape").exists() + assert not (output / "pr-plan.json").exists() + + def test_unsupported_file_mode_does_not_disappear_or_become_a_pass(tmp_path): repo, mapping, identity = setup(tmp_path) git(repo, "update-index", "--chmod=+x", "library/selected/resource.txt") diff --git a/tools/goal_checkpoint.py b/tools/goal_checkpoint.py index dddf8e983..c4a6cef6a 100644 --- a/tools/goal_checkpoint.py +++ b/tools/goal_checkpoint.py @@ -134,8 +134,6 @@ def decision(action: Action, *reasons: str) -> Recommendation: return decision("stop_budget", "observed_token_limit_reached") if state == "stopped": return decision("prepare_new_plan", "attempt_stopped_no_resume") - if not checkpoint["remaining"] and not checkpoint["waiting"]: - return decision("report_accepted", "all_criteria_explicitly_accepted") if checkpoint["waiting"]: return decision("wait_dependency", *("dependency:" + d for d in checkpoint["waiting"])) if state == "completed": diff --git a/tools/skill_eval_compare.py b/tools/skill_eval_compare.py index e7be428fe..7828a409e 100644 --- a/tools/skill_eval_compare.py +++ b/tools/skill_eval_compare.py @@ -21,6 +21,7 @@ from inferrsi.model.anthropic_agent import SOURCE as AGENT_SOURCE from inferrsi.model.anthropic_agent import AnthropicAgentClient, validate_schema, validate_spec from tools.skill_eval_pilot import PROMPT, SCHEMA, observed_tokens +from tools.skill_eval_verifier import verify ROOT = Path(__file__).resolve().parents[1] ROLES = ("baseline", "candidate") @@ -366,6 +367,43 @@ def load_plan(output: Path) -> dict: for p in folder.rglob("*") if p.is_file()} if actual != set(plan["files"]) | {"plan/freeze.json"}: raise ValueError("Frozen input file set changed") + # The freeze hash covers every byte of freeze.json, so an edited copy can + # always carry a matching hash. Re-derive what the frozen inputs determine + # instead of trusting the fields that merely state it. + limits = plan["limits"] + if set(limits) != {"repeats", "max_task_calls", "timeout_seconds", "campaign_seconds", + "token_stop_new_trials"}: + raise ValueError("Frozen execution limits changed shape") + for name in limits: + if type(limits[name]) is not int or limits[name] <= 0: + raise ValueError(f"Frozen limit {name} must be a positive integer") + tasks = read_json(output / "plan/tasks.json") + if plan["schedule"] != build_schedule(tasks, limits["repeats"]): + raise ValueError("Frozen schedule disagrees with the frozen tasks and repeats") + if len(plan["schedule"]) > limits["max_task_calls"]: + raise ValueError("Frozen schedule exceeds max_task_calls") + if set(plan["model"]) != {"name", "effort"} or not all( + isinstance(value, str) and value for value in plan["model"].values()): + raise ValueError("Frozen model settings changed shape") + for role in ROLES: + # Bundle bytes are covered by `files` above; the arm only names the commit. + arm = plan["arms"][role] + if set(arm) != {"git_revision", "bundle"} or not re.fullmatch( + r"[0-9a-f]{40}", arm["git_revision"]): + raise ValueError("Frozen arm does not identify a Git commit") + before = plan["arms"]["baseline"]["bundle"] + after = plan["arms"]["candidate"]["bundle"] + if plan["mode"] == "addition" and before is not None: + raise ValueError("Addition baseline must not carry a bundle revision") + if after is None: + raise ValueError("Candidate arm requires an exported bundle") + if plan["same_bundle"] != (before is not None and before["revision"] == after["revision"]): + raise ValueError("Frozen identical-bundle flag disagrees with the frozen arms") + frozen_exam = {"cases": "plan/tasks.json", "schema": "plan/schema.json", + "grader": "validator/verify.py"} + for name, path in frozen_exam.items(): + if plan["exam"].get(name) != plan["files"].get(path): + raise ValueError(f"Frozen exam hash disagrees with the frozen file list: {name}") return plan @@ -407,7 +445,7 @@ def receipt_error(raw_record: bytes, receipt_path: Path, label: str) -> str | No def execute(output: Path, authorization: str, *, client=None, campaign_deadline: float | None = None, - stop_on_unknown_usage: bool = False) -> dict: + stop_on_unknown_usage: bool = True) -> dict: if authorization != sha256(output / "plan/freeze.json"): raise ValueError("Execution requires authorization for this exact plan SHA-256") plan = load_plan(output) @@ -427,9 +465,19 @@ def execute(output: Path, authorization: str, *, client=None, if campaign_deadline is not None: deadline = min(deadline, campaign_deadline) tokens, completed, stop_reason = 0, 0, None + budget_unknown = False for entry in plan["schedule"]: - if time.monotonic() >= deadline or tokens >= plan["limits"]["token_stop_new_trials"]: - stop_reason = "time_budget" if time.monotonic() >= deadline else "token_budget" + if time.monotonic() >= deadline: + stop_reason = "time_budget" + break + if tokens >= plan["limits"]["token_stop_new_trials"]: + stop_reason = "token_budget" + break + if budget_unknown and stop_on_unknown_usage: + # An unknown budget cannot be shown to be within the authorized + # limit, so no further trial is started. Checked before the next + # trial, not after the last one, so a finished schedule stays finished. + stop_reason = "unknown_usage" break task = tasks[entry["task_id"]] role = entry["condition"] @@ -459,10 +507,8 @@ def execute(output: Path, authorization: str, *, client=None, stop_reason = "client_" + result["execution_state"] break usage = normalize_usage(result["observations"]["usage"]) - if stop_on_unknown_usage and (usage is None or any( - usage.get(key) is None for key in ("input_tokens", "output_tokens"))): - stop_reason = "unknown_usage" - break + budget_unknown = usage is None or any( + usage.get(key) is None for key in ("input_tokens", "output_tokens")) status = {"client_runtime": runtime, "recorded_trials": completed, "expected_trials": len(plan["schedule"]), "observed_tokens": tokens, "stop_reason": stop_reason, @@ -595,6 +641,11 @@ def read_report(output: Path) -> dict: "observed_tokens": None, "stop_reason": None, "client_runtime": None} if error is None: status = json.loads(raw_status) + # A receipt is an unkeyed hash: a rewritten tree can carry a consistent one. + # Re-grading retained answers with trusted code is what makes a PASS/FAIL + # verdict independent of the artifact copy the agent could reach. + verify_retained_grader(plan, [t for t in trials if not t["missing"]], + {task["id"]: task for task in read_json(output / "plan/tasks.json")}) result = { "mode": plan["mode"], "skill": plan["skill"], "plan_sha256": sha256(output / "plan/freeze.json"), @@ -617,6 +668,29 @@ def read_report(output: Path) -> dict: return result +def verify_retained_grader(plan: dict, trials: list[dict], tasks: dict) -> None: + """Re-derive every positive verdict with trusted code, not the frozen copy. + + `validator/verify.py` lives beside the agent's workspace and the agent runs + first, so those bytes cannot establish a grade on their own. When the frozen + exam is the repository verifier, its verdict is computed here again from the + retained answer. A custom grader has no in-tree authority to re-derive from; + that exam is reported as unverified rather than silently trusted. + """ + if plan["exam"]["grader"] != sha256(ROOT / "tools/skill_eval_verifier.py"): + for trial in trials: + if trial["verdict"] in ("PASS", "FAIL"): + trial["grader_trust"] = "frozen_copy_only" + return + for trial in trials: + if trial["verdict"] not in ("PASS", "FAIL"): + continue + checks = verify(tasks[trial["task_id"]], json.loads(trial["output"])) + if checks != trial["checks"] or checks["passed"] != (trial["verdict"] == "PASS"): + raise ValueError("Untrusted grading disagrees with the trusted grader") + trial["grader_trust"] = "reverified" + + def report(output: Path) -> dict: result = read_report(output) write_json(output / "report.json", result) diff --git a/tools/skill_eval_pr.py b/tools/skill_eval_pr.py index 5b5011676..373a40fb8 100644 --- a/tools/skill_eval_pr.py +++ b/tools/skill_eval_pr.py @@ -72,6 +72,10 @@ def prepare(repo, mapping_path, identity, output): for item in mapping["skills"]: if item["skill"] not in selected: continue + # The mapping is trusted repository data, but a path is a path: apply the + # safe-ID rule here, before building any output path from it. + if not re.fullmatch(r"[a-z0-9][a-z0-9._-]*", item["skill"]): + raise ValueError("Unsafe skill ID in the PR case mapping") entrypoint = mapping["skill_root"] + "/" + item["entrypoint"] if not blob_exists(repo, identity["head"], entrypoint): for entry in coverage: From 01bd59c4cdcbc6733db37c438ef20ce2dea3c426 Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Wed, 28 Oct 2026 12:55:49 +0800 Subject: [PATCH 4/5] fix: record unexpected termination failures and re-derive comparison authority - runner: the termination path records every failure, not only OSError/ValueError/SubprocessError. A RuntimeError from a hook or from report generation previously escaped while the last persisted state stayed running or a premature completed. - compare: an exam graded by a custom grader has no in-tree re-derivation, so its stated PASS/FAIL is moved to stated_verdict and the trial is INCONCLUSIVE. Re-grading now runs before pairs are formed, so an unverifiable grade cannot become a fix or regression outcome. - compare: exported bundle bytes are compared against each arm's stated resources, and the identical-bundle claim is re-derived from those bytes instead of trusting the freeze metadata on both sides. Agent-assisted-by: Claude Code (author) Signed-off-by: GUOGUO --- src/inferrsi/harness/runner.py | 8 ++++-- tests/test_runtime.py | 45 +++++++++++++++++++++++++++++ tests/test_skill_eval_compare.py | 49 ++++++++++++++++++++++++++++++++ tools/skill_eval_compare.py | 37 +++++++++++++++++++----- 4 files changed, 129 insertions(+), 10 deletions(-) diff --git a/src/inferrsi/harness/runner.py b/src/inferrsi/harness/runner.py index 8ba4a212d..babefe5f0 100644 --- a/src/inferrsi/harness/runner.py +++ b/src/inferrsi/harness/runner.py @@ -124,9 +124,11 @@ def terminate() -> None: "complete", state=state, verdict=result["verdict"], error=error ) write_report(run_dir) - except (OSError, ValueError, subprocess.SubprocessError) as exc: - # The run did not reach a finished state. Keep the measurement error - # visible beside the termination failure instead of replacing it. + except Exception as exc: + # The run did not reach a finished state. Every termination failure is + # recorded, not only I/O-shaped ones: the hook and report paths are + # caller-supplied, and an escaped RuntimeError would otherwise leave + # `running` (or a premature `completed`) as the last persisted state. record_status( "termination", state="termination_failed", diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 24937a8fc..7b12c56d7 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -470,3 +470,48 @@ def test_summary_ignores_a_timed_out_trial_metric(self): if __name__ == "__main__": unittest.main() + + +class TerminationTests(unittest.TestCase): + """A failure in the termination path must still leave a truthful status.""" + + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.repo = self.root / "repo" + self.repo.mkdir() + (self.repo / "source.txt").write_text("initial source\n", encoding="utf-8") + for args in ( + ("init", "--quiet"), + ("add", "source.txt"), + ("-c", "user.name=InferRSI test", "-c", "user.email=test@example.invalid", + "-c", "commit.gpgsign=false", "commit", "--quiet", "-s", "-m", "fixture"), + ): + subprocess.run(["git", "-C", str(self.repo), *args], check=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self.spec = RunSpec( + repo=str(self.repo), + baseline_command=benchmark(100), + candidate_command=benchmark(80), + repetitions=1, + validation_command=(sys.executable, "-c", "print('ok')"), + output_dir=str(self.root / "runs"), + max_minutes=1, + ) + + def test_unexpected_termination_error_is_recorded_not_escaped(self): + # Report generation fails with an error outside the documented + # OSError/ValueError contract, after `completed` was already written. + # That failure must be persisted as the terminal state instead of + # escaping while `completed` remains the last recorded status. + with patch("inferrsi.harness.runner.write_report", + side_effect=RuntimeError("report backend crashed")): + with self.assertRaises(RuntimeError): + run_with_adapters(self.spec, {"backend": "cpu", "test": True}) + run = sorted((self.root / "runs").iterdir())[-1] + status = json.loads((run / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "termination_failed") + self.assertEqual(status["phase"], "termination") + self.assertEqual(status["requested_state"], "completed") + self.assertIn("report backend crashed", status["termination_error"]) diff --git a/tests/test_skill_eval_compare.py b/tests/test_skill_eval_compare.py index 8f70d17cd..d4afbcb48 100644 --- a/tests/test_skill_eval_compare.py +++ b/tests/test_skill_eval_compare.py @@ -1044,3 +1044,52 @@ def test_answer_keys_are_rejected_by_supported_input_shape(tmp_path, extra): with pytest.raises(ValueError, match="input"): compare.prepare(config, repo, output) assert not output.exists() + + +def test_unverifiable_custom_grader_cannot_produce_a_pair_outcome(tmp_path): + # A custom grader has no in-tree authority to re-derive from, and the retained + # record is writable by the graded agent. Its stated PASS/FAIL must not become + # a fix, regression or unchanged outcome. + config, repo = configuration(tmp_path, ids=("one",)) + raw = json.loads(config.read_text()) + # Same contract, different bytes: the in-tree verifier cannot be identified + # by hash, so this exam has no trusted re-derivation. + grader = tmp_path / "custom_verifier.py" + grader.write_text( + (ROOT / "tools/skill_eval_verifier.py").read_text(encoding="utf-8") + + "\n# custom exam marker\n", + encoding="utf-8", + ) + raw["grader"] = grader.name + write_json(config, raw) + output = tmp_path / "run" + authorization = compare.prepare(config, repo, output) + result = compare.execute(output, authorization, client=FixedOutputClient(tmp_path)) + trial = result["pairs"][0]["baseline"] + assert result["counts"]["inconclusive"] == 1 + assert result["counts"]["unchanged_correct"] == 0 + assert trial["verdict"] == "INCONCLUSIVE" + assert trial["stated_verdict"] in ("PASS", "FAIL") + assert trial["grader_trust"] == "unverified" + assert trial["evaluation_error"] == "unverified_custom_grader" + + +def test_forged_identical_bundle_claim_is_rejected_against_exported_bytes(tmp_path): + # Both sides of the identical-bundle invariant come from mutable freeze + # metadata. Retargeting only the candidate revision must not make two + # different exported bundles read as the same skill content. + config, repo = configuration(tmp_path, mode="update", ids=("one",)) + output = tmp_path / "run" + compare.prepare(config, repo, output) + plan = compare.load_plan(output) + assert plan["same_bundle"] is False + assert plan["arms"]["baseline"]["bundle"]["revision"] != ( + plan["arms"]["candidate"]["bundle"]["revision"]) + freeze = output / "plan/freeze.json" + forged = json.loads(freeze.read_text(encoding="utf-8")) + forged["arms"]["candidate"]["bundle"]["revision"] = ( + forged["arms"]["baseline"]["bundle"]["revision"]) + forged["same_bundle"] = True + write_json(freeze, forged) + with pytest.raises(ValueError, match="identical-bundle"): + compare.load_plan(output) diff --git a/tools/skill_eval_compare.py b/tools/skill_eval_compare.py index 7828a409e..a4c6b00e2 100644 --- a/tools/skill_eval_compare.py +++ b/tools/skill_eval_compare.py @@ -397,8 +397,23 @@ def load_plan(output: Path) -> dict: raise ValueError("Addition baseline must not carry a bundle revision") if after is None: raise ValueError("Candidate arm requires an exported bundle") - if plan["same_bundle"] != (before is not None and before["revision"] == after["revision"]): + # The arm's bundle metadata states a revision and its resources; the exported + # bytes are the in-tree authority for those resources. Compare them so an + # edited arm cannot claim identical skill content that the export contradicts. + for role in ROLES: + bundle = plan["arms"][role]["bundle"] + if bundle is None: + continue + export = output / "plan/bundles" / role + actual = {path.relative_to(export).as_posix(): sha256(path) + for path in sorted(export.rglob("*")) if path.is_file()} + if actual != bundle["resources"]: + raise ValueError(f"Exported bundle bytes disagree with the frozen {role} arm") + same_bundle = before is not None and before["revision"] == after["revision"] + if plan["same_bundle"] != same_bundle: raise ValueError("Frozen identical-bundle flag disagrees with the frozen arms") + if same_bundle and before["resources"] != after["resources"]: + raise ValueError("Frozen identical-bundle claim disagrees with the exported bytes") frozen_exam = {"cases": "plan/tasks.json", "schema": "plan/schema.json", "grader": "validator/verify.py"} for name, path in frozen_exam.items(): @@ -611,6 +626,12 @@ def condition_summary(trials: list[dict]) -> dict: def read_report(output: Path) -> dict: plan = load_plan(output) trials = [retained_trial(output, entry) for entry in plan["schedule"]] + # A receipt is an unkeyed hash: a rewritten tree can carry a consistent one. + # Re-grading retained answers with trusted code is what makes a PASS/FAIL + # verdict independent of the artifact copy the agent could reach. This runs + # before pairs are formed so an unverifiable grade cannot become an outcome. + verify_retained_grader(plan, [t for t in trials if not t["missing"]], + {task["id"]: task for task in read_json(output / "plan/tasks.json")}) index = {(t["task_id"], t["repeat"], t["condition"]): t for t in trials} pairs = [] for entry in plan["schedule"]: @@ -641,11 +662,6 @@ def read_report(output: Path) -> dict: "observed_tokens": None, "stop_reason": None, "client_runtime": None} if error is None: status = json.loads(raw_status) - # A receipt is an unkeyed hash: a rewritten tree can carry a consistent one. - # Re-grading retained answers with trusted code is what makes a PASS/FAIL - # verdict independent of the artifact copy the agent could reach. - verify_retained_grader(plan, [t for t in trials if not t["missing"]], - {task["id"]: task for task in read_json(output / "plan/tasks.json")}) result = { "mode": plan["mode"], "skill": plan["skill"], "plan_sha256": sha256(output / "plan/freeze.json"), @@ -680,7 +696,14 @@ def verify_retained_grader(plan: dict, trials: list[dict], tasks: dict) -> None: if plan["exam"]["grader"] != sha256(ROOT / "tools/skill_eval_verifier.py"): for trial in trials: if trial["verdict"] in ("PASS", "FAIL"): - trial["grader_trust"] = "frozen_copy_only" + # No in-tree authority can re-derive this grade, and the retained + # record is writable by the graded agent, so the stated outcome is + # not evidence. Keep the raw verdict visible under its own field, + # but do not let it count as a fix, regression or unchanged pair. + trial["stated_verdict"] = trial["verdict"] + trial["verdict"] = "INCONCLUSIVE" + trial["grader_trust"] = "unverified" + trial["evaluation_error"] = "unverified_custom_grader" return for trial in trials: if trial["verdict"] not in ("PASS", "FAIL"): From d5fd02cad046d42d3631c70cbfdd9a8bfbf08814 Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Wed, 28 Oct 2026 14:38:07 +0800 Subject: [PATCH 5/5] fix: record interruption inside the termination path KeyboardInterrupt during terminate() runs outside the caller's own handler, so the run kept its last persisted state: running when the final snapshot was interrupted, a premature completed when the report was. The termination path now handles KeyboardInterrupt explicitly and records termination_failed with the requested state and any original measurement error before re-raising. Agent-assisted-by: Claude Code (author) Signed-off-by: GUOGUO --- src/inferrsi/harness/runner.py | 14 ++++++++++++++ tests/test_runtime.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/src/inferrsi/harness/runner.py b/src/inferrsi/harness/runner.py index babefe5f0..279e9883b 100644 --- a/src/inferrsi/harness/runner.py +++ b/src/inferrsi/harness/runner.py @@ -124,6 +124,20 @@ def terminate() -> None: "complete", state=state, verdict=result["verdict"], error=error ) write_report(run_dir) + except KeyboardInterrupt as exc: + # Interruption during termination is outside the caller's own + # KeyboardInterrupt handler, so the state must be persisted here: + # otherwise it stays `running` (interrupted during the final snapshot) + # or a premature `completed` (interrupted while writing the report). + record_status( + "termination", + state="termination_failed", + verdict=None, + error=error, + requested_state=state, + termination_error=f"KeyboardInterrupt: {exc or 'interrupted during termination'}", + ) + raise except Exception as exc: # The run did not reach a finished state. Every termination failure is # recorded, not only I/O-shaped ones: the hook and report paths are diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 7b12c56d7..77214ffd0 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -515,3 +515,36 @@ def test_unexpected_termination_error_is_recorded_not_escaped(self): self.assertEqual(status["phase"], "termination") self.assertEqual(status["requested_state"], "completed") self.assertIn("report backend crashed", status["termination_error"]) + + def test_interruption_while_writing_the_report_is_recorded_as_termination(self): + # Ctrl-C inside terminate() is outside the caller's own KeyboardInterrupt + # handler. `completed` was already written, so the interruption must be + # persisted as a termination failure rather than left looking finished. + with patch("inferrsi.harness.runner.write_report", side_effect=KeyboardInterrupt): + with self.assertRaises(KeyboardInterrupt): + run_with_adapters(self.spec, {"backend": "cpu", "test": True}) + run = sorted((self.root / "runs").iterdir())[-1] + status = json.loads((run / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "termination_failed") + self.assertEqual(status["phase"], "termination") + self.assertEqual(status["requested_state"], "completed") + self.assertIn("KeyboardInterrupt", status["termination_error"]) + + def test_interruption_during_final_snapshot_is_recorded_as_termination(self): + # The final snapshot runs inside terminate(). Interrupting it must not + # leave `running` as the last persisted state. + calls = [] + + def snapshot(path, exclude=None): + calls.append(path) + if len(calls) > 3: + raise KeyboardInterrupt + return {"available": True, "snapshot_hash": "fixture", "path": str(path)} + + with self.assertRaises(KeyboardInterrupt): + run_with_adapters(self.spec, {"backend": "cpu", "test": True}, snapshot=snapshot) + run = sorted((self.root / "runs").iterdir())[-1] + status = json.loads((run / "status.json").read_text(encoding="utf-8")) + self.assertEqual(status["state"], "termination_failed") + self.assertEqual(status["phase"], "termination") + self.assertFalse((run / "evaluation.json").exists())