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..279e9883b 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,82 @@ 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 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 + # 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", + 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 +170,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/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() 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_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_runtime.py b/tests/test_runtime.py index a5781fae1..77214ffd0 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,195 @@ 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() + + +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"]) + + 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()) diff --git a/tests/test_skill_eval_compare.py b/tests/test_skill_eval_compare.py index a331bd5f4..d4afbcb48 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" @@ -957,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/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..a4c6b00e2 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,58 @@ 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") + # 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(): + 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 +460,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 +480,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 +522,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, @@ -565,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"]: @@ -617,6 +684,36 @@ 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"): + # 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"): + 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: