From 5417b9e18285366acceba34a0f546388893dfca2 Mon Sep 17 00:00:00 2001 From: Kristofer Jussmann Date: Tue, 20 Oct 2026 20:29:21 +0300 Subject: [PATCH 1/2] Add frozen benchmark evidence and candidate refinement workflows --- examples/benchmark-evidence/README.md | 131 +++++ schemas/candidate-evidence.schema.json | 1 + schemas/run-record.schema.json | 5 + src/harneloop/benchmarks.py | 735 +++++++++++++++++++++++++ src/harneloop/candidate.py | 40 ++ src/harneloop/cli.py | 92 +++- src/harneloop/evidence.py | 18 +- src/harneloop/refinement.py | 159 ++++++ src/harneloop/runs.py | 52 +- tests/test_benchmark_cli.py | 59 ++ tests/test_benchmarks.py | 502 +++++++++++++++++ tests/test_evidence_integrity.py | 58 ++ tests/test_refinement.py | 90 +++ 13 files changed, 1938 insertions(+), 4 deletions(-) create mode 100644 examples/benchmark-evidence/README.md create mode 100644 src/harneloop/benchmarks.py create mode 100644 src/harneloop/refinement.py create mode 100644 tests/test_benchmark_cli.py create mode 100644 tests/test_benchmarks.py create mode 100644 tests/test_evidence_integrity.py create mode 100644 tests/test_refinement.py diff --git a/examples/benchmark-evidence/README.md b/examples/benchmark-evidence/README.md new file mode 100644 index 0000000..6974864 --- /dev/null +++ b/examples/benchmark-evidence/README.md @@ -0,0 +1,131 @@ +# Compare a harness against its baseline + +Harneloop records a benchmark protocol, binds a candidate before held-out results, +and verifies that both arms cover the same tasks and trials. Your executor runs +the target agent and your grader evaluates its artifacts. No model provider, +benchmark dataset or media dependency is installed with this workflow. + +Start with `harneloop onboard` and map the actual environment in your unit. Keep +grader answers unavailable to the solving agent. Use development tasks to build +candidates, then evaluate a frozen candidate on held-out tasks without editing it. + +## Freeze the protocol + +Save a JSON manifest with these fields. The all-zero revisions below are format +placeholders: replace them with the actual immutable source and grader commits +or `sha256:<64 hex>` checksums. Extra task fields can record prompt hashes. + +```json +{ + "source": {"name": "your-task-suite", "revision": "0000000000000000000000000000000000000000"}, + "grader": {"name": "your-grader", "revision": "0000000000000000000000000000000000000000"}, + "tasks": [ + {"task_id": "dev-1", "split": "dev"}, + {"task_id": "test-1", "split": "test"} + ], + "repeats": 1, + "seed": 17, + "executor": "your-executor", + "model": "your-exact-model-id", + "effort": "your-setting-or-provider-default", + "settings": {"temperature": 0}, + "tools": {"allowed": ["your-tool"]}, + "budget_policy": {"max_turns": 20, "timeout_seconds": 600} +} +``` + +```sh +harneloop benchmark create ./unit pilot --manifest manifest.json +harneloop benchmark bind ./unit pilot --candidate-id cand-0003 +harneloop benchmark show ./unit pilot +``` + +`bind` requires a ready or validating candidate. It freezes that candidate's file +contents and modes; it can only happen once and before any result is recorded. +Changing the overlay requires a new protocol. Protocol and result records live +under `.evolve/benchmarks/`, outside candidate-editable material and thin exports. +The seed describes your selection or execution plan; it does not make a remote +model deterministic. + +## Run both arms and capture evidence + +Create a distinct run for every held-out task, trial and arm. Trial numbers begin +at zero. Baseline runs use the current harness without a candidate; candidate +runs use the frozen overlay through your executor's normal integration. + +```sh +harneloop run start ./unit --task "Baseline test-1" --benchmark-id pilot --benchmark-task-id test-1 --benchmark-trial 0 --benchmark-arm baseline +harneloop run start ./unit --task "Candidate test-1" --candidate-id cand-0003 --benchmark-id pilot --benchmark-task-id test-1 --benchmark-trial 0 --benchmark-arm candidate +``` + +Run the executor and grader with the same model, tools, settings and budget for +both arms. Attach their outputs with `harneloop artifact add`, then finish +each run with `harneloop run finish`. Use `--help` for those commands' required +arguments. Execution success only means the executor finished; a failed grader +score is still recorded separately. + +Save each result as JSON. Copy `observed_conditions` exactly from the six manifest +fields shown below, using actual observed settings. Replace run and artifact IDs +with the IDs produced by your run. Repeat for every task, trial and arm. + +```json +{ + "observed_conditions": { + "executor": "your-executor", + "model": "your-exact-model-id", + "effort": "your-setting-or-provider-default", + "settings": {"temperature": 0}, + "tools": {"allowed": ["your-tool"]}, + "budget_policy": {"max_turns": 20, "timeout_seconds": 600} + }, + "arm": "baseline", + "task_id": "test-1", + "trial": 0, + "run_id": "run-0001", + "score": 1, + "artifact_ids": ["artifact-0001"], + "status": "scored" +} +``` + +```sh +harneloop benchmark record ./unit pilot --result result.json +harneloop benchmark compare ./unit pilot --candidate-id cand-0003 +``` + +Scores range from 0 to 1; higher is better. A scored zero is an evaluated failure. +Use `status: "failed"`, score `0` and a concrete `reason` for an execution failure. +Use `status: "unavailable"`, score `null` and a reason for an unavailable evaluation. +These rows remain in coverage counts. Optional `metrics` can contain measured +`duration_seconds`, `input_tokens`, `output_tokens` and `cost_usd`; omitted values +remain unknown, rather than becoming zero. + +Comparison rejects missing pairs, duplicate cells, reused run IDs, condition +drift, changed candidate files and changed stored artifacts. It reports per-task +deltas, wins, losses, ties and unavailable evaluations. It does not promote a +candidate. Choose promotion using the unit's normal validation process and the +scope of the evidence; a tie or small pilot does not establish an improvement. + +## Compose candidates and retain lessons + +```sh +harneloop refine compose ./unit --candidate-id cand-0001 --candidate-id cand-0002 --summary "Combine compatible improvements" +harneloop refine learn ./unit --cue "boundary inputs" --lesson "Check both sides of the boundary" --run-id run-0001 --artifact-id artifact-0001 +harneloop refine lessons ./unit --query "boundary" +``` + +Composition unions compatible file overlays from open candidates on one current +base and one plane. Conflicting bytes or modes require an explicit resolution. +It preserves source candidates and the strongest validation tier, and carries no +validation evidence into the new candidate: validate the composition afresh. + +Lessons are operator-authored observations linked to finished runs and captured +artifacts. Retrieval checks their hashes again. They are explicit context you +can select for an agent, rather than automatically injected universal rules. + +New candidate runs and evidence records fingerprint the overlay. Evidence-backed +promotion rechecks those fingerprints and captured artifact hashes; edits need +fresh evidence. Legacy records without fingerprints remain readable. Checksums +detect ordinary drift, but a local operator can rewrite records and checksums. +Harneloop does not authenticate remote execution settings, prove task isolation, +or replace an executor sandbox or independent grader. diff --git a/schemas/candidate-evidence.schema.json b/schemas/candidate-evidence.schema.json index 3a2c140..9ea027e 100644 --- a/schemas/candidate-evidence.schema.json +++ b/schemas/candidate-evidence.schema.json @@ -15,6 +15,7 @@ "enum": ["structural", "targeted", "representative", "full"] }, "candidate_base_version": { "type": ["string", "null"] }, + "candidate_sha256": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" }, "run_id": { "type": ["string", "null"] }, "artifact_id": { "type": ["string", "null"] }, "path": { "type": ["string", "null"] }, diff --git a/schemas/run-record.schema.json b/schemas/run-record.schema.json index 4ca23ec..ad25f65 100644 --- a/schemas/run-record.schema.json +++ b/schemas/run-record.schema.json @@ -14,6 +14,11 @@ "harness_version": { "type": ["string", "null"] }, "candidate_id": { "type": ["string", "null"] }, "candidate_base_version": { "type": ["string", "null"] }, + "candidate_sha256": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" }, + "benchmark_id": { "type": ["string", "null"] }, + "benchmark_task_id": { "type": ["string", "null"] }, + "benchmark_trial": { "type": ["integer", "null"], "minimum": 0 }, + "benchmark_arm": { "type": ["string", "null"], "enum": ["baseline", "candidate", null] }, "candidate_validation_tier": { "type": ["string", "null"], "enum": ["structural", "targeted", "representative", "full", null] diff --git a/src/harneloop/benchmarks.py b/src/harneloop/benchmarks.py new file mode 100644 index 0000000..0f58589 --- /dev/null +++ b/src/harneloop/benchmarks.py @@ -0,0 +1,735 @@ +from __future__ import annotations + +import hashlib +import json +import math +import re +from pathlib import Path +from typing import Any + +from .candidate import candidate_digest +from .errors import HarneloopError +from .locking import file_lock, harness_lock_path +from .runs import VALID_RUN_STATUSES +from .versioning import ensure_unit, hash_file +from .yamlio import read_yaml, write_yaml + + +_BENCHMARK_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +_IMMUTABLE_REVISION = re.compile(r"^(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64}|sha256:[0-9a-fA-F]{64})$") +_RUN_ID = re.compile(r"^run-[A-Za-z0-9._-]{1,64}$") +_CONDITION_FIELDS = ("executor", "model", "effort", "settings", "tools", "budget_policy") +_METRIC_FIELDS = ("duration_seconds", "input_tokens", "output_tokens", "cost_usd") + + +def _benchmark_dir(unit_root: Path, benchmark_id: str) -> Path: + if not _BENCHMARK_ID.fullmatch(benchmark_id) or benchmark_id in {".", ".."}: + raise HarneloopError(f"Invalid benchmark ID: {benchmark_id!r}") + return unit_root / ".evolve" / "benchmarks" / benchmark_id + + +def _safe_unit_path(unit_root: Path, path: Path, label: str) -> Path: + """Resolve a unit path while rejecting symlinks and escapes at every component.""" + unit_root = unit_root.resolve() + try: + relative = path.relative_to(unit_root) + except ValueError: + raise HarneloopError(f"{label} path is outside the harness unit") from None + if any(part in {"", ".", ".."} for part in relative.parts): + raise HarneloopError(f"{label} path contains an unsafe component") + cursor = unit_root + for part in relative.parts: + cursor = cursor / part + if cursor.is_symlink(): + raise HarneloopError(f"{label} path cannot contain symlinks: {cursor}") + resolved = path.resolve() + if not resolved.is_relative_to(unit_root): + raise HarneloopError(f"{label} path escapes the harness unit") + return resolved + + +def _validate_immutable_revision(value: Any, label: str) -> None: + if not isinstance(value, str) or not _IMMUTABLE_REVISION.fullmatch(value): + raise HarneloopError( + f"{label} revision must be a 40/64-character commit SHA or sha256:<64 hex> checksum" + ) + + +def _json_safe(value: Any, path: str = "value") -> None: + if value is None or isinstance(value, (str, bool, int)): + return + if isinstance(value, float): + if not math.isfinite(value): + raise HarneloopError(f"{path} must contain only finite numbers") + return + if isinstance(value, list): + for index, item in enumerate(value): + _json_safe(item, f"{path}[{index}]") + return + if isinstance(value, dict): + for key, item in value.items(): + if not isinstance(key, str): + raise HarneloopError(f"{path} mapping keys must be strings") + _json_safe(item, f"{path}.{key}") + return + raise HarneloopError(f"{path} contains unsupported value type: {type(value).__name__}") + + +def _canonical_hash(data: dict[str, Any]) -> str: + _json_safe(data) + payload = json.dumps( + data, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _conditions_from(manifest: dict[str, Any]) -> dict[str, Any]: + return {key: manifest[key] for key in _CONDITION_FIELDS} + + +def _validate_manifest(manifest: dict[str, Any]) -> dict[str, Any]: + if not isinstance(manifest, dict): + raise HarneloopError("Benchmark manifest must be a mapping") + required = { + "source", + "tasks", + "repeats", + "seed", + *_CONDITION_FIELDS, + "grader", + } + missing = sorted(required - manifest.keys()) + if missing: + raise HarneloopError(f"Benchmark manifest is missing required fields: {', '.join(missing)}") + _json_safe(manifest, "manifest") + + source = manifest["source"] + if ( + not isinstance(source, dict) + or not _nonempty_text(source.get("name")) + or not _nonempty_text(source.get("revision")) + ): + raise HarneloopError("Benchmark source requires a nonempty name and immutable revision") + _validate_immutable_revision(source["revision"], "Benchmark source") + grader = manifest["grader"] + if ( + not isinstance(grader, dict) + or not _nonempty_text(grader.get("name")) + or not _nonempty_text(grader.get("revision")) + ): + raise HarneloopError("Benchmark grader requires a nonempty name and immutable revision") + _validate_immutable_revision(grader["revision"], "Benchmark grader") + repeats = manifest["repeats"] + if isinstance(repeats, bool) or not isinstance(repeats, int) or repeats < 1: + raise HarneloopError("Benchmark repeats must be a positive integer") + if manifest["seed"] is None or isinstance(manifest["seed"], (dict, list)): + raise HarneloopError("Benchmark seed must be a scalar value") + tasks = manifest["tasks"] + if not isinstance(tasks, list) or not tasks: + raise HarneloopError("Benchmark tasks must be a nonempty list") + seen: set[str] = set() + splits: set[str] = set() + for task in tasks: + if not isinstance(task, dict) or not _nonempty_text(task.get("task_id")): + raise HarneloopError("Each benchmark task requires a nonempty task_id") + task_id = task["task_id"] + if task_id in seen: + raise HarneloopError(f"Duplicate benchmark task ID: {task_id}") + seen.add(task_id) + split = task.get("split") + if split not in {"dev", "test"}: + raise HarneloopError(f"Benchmark task `{task_id}` split must be `dev` or `test`") + splits.add(split) + if splits != {"dev", "test"}: + raise HarneloopError("Benchmark manifest must include both dev and test tasks") + + for field in ("executor", "model", "effort"): + if not _nonempty_text(manifest[field]): + raise HarneloopError(f"Benchmark {field} must be nonempty") + for field in ("settings", "tools", "budget_policy"): + if not isinstance(manifest[field], (dict, list)): + raise HarneloopError(f"Benchmark {field} must be a mapping or list") + if not manifest["budget_policy"]: + raise HarneloopError("Benchmark budget_policy must describe the fixed execution budget") + binding = manifest.get("candidate_binding") + if binding is not None and ( + not isinstance(binding, dict) + or set(binding) != {"candidate_id", "candidate_sha256"} + or not _nonempty_text(binding.get("candidate_id")) + or not isinstance(binding.get("candidate_sha256"), str) + or not re.fullmatch(r"[a-fA-F0-9]{64}", binding["candidate_sha256"]) + ): + raise HarneloopError("Benchmark candidate_binding is invalid") + # Store a detached JSON-compatible copy so later caller mutation cannot alter the frozen record. + return json.loads(json.dumps(manifest, ensure_ascii=False, allow_nan=False)) + + +def _nonempty_text(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) + + +def create_benchmark( + unit_root: Path, benchmark_id: str, manifest: dict[str, Any] +) -> dict[str, Any]: + """Freeze a benchmark design and its acting-condition fingerprints. + + Fingerprints provide integrity/provenance checks; they are not signatures or proof + that a remote model actually used the attested executor and settings. + """ + unit_root = unit_root.resolve() + ensure_unit(unit_root) + root = _safe_unit_path(unit_root, _benchmark_dir(unit_root, benchmark_id), "Benchmark") + if isinstance(manifest, dict) and "candidate_binding" in manifest: + raise HarneloopError("Candidate binding must be created with benchmark bind after dev refinement") + frozen = _validate_manifest(manifest) + record = { + "schema_version": "0.1", + "id": benchmark_id, + **frozen, + "manifest_sha256": _canonical_hash(frozen), + "conditions_sha256": _canonical_hash(_conditions_from(frozen)), + } + with file_lock(harness_lock_path(unit_root, f"benchmark-{benchmark_id}")): + if root.exists(): + raise HarneloopError(f"Benchmark manifest is already frozen: {benchmark_id}") + root.mkdir(parents=True, exist_ok=False) + write_yaml(root / "manifest.yaml", record) + return record + + +def read_benchmark(unit_root: Path, benchmark_id: str) -> dict[str, Any]: + unit_root = unit_root.resolve() + ensure_unit(unit_root) + root = _safe_unit_path(unit_root, _benchmark_dir(unit_root, benchmark_id), "Benchmark") + path = _safe_unit_path(unit_root, root / "manifest.yaml", "Benchmark manifest") + if not path.is_file(): + raise HarneloopError(f"Benchmark does not exist: {benchmark_id}") + record = read_yaml(path) + if record.get("id") != benchmark_id: + raise HarneloopError(f"Frozen benchmark ID mismatch: {benchmark_id}") + body = { + key: value + for key, value in record.items() + if key not in {"manifest_sha256", "conditions_sha256"} + } + body.pop("schema_version", None) + body.pop("id", None) + if record.get("manifest_sha256") != _canonical_hash(body): + raise HarneloopError(f"Frozen benchmark manifest hash mismatch: {benchmark_id}") + _validate_manifest(body) + if record.get("conditions_sha256") != _canonical_hash(_conditions_from(body)): + raise HarneloopError(f"Frozen benchmark conditions hash mismatch: {benchmark_id}") + return record + + +def bind_benchmark_candidate( + unit_root: Path, benchmark_id: str, candidate_id: str +) -> dict[str, Any]: + """Freeze one candidate identity before any heldout result is recorded.""" + unit_root = unit_root.resolve() + ensure_unit(unit_root) + root = _safe_unit_path(unit_root, _benchmark_dir(unit_root, benchmark_id), "Benchmark") + with file_lock(harness_lock_path(unit_root, f"benchmark-{benchmark_id}")): + record = read_benchmark(unit_root, benchmark_id) + if record.get("candidate_binding") is not None: + raise HarneloopError(f"Benchmark candidate is already bound: {benchmark_id}") + results = _safe_unit_path(unit_root, _result_dir(root), "Benchmark results") + if results.exists() and any(results.iterdir()): + raise HarneloopError("Candidate must be bound before any heldout benchmark rows are recorded") + digest = candidate_digest(unit_root, candidate_id) + candidate = read_yaml(unit_root / "candidates" / candidate_id / "candidate.yaml") + if candidate.get("id") != candidate_id: + raise HarneloopError(f"Candidate identity mismatch: {candidate_id}") + if candidate.get("status") not in {"ready", "validating"}: + raise HarneloopError("Mark the dev-refined candidate ready before benchmark binding") + record["candidate_binding"] = { + "candidate_id": candidate_id, + "candidate_sha256": digest, + } + body = { + key: value + for key, value in record.items() + if key not in {"manifest_sha256", "conditions_sha256", "schema_version", "id"} + } + record["manifest_sha256"] = _canonical_hash(body) + path = _safe_unit_path(unit_root, root / "manifest.yaml", "Benchmark manifest") + write_yaml(path, record) + return record + + +def _result_dir(root: Path) -> Path: + return root / "results" + + +def _read_safe_run(unit_root: Path, run_id: str) -> dict[str, Any]: + if not isinstance(run_id, str) or not _RUN_ID.fullmatch(run_id): + raise HarneloopError(f"Invalid Harneloop run ID: {run_id!r}") + path = _safe_unit_path( + unit_root, + unit_root / "runtime" / "runs" / run_id / "run.yaml", + "Harneloop run record", + ) + if not path.is_file(): + raise HarneloopError(f"Run does not exist: {run_id}") + return read_yaml(path) + + +def _row_key(arm: str, task_id: str, trial: int) -> str: + return _canonical_hash({"arm": arm, "task_id": task_id, "trial": trial}) + + +def _validate_finite_score(score: Any) -> float: + if isinstance(score, bool) or not isinstance(score, (int, float)) or not math.isfinite(score): + raise HarneloopError("Benchmark score must be a finite numeric value from 0 to 1") + if not 0 <= score <= 1: + raise HarneloopError("Benchmark score must be between 0 and 1 inclusive") + return float(score) + + +def _validate_metrics(metrics: dict[str, Any] | None) -> dict[str, int | float]: + if metrics is None: + return {} + if not isinstance(metrics, dict): + raise HarneloopError("Benchmark metrics must be a mapping") + unknown = sorted(set(metrics) - set(_METRIC_FIELDS)) + if unknown: + raise HarneloopError(f"Unsupported benchmark metrics: {', '.join(unknown)}") + normalized: dict[str, int | float] = {} + for key, value in metrics.items(): + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + ): + raise HarneloopError(f"Benchmark metric `{key}` must be finite and nonnegative") + if key in {"input_tokens", "output_tokens"} and not isinstance(value, int): + raise HarneloopError(f"Benchmark metric `{key}` must be an integer token count") + normalized[key] = value + return normalized + + +def _artifact_records( + unit_root: Path, + run_id: str, + run: dict[str, Any], + artifact_ids: list[str], +) -> list[dict[str, Any]]: + if not isinstance(run_id, str) or not _RUN_ID.fullmatch(run_id): + raise HarneloopError(f"Invalid Harneloop run ID: {run_id!r}") + if ( + not isinstance(artifact_ids, list) + or not artifact_ids + or any(not _nonempty_text(item) for item in artifact_ids) + or len(set(artifact_ids)) != len(artifact_ids) + ): + raise HarneloopError("Benchmark result must reference one or more unique stored artifacts") + artifacts = run.get("artifacts") or [] + result: list[dict[str, Any]] = [] + for artifact_id in artifact_ids: + artifact = next((item for item in artifacts if item.get("id") == artifact_id), None) + if artifact is None: + raise HarneloopError(f"Artifact `{artifact_id}` is not attached to run `{run_id}`") + stored_path = artifact.get("stored_path") + if not _nonempty_text(stored_path): + raise HarneloopError(f"Artifact `{artifact_id}` has no stored path") + stored = Path(stored_path) + if stored.is_absolute() or ".." in stored.parts: + raise HarneloopError(f"Artifact `{artifact_id}` has an unsafe stored path") + path = _safe_unit_path(unit_root, unit_root / stored, "Run artifact") + expected_root = _safe_unit_path( + unit_root, unit_root / "runtime" / "artifacts" / run_id, "Run artifact directory" + ) + if not path.is_relative_to(expected_root) or not path.is_file(): + raise HarneloopError( + f"Artifact `{artifact_id}` is missing or outside its run artifact directory" + ) + actual_hash = hash_file(path) + if artifact.get("sha256") != actual_hash: + raise HarneloopError(f"Artifact `{artifact_id}` hash mismatch") + result.append( + { + "id": artifact_id, + "stored_path": path.relative_to(unit_root).as_posix(), + "sha256": actual_hash, + } + ) + return result + + +def record_benchmark_result( + unit_root: Path, + benchmark_id: str, + *, + observed_conditions: dict[str, Any], + arm: str, + task_id: str, + trial: int, + run_id: str, + score: int | float | None, + artifact_ids: list[str], + status: str = "scored", + reason: str | None = None, + metrics: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Record an externally graded row tied to a finished Harneloop run. + + `observed_conditions` is operator-attested provenance, not cryptographic proof + of a remote model's identity or behavior. + """ + unit_root = unit_root.resolve() + ensure_unit(unit_root) + root = _benchmark_dir(unit_root, benchmark_id) + benchmark = read_benchmark(unit_root, benchmark_id) + if arm not in {"baseline", "candidate"}: + raise HarneloopError("Benchmark result arm must be `baseline` or `candidate`") + if not _nonempty_text(task_id): + raise HarneloopError("Benchmark task_id must be nonempty") + repeats = benchmark["repeats"] + if isinstance(trial, bool) or not isinstance(trial, int) or not 0 <= trial < repeats: + raise HarneloopError(f"Benchmark trial must be an integer from 0 to {repeats - 1}") + task = next((item for item in benchmark["tasks"] if item["task_id"] == task_id), None) + if task is None or task["split"] != "test": + raise HarneloopError( + f"Benchmark results can only be recorded for a frozen test split task: {task_id}" + ) + binding = benchmark.get("candidate_binding") + if not isinstance(binding, dict): + raise HarneloopError("Bind the candidate before recording any heldout benchmark rows") + if status not in {"scored", "failed", "unavailable"}: + raise HarneloopError("Benchmark result status must be `scored`, `failed`, or `unavailable`") + if status == "scored": + normalized_score: float | None = _validate_finite_score(score) + if reason: + raise HarneloopError("A scored benchmark row cannot include an unavailable reason") + elif status == "failed": + normalized_score = _validate_finite_score(score) + if normalized_score != 0: + raise HarneloopError("A failed benchmark row must have score 0") + if not _nonempty_text(reason): + raise HarneloopError("Failed benchmark rows require a concrete failure reason") + else: + if score is not None or not _nonempty_text(reason): + raise HarneloopError("Unavailable rows require a null score and a concrete reason") + normalized_score = None + + if ( + not isinstance(observed_conditions, dict) + or set(observed_conditions) != set(_CONDITION_FIELDS) + ): + raise HarneloopError( + f"Observed conditions must contain exactly: {', '.join(_CONDITION_FIELDS)}" + ) + actual_conditions_hash = _canonical_hash(observed_conditions) + if actual_conditions_hash != benchmark["conditions_sha256"]: + raise HarneloopError( + "Observed conditions do not match the frozen benchmark settings and budget policy" + ) + run = _read_safe_run(unit_root, run_id) + if ( + run.get("benchmark_id") != benchmark_id + or run.get("benchmark_task_id") != task_id + or run.get("benchmark_trial") != trial + or run.get("benchmark_arm") != arm + ): + raise HarneloopError("Benchmark result does not match the run's frozen task/trial identity") + if run.get("status") not in VALID_RUN_STATUSES - {"running"}: + raise HarneloopError(f"Benchmark result requires a finished Harneloop run: {run_id}") + if run.get("status") != "succeeded" and status == "scored": + raise HarneloopError("A failed or stopped Harneloop run cannot be reported as scored") + if arm == "baseline" and run.get("candidate_id") is not None: + raise HarneloopError("Baseline benchmark rows must reference a run without a candidate") + if arm == "candidate": + if run.get("candidate_id") != binding["candidate_id"]: + raise HarneloopError("Candidate benchmark run does not match the preregistered candidate") + current_digest = candidate_digest(unit_root, binding["candidate_id"]) + if run.get("candidate_sha256") != binding["candidate_sha256"] or current_digest != binding["candidate_sha256"]: + raise HarneloopError("Candidate overlay changed after benchmark binding or run start") + artifacts = _artifact_records(unit_root, run_id, run, artifact_ids) + normalized_metrics = _validate_metrics(metrics) + + key_hash = _row_key(arm, task_id, trial) + row = { + "schema_version": "0.1", + "benchmark_id": benchmark_id, + "manifest_sha256": benchmark["manifest_sha256"], + "conditions_sha256": benchmark["conditions_sha256"], + "arm": arm, + "task_id": task_id, + "trial": trial, + "run_id": run_id, + "candidate_id": run.get("candidate_id"), + "candidate_sha256": run.get("candidate_sha256"), + "harness_version": run.get("harness_version"), + "candidate_base_version": run.get("candidate_base_version"), + "status": status, + "score": normalized_score, + "reason": reason.strip() if isinstance(reason, str) and reason.strip() else None, + "observed_conditions": observed_conditions, + "artifacts": artifacts, + "metrics": normalized_metrics, + } + row["record_sha256"] = _canonical_hash(row) + destination = _result_dir(root) / f"{arm}-{key_hash}.yaml" + with file_lock(harness_lock_path(unit_root, f"benchmark-{benchmark_id}")): + # Re-read under the shared benchmark lock so concurrent writers cannot replace a row. + latest = read_benchmark(unit_root, benchmark_id) + if latest.get("manifest_sha256") != benchmark["manifest_sha256"]: + raise HarneloopError("Benchmark manifest changed while recording a result") + results_root = _safe_unit_path(unit_root, _result_dir(root), "Benchmark results") + if results_root.exists(): + existing = _read_rows(unit_root, root, latest) + if any(item.get("run_id") == run_id for item in existing.values()): + raise HarneloopError("A Harneloop run can back only one benchmark task/trial row") + if destination.exists(): + raise HarneloopError( + f"Benchmark result already exists for {arm}/{task_id}/trial-{trial}" + ) + _safe_unit_path(unit_root, destination, "Benchmark result") + destination.parent.mkdir(parents=True, exist_ok=True) + write_yaml(destination, row) + return row + + +def _read_rows( + unit_root: Path, root: Path, benchmark: dict[str, Any] +) -> dict[tuple[str, str, int], dict[str, Any]]: + rows: dict[tuple[str, str, int], dict[str, Any]] = {} + results_root = _safe_unit_path(unit_root, _result_dir(root), "Benchmark results") + for path in sorted(results_root.glob("*.yaml")): + _safe_unit_path(unit_root, path, "Benchmark result") + row = read_yaml(path) + record_hash = row.get("record_sha256") + body = {key: value for key, value in row.items() if key != "record_sha256"} + if record_hash != _canonical_hash(body): + raise HarneloopError(f"Benchmark result record hash mismatch: {path.name}") + if ( + row.get("benchmark_id") != benchmark["id"] + or row.get("manifest_sha256") != benchmark["manifest_sha256"] + ): + raise HarneloopError( + f"Benchmark result references a different frozen manifest: {path.name}" + ) + if row.get("conditions_sha256") != benchmark["conditions_sha256"]: + raise HarneloopError(f"Benchmark result conditions hash mismatch: {path.name}") + if _canonical_hash(row.get("observed_conditions")) != benchmark["conditions_sha256"]: + raise HarneloopError(f"Benchmark result observed conditions mismatch: {path.name}") + arm = row.get("arm") + task_id = row.get("task_id") + trial = row.get("trial") + if arm not in {"baseline", "candidate"} or not _nonempty_text(task_id): + raise HarneloopError(f"Invalid benchmark result identity: {path.name}") + if ( + isinstance(trial, bool) + or not isinstance(trial, int) + or not 0 <= trial < benchmark["repeats"] + ): + raise HarneloopError(f"Invalid benchmark result trial: {path.name}") + if row.get("status") == "scored": + _validate_finite_score(row.get("score")) + if row.get("reason"): + raise HarneloopError( + f"Scored benchmark result has an unavailable reason: {path.name}" + ) + elif row.get("status") == "failed": + if ( + _validate_finite_score(row.get("score")) != 0 + or not _nonempty_text(row.get("reason")) + ): + raise HarneloopError(f"Invalid failed benchmark result: {path.name}") + elif row.get("status") == "unavailable": + if row.get("score") is not None or not _nonempty_text(row.get("reason")): + raise HarneloopError(f"Invalid unavailable benchmark result: {path.name}") + else: + raise HarneloopError(f"Invalid benchmark result status: {path.name}") + _validate_metrics(row.get("metrics") or {}) + key = (arm, task_id, trial) + if key in rows: + raise HarneloopError(f"Duplicate benchmark result key: {key}") + if any(existing.get("run_id") == row.get("run_id") for existing in rows.values()): + raise HarneloopError("A Harneloop run can back only one benchmark task/trial row") + run = _read_safe_run(unit_root, row.get("run_id")) + if run.get("status") not in VALID_RUN_STATUSES - {"running"}: + raise HarneloopError(f"Benchmark result references an unfinished run: {row['run_id']}") + if run.get("status") != "succeeded" and row.get("status") == "scored": + raise HarneloopError(f"Failed or stopped run is marked scored: {row['run_id']}") + if run.get("candidate_id") != row.get("candidate_id"): + raise HarneloopError( + f"Benchmark result candidate linkage changed for run {row['run_id']}" + ) + if run.get("candidate_sha256") != row.get("candidate_sha256"): + raise HarneloopError(f"Benchmark run candidate digest changed for run {row['run_id']}") + if ( + run.get("harness_version") != row.get("harness_version") + or run.get("candidate_base_version") != row.get("candidate_base_version") + ): + raise HarneloopError(f"Benchmark run version linkage changed for run {row['run_id']}") + if ( + run.get("benchmark_id") != benchmark["id"] + or run.get("benchmark_task_id") != task_id + or run.get("benchmark_trial") != trial + or run.get("benchmark_arm") != arm + ): + raise HarneloopError(f"Benchmark run task/trial linkage changed for run {row['run_id']}") + if arm == "candidate": + binding = benchmark.get("candidate_binding") + if ( + not isinstance(binding, dict) + or run.get("candidate_id") != binding["candidate_id"] + or run.get("candidate_sha256") != binding["candidate_sha256"] + or candidate_digest(unit_root, binding["candidate_id"]) != binding["candidate_sha256"] + ): + raise HarneloopError(f"Candidate content no longer matches benchmark binding: {row['run_id']}") + artifacts = row.get("artifacts") + if not isinstance(artifacts, list) or any(not isinstance(item, dict) for item in artifacts): + raise HarneloopError(f"Invalid benchmark artifact references: {path.name}") + artifact_ids = [item.get("id") for item in artifacts] + checked_artifacts = _artifact_records(unit_root, row["run_id"], run, artifact_ids) + if checked_artifacts != artifacts: + raise HarneloopError( + f"Benchmark evidence artifact linkage changed for run {row['run_id']}" + ) + rows[key] = row + return rows + + +def compare_benchmark(unit_root: Path, benchmark_id: str, *, candidate_id: str) -> dict[str, Any]: + """Summarize paired heldout scores; this does not run a provider or grader.""" + unit_root = unit_root.resolve() + ensure_unit(unit_root) + root = _benchmark_dir(unit_root, benchmark_id) + benchmark = read_benchmark(unit_root, benchmark_id) + if not _nonempty_text(candidate_id): + raise HarneloopError("Candidate ID must be nonempty") + binding = benchmark.get("candidate_binding") + if not isinstance(binding, dict) or binding.get("candidate_id") != candidate_id: + raise HarneloopError("Comparison candidate does not match the preregistered benchmark binding") + if candidate_digest(unit_root, candidate_id) != binding.get("candidate_sha256"): + raise HarneloopError("Candidate overlay changed after benchmark binding") + tasks = [item["task_id"] for item in benchmark["tasks"] if item["split"] == "test"] + expected = { + (arm, task_id, trial) + for arm in ("baseline", "candidate") + for task_id in tasks + for trial in range(benchmark["repeats"]) + } + with file_lock(harness_lock_path(unit_root, f"benchmark-{benchmark_id}")): + benchmark = read_benchmark(unit_root, benchmark_id) + rows = _read_rows(unit_root, root, benchmark) + unexpected = set(rows) - expected + if unexpected: + raise HarneloopError( + "Benchmark contains results outside the frozen test design: " + f"{sorted(unexpected)}" + ) + missing = expected - set(rows) + if missing: + raise HarneloopError( + f"Benchmark result set is incomplete; missing {len(missing)} arm/task/trial rows" + ) + for key, row in rows.items(): + arm = key[0] + if arm == "baseline" and row.get("candidate_id") is not None: + raise HarneloopError("Baseline benchmark row references a candidate-linked run") + if arm == "candidate" and row.get("candidate_id") != candidate_id: + raise HarneloopError("Candidate benchmark row references a different candidate") + baseline_versions = { + rows[("baseline", task_id, trial)].get("harness_version") + for task_id in tasks + for trial in range(benchmark["repeats"]) + } + candidate_base_versions = { + rows[("candidate", task_id, trial)].get("candidate_base_version") + for task_id in tasks + for trial in range(benchmark["repeats"]) + } + if len(baseline_versions) != 1 or candidate_base_versions != baseline_versions: + raise HarneloopError( + "Baseline runs and candidate runs do not share the same harness base version" + ) + + expected_pairs = len(tasks) * benchmark["repeats"] + arms: dict[str, dict[str, int]] = {} + for arm in ("baseline", "candidate"): + arm_rows = [row for (row_arm, _, _), row in rows.items() if row_arm == arm] + unavailable = sum(row["status"] == "unavailable" for row in arm_rows) + failed = sum(row["status"] == "failed" or row.get("score") == 0 for row in arm_rows) + arms[arm] = { + "expected": expected_pairs, + "scored": len(arm_rows) - unavailable, + "failed": failed, + "unavailable": unavailable, + } + + paired = {"wins": 0, "losses": 0, "ties": 0, "scored": 0, "unavailable": 0} + per_task: dict[str, dict[str, Any]] = {} + deltas: list[dict[str, Any]] = [] + for task_id in tasks: + task_result = {"wins": 0, "losses": 0, "ties": 0, "unavailable": 0, "trials": []} + for trial in range(benchmark["repeats"]): + baseline = rows[("baseline", task_id, trial)] + candidate = rows[("candidate", task_id, trial)] + if baseline["status"] == "unavailable" or candidate["status"] == "unavailable": + paired["unavailable"] += 1 + task_result["unavailable"] += 1 + task_result["trials"].append({"trial": trial, "outcome": "unavailable"}) + continue + delta = candidate["score"] - baseline["score"] + paired["scored"] += 1 + if delta > 0: + outcome = "win" + paired["wins"] += 1 + task_result["wins"] += 1 + elif delta < 0: + outcome = "loss" + paired["losses"] += 1 + task_result["losses"] += 1 + else: + outcome = "tie" + paired["ties"] += 1 + task_result["ties"] += 1 + delta_row = { + "task_id": task_id, + "trial": trial, + "baseline_score": baseline["score"], + "candidate_score": candidate["score"], + "delta": delta, + } + deltas.append(delta_row) + task_result["trials"].append({"trial": trial, "outcome": outcome, **delta_row}) + per_task[task_id] = task_result + + metric_observations: dict[str, list[dict[str, Any]]] = {field: [] for field in _METRIC_FIELDS} + for (arm, task_id, trial), row in sorted(rows.items()): + for field, value in (row.get("metrics") or {}).items(): + metric_observations[field].append( + {"arm": arm, "task_id": task_id, "trial": trial, "value": value} + ) + metric_observations = {field: values for field, values in metric_observations.items() if values} + + return { + "benchmark_id": benchmark_id, + "manifest_sha256": benchmark["manifest_sha256"], + "conditions_sha256": benchmark["conditions_sha256"], + "candidate_id": candidate_id, + "candidate_sha256": binding["candidate_sha256"], + "heldout_task_ids": tasks, + "repeats": benchmark["repeats"], + "expected_pairs": expected_pairs, + "arms": arms, + "paired": paired, + "per_task": per_task, + "paired_deltas": deltas, + "metric_observations": metric_observations, + "score_contract": "finite numeric values from 0 to 1; higher is better", + "provenance_note": ( + "Observed executor/model conditions and external grades are operator-attested. " + "Hashes identify content and detect unmatched edits, but a workspace operator can " + "rewrite records and recompute hashes; they do not authenticate graders or prove " + "remote model identity." + ), + } diff --git a/src/harneloop/candidate.py b/src/harneloop/candidate.py index 90678f9..0fdd987 100644 --- a/src/harneloop/candidate.py +++ b/src/harneloop/candidate.py @@ -1,5 +1,8 @@ from __future__ import annotations +import hashlib +import json +import stat from pathlib import Path from .errors import HarneloopError @@ -47,6 +50,43 @@ def read_candidate(unit_root: Path, candidate_id: str) -> dict[str, object]: return read_yaml(candidate_path(unit_root, candidate_id) / "candidate.yaml") +def candidate_digest(unit_root: Path, candidate_id: str) -> str: + """Fingerprint the candidate overlay files that define its executable changes.""" + unit_root = unit_root.resolve() + root = candidate_path(unit_root, candidate_id) + if (unit_root / "candidates").is_symlink() or root.is_symlink() or not root.resolve().is_relative_to(unit_root): + raise HarneloopError(f"Candidate path escapes the harness unit: {candidate_id}") + if (root / "candidate.yaml").is_symlink(): + raise HarneloopError(f"Candidate metadata cannot be a symlink: {candidate_id}") + changes = root / "changes" + if changes.is_symlink(): + raise HarneloopError(f"Candidate changes path cannot be a symlink: {candidate_id}") + entries: list[dict[str, object]] = [] + if changes.exists(): + for path in sorted(changes.rglob("*")): + if path.is_symlink(): + raise HarneloopError(f"Candidate overlay contains a symlink: {path}") + if not path.is_file(): + continue + resolved = path.resolve() + if not resolved.is_relative_to(changes.resolve()): + raise HarneloopError(f"Candidate overlay escapes its changes directory: {path}") + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + entries.append( + { + "path": path.relative_to(changes).as_posix(), + "sha256": digest.hexdigest(), + "size": path.stat().st_size, + "mode": stat.S_IMODE(path.stat().st_mode), + } + ) + payload = json.dumps(entries, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + def list_candidates(unit_root: Path, include_closed: bool = True) -> list[dict[str, object]]: candidates_root = unit_root.resolve() / "candidates" if not candidates_root.exists(): diff --git a/src/harneloop/cli.py b/src/harneloop/cli.py index 78edfef..c56e6b2 100644 --- a/src/harneloop/cli.py +++ b/src/harneloop/cli.py @@ -17,6 +17,8 @@ from .diagnostics import run_doctor from .environment import ENVIRONMENT_MODES, INTERACTION_MODES, connect_environment, render_environment_status from .errors import HarneloopError +from .benchmarks import create_benchmark, read_benchmark, record_benchmark_result, compare_benchmark +from .refinement import compose_candidates, add_lesson, list_lessons from .evidence import add_evidence from .intake import ( ACKNOWLEDGEMENT_BASES, @@ -64,6 +66,45 @@ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="harneloop", description="Harneloop harness-unit lifecycle engine") subparsers = parser.add_subparsers(dest="command") + benchmark = subparsers.add_parser( + "benchmark", help="Freeze external benchmark conditions and compare artifact-backed results", + description="External agents and graders produce results; Harneloop checks provenance and paired coverage. Scores are 0 to 1, higher is better.", + epilog="Workflow and JSON format: https://github.com/OpenRootdev/Harneloop/tree/main/examples/benchmark-evidence", + ) + benchmark_commands = benchmark.add_subparsers(dest="benchmark_command", required=True) + for name, help_text in [ + ("create", "Freeze a new JSON manifest; refuses to overwrite"), + ("show", "Show a verified frozen manifest"), + ("bind", "Freeze the candidate identity and content before held-out results"), + ("record", "Attach one JSON result to a finished run and stored artifacts"), + ("compare", "Verify complete matched held-out trials and report paired outcomes"), + ]: + command = benchmark_commands.add_parser(name, help=help_text) + add_unit_reference(command) + command.add_argument("benchmark_id", help="Unit-local benchmark identifier") + if name == "create": + command.add_argument("--manifest", type=Path, required=True, help="JSON manifest with source/tasks/conditions/grader") + elif name == "record": + command.add_argument("--result", type=Path, required=True, help="JSON row including observed_conditions, arm, task_id, trial, run_id, score and artifact_ids") + elif name in {"bind", "compare"}: + command.add_argument("--candidate-id", required=True) + + refine = subparsers.add_parser("refine", help="Compose candidate overlays and retrieve evidence-linked lessons") + refine_commands = refine.add_subparsers(dest="refine_command", required=True) + compose = refine_commands.add_parser("compose", help="Combine compatible candidates; fresh validation is required") + add_unit_reference(compose) + compose.add_argument("--candidate-id", action="append", required=True) + compose.add_argument("--summary", required=True) + learn = refine_commands.add_parser("learn", help="Store an observation linked to a finished run artifact") + add_unit_reference(learn) + learn.add_argument("--cue", required=True) + learn.add_argument("--lesson", required=True) + learn.add_argument("--run-id", required=True) + learn.add_argument("--artifact-id", required=True) + lessons = refine_commands.add_parser("lessons", help="Retrieve lessons and recheck their source artifacts") + add_unit_reference(lessons) + lessons.add_argument("--query", default="") + onboard_parser = subparsers.add_parser("onboard", help="Print the new-harness onboarding checklist") onboard_parser.add_argument("--format", choices=["markdown", "json"], default="markdown") @@ -254,6 +295,10 @@ def build_parser() -> argparse.ArgumentParser: run_start.add_argument("--task", required=True) run_start.add_argument("--candidate-id") run_start.add_argument("--attempt-id") + run_start.add_argument("--benchmark-id") + run_start.add_argument("--benchmark-task-id") + run_start.add_argument("--benchmark-trial", type=int) + run_start.add_argument("--benchmark-arm", choices=["baseline", "candidate"]) run_finish = run_subparsers.add_parser("finish", help="Finish a run record") add_unit_reference(run_finish) @@ -303,6 +348,16 @@ def coerce_cli_value(value: str) -> object: return value +def read_json_object(path: Path) -> dict: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, ValueError) as error: + raise HarneloopError(f"Cannot read JSON input: {error}") from error + if not isinstance(data, dict): + raise HarneloopError("JSON input must be an object") + return data + + def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) @@ -319,6 +374,37 @@ def main(argv: list[str] | None = None) -> int: if hasattr(args, "unit") and args.command != "units": args.unit = resolve_unit_reference(args.unit) + if args.command == "benchmark": + if args.benchmark_command == "create": + data = create_benchmark(args.unit, args.benchmark_id, read_json_object(args.manifest)) + elif args.benchmark_command == "show": + data = read_benchmark(args.unit, args.benchmark_id) + elif args.benchmark_command == "bind": + from .benchmarks import bind_benchmark_candidate + data = bind_benchmark_candidate(args.unit, args.benchmark_id, args.candidate_id) + elif args.benchmark_command == "compare": + data = compare_benchmark(args.unit, args.benchmark_id, candidate_id=args.candidate_id) + else: + row = read_json_object(args.result) + required = {"observed_conditions", "arm", "task_id", "trial", "run_id", "score", "artifact_ids"} + allowed = required | {"status", "reason", "metrics"} + if set(row) - allowed or required - set(row): + raise HarneloopError("Benchmark result has unknown fields or is missing required fields") + data = record_benchmark_result(args.unit, args.benchmark_id, **row) + print(json.dumps(data, indent=2)) + return 0 + + if args.command == "refine": + if args.refine_command == "compose": + candidate = compose_candidates(args.unit, args.candidate_id, args.summary) + data = read_candidate(args.unit, candidate.name) + elif args.refine_command == "learn": + data = add_lesson(args.unit, args.cue, args.lesson, args.run_id, args.artifact_id) + else: + data = list_lessons(args.unit, args.query) + print(json.dumps(data, indent=2)) + return 0 + if args.command == "onboard": if args.format == "json": print(json.dumps(render_onboarding_json(), indent=2)) @@ -589,7 +675,11 @@ def main(argv: list[str] | None = None) -> int: if args.command == "run": if args.run_command == "start": - path = start_run(args.unit, args.task, args.candidate_id, args.attempt_id) + benchmark_fields = {} + if any(value is not None for value in [args.benchmark_id, args.benchmark_task_id, args.benchmark_trial, args.benchmark_arm]): + benchmark_fields = {"benchmark_id": args.benchmark_id, "benchmark_task_id": args.benchmark_task_id, + "benchmark_trial": args.benchmark_trial, "benchmark_arm": args.benchmark_arm} + path = start_run(args.unit, args.task, args.candidate_id, args.attempt_id, **benchmark_fields) print(f"Started run: {path.name}") return 0 if args.run_command == "finish": diff --git a/src/harneloop/evidence.py b/src/harneloop/evidence.py index b0f7adc..e4e919e 100644 --- a/src/harneloop/evidence.py +++ b/src/harneloop/evidence.py @@ -3,12 +3,12 @@ from pathlib import Path from typing import Any -from .candidate import VALIDATION_TIER_RANK, read_candidate, validate_tier +from .candidate import VALIDATION_TIER_RANK, candidate_digest, read_candidate, validate_tier from .errors import HarneloopError from .locking import file_lock, harness_lock_path from .runs import read_run from .state import now_iso -from .versioning import candidate_root, ensure_unit +from .versioning import candidate_root, ensure_unit, hash_file from .yamlio import read_yaml, write_yaml @@ -39,6 +39,12 @@ def validate_evidence_references(unit_root: Path, record: dict[str, Any]) -> Non run_id = record.get("run_id") artifact_id = record.get("artifact_id") evidence_path = record.get("path") + candidate_id = record.get("candidate_id") + # New evidence records bind all observations to the overlay being assessed. + # Legacy records without a fingerprint remain readable. + if candidate_id and record.get("candidate_sha256"): + if candidate_digest(unit_root, str(candidate_id)) != record["candidate_sha256"]: + raise HarneloopError("Candidate overlay changed after evidence was attached; collect fresh evidence") if artifact_id and not run_id: raise HarneloopError(f"Evidence artifact `{artifact_id}` requires a run_id") @@ -49,6 +55,9 @@ def validate_evidence_references(unit_root: Path, record: dict[str, Any]) -> Non if not run_file.is_file(): raise HarneloopError(f"Run does not exist: {run_id}") run_record = read_run(unit_root, str(run_id)) + if candidate_id and run_record.get("candidate_id") == candidate_id and run_record.get("candidate_sha256"): + if candidate_digest(unit_root, str(candidate_id)) != run_record["candidate_sha256"]: + raise HarneloopError("Candidate overlay changed since the referenced run; rerun validation") if artifact_id: artifacts = (run_record.get("artifacts") or []) if run_record else [] @@ -59,6 +68,10 @@ def validate_evidence_references(unit_root: Path, record: dict[str, Any]) -> Non stored_file = (unit_root / str(stored_path)).resolve() if stored_path else None if stored_file is None or not stored_file.is_file(): raise HarneloopError(f"Artifact stored file does not exist for `{artifact_id}` in run `{run_id}`") + if not stored_file.is_relative_to(unit_root.resolve()): + raise HarneloopError("Stored artifact escapes its harness unit") + if artifact.get("sha256") and hash_file(stored_file) != artifact["sha256"]: + raise HarneloopError("Stored artifact changed after capture; collect fresh evidence") if evidence_path: referenced_file = Path(str(evidence_path)) @@ -99,6 +112,7 @@ def add_evidence( "outcome": outcome, "validation_tier": selected_tier, "candidate_base_version": candidate.get("base_version"), + "candidate_sha256": candidate_digest(unit_root, candidate_id), "run_id": run_id, "artifact_id": artifact_id, "path": str(resolved_path) if resolved_path else None, diff --git a/src/harneloop/refinement.py b/src/harneloop/refinement.py new file mode 100644 index 0000000..3351e27 --- /dev/null +++ b/src/harneloop/refinement.py @@ -0,0 +1,159 @@ +"""Compose experimental changes and retrieve evidence-linked observations. + +Composition never inherits source validation. Lessons are operator-authored +observations, not universal facts or authenticated external grader verdicts. +""" +from __future__ import annotations + +import hashlib +import json +import stat +from pathlib import Path +from typing import Any + +from .candidate import create_candidate, read_candidate, validate_tier, OPEN_CANDIDATE_STATUSES, VALIDATION_TIER_RANK +from .errors import HarneloopError +from .locking import file_lock, harness_lock_path +from .paths import is_protected_candidate_path, is_secret_path +from .runs import read_run +from .versioning import ensure_unit, hash_file +from .yamlio import read_yaml, write_yaml + + +def _identifier(value: str, prefix: str) -> str: + if not isinstance(value, str) or not value.startswith(prefix) or not value[len(prefix):].isdigit(): + raise HarneloopError(f"Invalid {prefix} identifier") + return value + + +def _contained(root: Path, path: Path) -> Path: + if not path.resolve().is_relative_to(root.resolve()): + raise HarneloopError("Evidence or overlay path escapes its harness unit") + for part in [path, *path.parents]: + if part == root: + break + if part.is_symlink(): + raise HarneloopError("Symlinked evidence and overlays are not supported") + return path + + +def compose_candidates(unit_root: Path, candidate_ids: list[str], summary: str) -> Path: + """Union compatible overlays on one base. Conflicts require an agent's edit. + + Sources remain open, the new candidate receives no inherited evidence, + and normal promotion still requires fresh validation of the composition. + """ + unit_root = unit_root.resolve() + ensure_unit(unit_root) + if (not isinstance(summary, str) or not summary.strip() or not isinstance(candidate_ids, list) + or any(not isinstance(item, str) for item in candidate_ids) + or len(set(candidate_ids)) != len(candidate_ids) or len(candidate_ids) < 2): + raise HarneloopError("Composition requires a summary and at least two distinct candidates") + with file_lock(harness_lock_path(unit_root, "lifecycle")): + current = read_yaml(unit_root / "unit.yaml").get("current_version") + records, files, sources = [], {}, [] + for candidate_id in candidate_ids: + _identifier(candidate_id, "cand-") + record = read_candidate(unit_root, candidate_id) + if record.get("base_version") != current or record.get("status") not in OPEN_CANDIDATE_STATUSES - {"needs_rebase"}: + raise HarneloopError("Composition requires open candidates on the current base") + overlay = _contained(unit_root, unit_root / "candidates" / candidate_id / "changes") + inventory = [] + for path in sorted(overlay.rglob("*")): + _contained(unit_root, path) + if not path.is_file(): + continue + name = path.relative_to(overlay).as_posix() + if is_protected_candidate_path(name) or is_secret_path(path): + raise HarneloopError(f"Composition source contains a protected or secret path: {name}") + content = path.read_bytes() + mode = stat.S_IMODE(path.stat().st_mode) + if name in files and files[name] != (content, mode): + raise HarneloopError(f"Composition conflict at {name}; resolve it explicitly in a new candidate") + files[name] = (content, mode) + inventory.append({"path": name, "sha256": hashlib.sha256(content).hexdigest(), "mode": mode}) + records.append(record) + sources.append({"candidate_id": candidate_id, "files": inventory}) + planes = {r.get("plane", "mixed") for r in records} + if len(planes) != 1: + raise HarneloopError("Keep target-harness, evaluation and infrastructure composition separate") + if not files: + raise HarneloopError("Composition has no file changes") + tier = max((validate_tier(str(r.get("validation_tier", "targeted"))) for r in records), key=VALIDATION_TIER_RANK.__getitem__) + candidate = create_candidate(unit_root, summary, plane=planes.pop(), validation_tier=tier) + for name, (content, mode) in files.items(): + target = candidate / "changes" / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(content) + target.chmod(mode) + metadata = read_candidate(unit_root, candidate.name) + metadata["composition"] = {"sources": sources, "requires_fresh_validation": True} + write_yaml(candidate / "candidate.yaml", metadata) + return candidate + + +def _lesson_evidence(unit_root: Path, run_id: str, artifact_id: str) -> dict: + _identifier(run_id, "run-") + _identifier(artifact_id, "artifact-") + _contained(unit_root, unit_root / "runtime/runs" / run_id / "run.yaml") + run = read_run(unit_root, run_id) + if run.get("status") not in {"succeeded", "failed", "stopped"}: + raise HarneloopError("A lesson requires a finished run") + artifact = next((a for a in run.get("artifacts", []) if a.get("id") == artifact_id), None) + if not artifact or not isinstance(artifact.get("stored_path"), str): + raise HarneloopError("Lesson artifact is not present in the run") + path = _contained(unit_root, unit_root / artifact["stored_path"]) + if not path.resolve().is_relative_to((unit_root / "runtime/artifacts" / run_id).resolve()): + raise HarneloopError("Lesson artifact is not in the recorded run's storage") + if not path.is_file() or hash_file(path) != artifact.get("sha256"): + raise HarneloopError("Lesson evidence is missing or changed") + return {"run_id": run_id, "artifact_id": artifact_id, "sha256": artifact["sha256"]} + + +def _digest(data: dict) -> str: + return hashlib.sha256(json.dumps(data, sort_keys=True, ensure_ascii=False).encode()).hexdigest() + + +def add_lesson(unit_root: Path, cue: str, lesson: str, run_id: str, artifact_id: str) -> dict[str, Any]: + unit_root = unit_root.resolve() + ensure_unit(unit_root) + if not isinstance(cue, str) or not isinstance(lesson, str) or not cue.strip() or not lesson.strip(): + raise HarneloopError("Lesson cue and observation cannot be empty") + with file_lock(harness_lock_path(unit_root, "lessons")): + evidence = _lesson_evidence(unit_root, run_id, artifact_id) + root = _contained(unit_root, unit_root / ".evolve/lessons") + root.mkdir(parents=True, exist_ok=True) + numbers = [int(p.stem[7:]) for p in root.glob("lesson-*.yaml") if p.stem[7:].isdigit()] + lesson_id = f"lesson-{max(numbers, default=0)+1:04d}" + row = {"id": lesson_id, "cue": cue.strip(), "lesson": lesson.strip(), "evidence": evidence, + "scope": "operator-authored observation supported by the referenced artifact"} + row["sha256"] = _digest(row) + write_yaml(root / f"{lesson_id}.yaml", row) + return row + + +def list_lessons(unit_root: Path, query: str = "") -> list[dict[str, Any]]: + unit_root = unit_root.resolve() + ensure_unit(unit_root) + if not isinstance(query, str): + raise HarneloopError("Lesson query must be text") + results = [] + with file_lock(harness_lock_path(unit_root, "lessons")): + root = _contained(unit_root, unit_root / ".evolve/lessons") + for path in sorted(root.glob("lesson-*.yaml")): + _contained(unit_root, path) + row = read_yaml(path) + if (not isinstance(row, dict) or not isinstance(row.get("cue"), str) + or not isinstance(row.get("lesson"), str) or not isinstance(row.get("evidence"), dict) + or not isinstance(row["evidence"].get("run_id"), str) + or not isinstance(row["evidence"].get("artifact_id"), str)): + raise HarneloopError("Lesson record is malformed; inspect its cue, lesson and evidence references") + original = {k: v for k, v in row.items() if k != "sha256"} + if row.get("sha256") != _digest(original): + raise HarneloopError("Lesson record has changed") + evidence = _lesson_evidence(unit_root, row["evidence"]["run_id"], row["evidence"]["artifact_id"]) + if evidence != row["evidence"]: + raise HarneloopError("Lesson evidence changed") + if not query or query.casefold() in (row["cue"]+" "+row["lesson"]).casefold(): + results.append(row) + return results diff --git a/src/harneloop/runs.py b/src/harneloop/runs.py index c4b1510..f974d9c 100644 --- a/src/harneloop/runs.py +++ b/src/harneloop/runs.py @@ -4,7 +4,7 @@ from pathlib import Path from typing import Any -from .candidate import OPEN_CANDIDATE_STATUSES, read_candidate +from .candidate import OPEN_CANDIDATE_STATUSES, candidate_digest, read_candidate from .errors import HarneloopError from .intake import ensure_intake_ready from .locking import file_lock, harness_lock_path @@ -62,16 +62,61 @@ def start_run( task: str, candidate_id: str | None = None, attempt_id: str | None = None, + *, + benchmark_id: str | None = None, + benchmark_task_id: str | None = None, + benchmark_trial: int | None = None, + benchmark_arm: str | None = None, ) -> Path: unit_root = unit_root.resolve() ensure_unit(unit_root) ensure_intake_ready(unit_root) candidate: dict[str, Any] | None = None + candidate_sha256: str | None = None if candidate_id: + candidate_sha256 = candidate_digest(unit_root, candidate_id) candidate = read_candidate(unit_root, candidate_id) status = str(candidate.get("status", "draft")) if status not in OPEN_CANDIDATE_STATUSES or status == "needs_rebase": raise HarneloopError(f"Candidate `{candidate_id}` cannot be tested while its status is `{status}`") + benchmark_fields = (benchmark_id, benchmark_task_id, benchmark_trial, benchmark_arm) + if any(value is not None for value in benchmark_fields): + if ( + not all(value is not None for value in benchmark_fields) + or not isinstance(benchmark_id, str) + or not benchmark_id.strip() + or not isinstance(benchmark_task_id, str) + or not benchmark_task_id.strip() + or isinstance(benchmark_trial, bool) + or not isinstance(benchmark_trial, int) + or benchmark_trial < 0 + or benchmark_arm not in {"baseline", "candidate"} + or (benchmark_arm == "baseline" and candidate_id is not None) + or (benchmark_arm == "candidate" and candidate_id is None) + ): + raise HarneloopError("Benchmark run identity is incomplete or inconsistent") + from .benchmarks import read_benchmark + + benchmark = read_benchmark(unit_root, benchmark_id) + task_record = next( + (item for item in benchmark["tasks"] if item["task_id"] == benchmark_task_id), + None, + ) + binding = benchmark.get("candidate_binding") + if ( + task_record is None + or task_record.get("split") != "test" + or benchmark_trial >= benchmark["repeats"] + or not isinstance(binding, dict) + ): + raise HarneloopError("Benchmark run must match a bound heldout task and trial") + if candidate_digest(unit_root, binding["candidate_id"]) != binding["candidate_sha256"]: + raise HarneloopError("Candidate overlay changed after benchmark binding") + if benchmark_arm == "candidate": + if candidate_id != binding.get("candidate_id"): + raise HarneloopError("Candidate run does not match the preregistered benchmark candidate") + elif candidate_id is not None: + raise HarneloopError("Baseline benchmark runs cannot use a candidate overlay") with file_lock(harness_lock_path(unit_root, "runs")): unit_meta = read_yaml(unit_root / "unit.yaml") run_id = next_run_id(unit_root) @@ -91,7 +136,12 @@ def start_run( "harness_version": unit_meta.get("current_version"), "candidate_id": candidate_id, "candidate_base_version": candidate.get("base_version") if candidate else None, + "candidate_sha256": candidate_sha256, "candidate_validation_tier": candidate.get("validation_tier") if candidate else None, + "benchmark_id": benchmark_id, + "benchmark_task_id": benchmark_task_id, + "benchmark_trial": benchmark_trial, + "benchmark_arm": benchmark_arm, "attempt_id": attempt_id, "evaluation_status": "pending", "evaluation_outcome": None, diff --git a/tests/test_benchmark_cli.py b/tests/test_benchmark_cli.py new file mode 100644 index 0000000..69f80d7 --- /dev/null +++ b/tests/test_benchmark_cli.py @@ -0,0 +1,59 @@ +import contextlib +import io +import json +import tempfile +import unittest +from pathlib import Path + +from harneloop.cli import main +from harneloop.unit import init_unit + + +class BenchmarkCliTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.unit = init_unit(self.root / "unit", "cli-test", "CLI Test") + self.manifest = { + "source": {"name": "external-benchmark", "revision": "a" * 40}, + "tasks": [{"task_id": "dev-1", "split": "dev"}, {"task_id": "test-1", "split": "test"}], + "repeats": 1, "seed": 42, "executor": "external-agent", "model": "fixed-model", + "effort": "high", "settings": {}, "tools": {"terminal": True}, + "budget_policy": {"calls": 10}, "grader": {"name": "deterministic", "revision": "b" * 40}, + } + + def call(self, args): + output, errors = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + status = main(args) + return status, output.getvalue(), errors.getvalue() + + def test_create_and_show_frozen_manifest(self): + path = self.root / "manifest.json" + path.write_text(json.dumps(self.manifest), encoding="utf-8") + status, output, _ = self.call(["benchmark", "create", str(self.unit), "pilot", "--manifest", str(path)]) + self.assertEqual(status, 0) + digest = json.loads(output)["manifest_sha256"] + status, output, _ = self.call(["benchmark", "show", str(self.unit), "pilot"]) + self.assertEqual(status, 0) + self.assertEqual(json.loads(output)["manifest_sha256"], digest) + + def test_invalid_json_returns_domain_error_without_traceback(self): + path = self.root / "bad.json" + path.write_text("{bad", encoding="utf-8") + status, _, error = self.call(["benchmark", "create", str(self.unit), "pilot", "--manifest", str(path)]) + self.assertEqual(status, 2) + self.assertIn("error:", error) + self.assertNotIn("Traceback", error) + + def test_record_rejects_unexpected_fields(self): + path = self.root / "bad-row.json" + path.write_text('{"unit_root":"escape"}', encoding="utf-8") + status, _, error = self.call(["benchmark", "record", str(self.unit), "pilot", "--result", str(path)]) + self.assertEqual(status, 2) + self.assertIn("error:", error) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_benchmarks.py b/tests/test_benchmarks.py new file mode 100644 index 0000000..4bdafc8 --- /dev/null +++ b/tests/test_benchmarks.py @@ -0,0 +1,502 @@ +from __future__ import annotations + +import tempfile +import unittest +from unittest.mock import patch +from pathlib import Path + +from harneloop.benchmarks import ( + bind_benchmark_candidate, + compare_benchmark, + create_benchmark, + read_benchmark, + record_benchmark_result, +) +from harneloop.candidate import create_candidate, set_candidate_status +from harneloop.errors import HarneloopError +from harneloop.intake import acknowledge_intake +from harneloop.runs import add_artifact, finish_run, read_run, start_run +from harneloop.unit import init_unit + + +class BenchmarkEvidenceTests(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + root = Path(self.temp_dir.name) + self.unit = init_unit(root / "unit", "benchmark-unit", "Benchmark Unit") + acknowledge_intake(self.unit, basis="user_delegated", note="Test delegates setup context.") + self.candidate = create_candidate(self.unit, "Improve task solving", plane="target_harness") + set_candidate_status(self.unit, self.candidate.name, "ready") + self.manifest = { + "source": {"name": "sample-eval", "revision": "sha256:" + "a" * 64}, + "tasks": [ + {"task_id": "dev-1", "split": "dev"}, + {"task_id": "test-1", "split": "test"}, + {"task_id": "test-2", "split": "test"}, + ], + "repeats": 2, + "seed": 17, + "executor": "codex-subagent", + "model": "luna-6", + "effort": "medium", + "settings": {"temperature": 0}, + "tools": {"filesystem": True, "terminal": True}, + "budget_policy": {"turn_cap": 20, "wall_seconds": 900}, + "grader": {"name": "official-evaluator", "revision": "b" * 40}, + } + self.observed_conditions = { + key: self.manifest[key] + for key in ("executor", "model", "effort", "settings", "tools", "budget_policy") + } + create_benchmark(self.unit, "pilot", self.manifest) + self.benchmark = bind_benchmark_candidate(self.unit, "pilot", self.candidate.name) + + def _finished_run( + self, + arm: str, + label: str, + content: str, + *, + task_id: str = "test-1", + trial: int = 0, + run_status: str = "succeeded", + ) -> tuple[str, str]: + candidate_id = self.candidate.name if arm == "candidate" else None + run_root = start_run( + self.unit, + task=label, + candidate_id=candidate_id, + benchmark_id="pilot", + benchmark_task_id=task_id, + benchmark_trial=trial, + benchmark_arm=arm, + ) + run_id = run_root.name + artifact = self.unit.parent / f"{run_id}-{arm}.patch" + artifact.write_text(content, encoding="utf-8") + artifact_record = add_artifact(self.unit, run_id, artifact, kind="patch") + finish_run(self.unit, run_id, run_status, "Task attempt complete") + return run_id, artifact_record["id"] + + def _record( + self, + arm: str, + task_id: str, + trial: int, + score: int | float | None, + *, + status: str = "scored", + reason: str | None = None, + ) -> dict: + run_id, artifact_id = self._finished_run( + arm, + f"{arm}:{task_id}:{trial}", + f"{score}\n", + task_id=task_id, + trial=trial, + ) + return record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm=arm, + task_id=task_id, + trial=trial, + run_id=run_id, + score=score, + artifact_ids=[artifact_id], + status=status, + reason=reason, + ) + + def test_manifest_is_frozen_and_fingerprinted(self) -> None: + created = read_benchmark(self.unit, "pilot") + + self.assertEqual(created["source"], self.manifest["source"]) + self.assertEqual(len(created["manifest_sha256"]), 64) + self.assertEqual(len(created["conditions_sha256"]), 64) + with self.assertRaises(HarneloopError): + create_benchmark(self.unit, "pilot", self.manifest) + + def test_candidate_binding_is_once_only(self) -> None: + with self.assertRaisesRegex(HarneloopError, "already bound"): + bind_benchmark_candidate(self.unit, "pilot", self.candidate.name) + + def test_candidate_must_be_marked_ready_before_binding(self) -> None: + other = create_candidate(self.unit, "Still refining", plane="target_harness") + create_benchmark(self.unit, "not-ready", self.manifest) + with self.assertRaisesRegex(HarneloopError, "Mark the dev-refined candidate ready"): + bind_benchmark_candidate(self.unit, "not-ready", other.name) + + def test_manifest_cannot_self_attest_candidate_binding(self) -> None: + manifest = { + **self.manifest, + "candidate_binding": { + "candidate_id": self.candidate.name, + "candidate_sha256": "0" * 64, + }, + } + with self.assertRaisesRegex(HarneloopError, "benchmark bind"): + create_benchmark(self.unit, "self-bound", manifest) + + def test_heldout_run_cannot_start_before_candidate_binding(self) -> None: + create_benchmark(self.unit, "unbound", self.manifest) + with self.assertRaisesRegex(HarneloopError, "bound heldout task"): + start_run( + self.unit, + task="unbound heldout", + benchmark_id="unbound", + benchmark_task_id="test-1", + benchmark_trial=0, + benchmark_arm="baseline", + ) + + def test_manifest_rejects_mutable_source_or_grader_revision(self) -> None: + manifest = {**self.manifest, "source": {"name": "sample-eval", "revision": "main"}} + with self.assertRaisesRegex(HarneloopError, "commit SHA"): + create_benchmark(self.unit, "mutable-source", manifest) + manifest = {**self.manifest, "grader": {"name": "grader", "revision": "latest"}} + with self.assertRaisesRegex(HarneloopError, "commit SHA"): + create_benchmark(self.unit, "mutable-grader", manifest) + + def test_candidate_overlay_change_after_binding_is_rejected(self) -> None: + changes = self.unit / "candidates" / self.candidate.name / "changes" + (changes / "solver.py").write_text("def solve(): return 1\n", encoding="utf-8") + with self.assertRaisesRegex(HarneloopError, "Candidate overlay changed"): + self._record("candidate", "test-1", 0, 1.0) + + def test_run_task_identity_must_match_recorded_cell(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "wrong task", "patch") + with self.assertRaisesRegex(HarneloopError, "frozen task/trial identity"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-2", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + + def test_failed_run_cannot_be_claimed_as_scored(self) -> None: + run_id, artifact_id = self._finished_run( + "baseline", "failed task", "partial", run_status="failed" + ) + with self.assertRaisesRegex(HarneloopError, "cannot be reported as scored"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + + def test_one_run_cannot_be_reused_for_another_result(self) -> None: + row = self._record("baseline", "test-1", 0, 1.0) + with self.assertRaisesRegex(HarneloopError, "only one benchmark task/trial row"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=row["run_id"], + score=1.0, + artifact_ids=[artifact["id"] for artifact in row["artifacts"]], + ) + + def test_symlinked_artifact_run_root_is_rejected(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "symlink test", "patch") + run_artifacts = self.unit / "runtime" / "artifacts" / run_id + original_is_symlink = Path.is_symlink + + def report_symlink(path: Path) -> bool: + return path == run_artifacts or original_is_symlink(path) + + with patch.object(Path, "is_symlink", report_symlink): + with self.assertRaisesRegex(HarneloopError, "symlinks"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + + def test_symlinked_run_record_root_is_rejected(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "run path symlink", "patch") + run_root = self.unit / "runtime" / "runs" / run_id + original_is_symlink = Path.is_symlink + + def report_symlink(path: Path) -> bool: + return path == run_root or original_is_symlink(path) + + with patch.object(Path, "is_symlink", report_symlink): + with self.assertRaisesRegex(HarneloopError, "symlinks"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + + def test_symlinked_benchmark_storage_root_is_rejected(self) -> None: + benchmark_root = self.unit / ".evolve" / "benchmarks" / "pilot" + original_is_symlink = Path.is_symlink + + def report_symlink(path: Path) -> bool: + return path == benchmark_root or original_is_symlink(path) + + with patch.object(Path, "is_symlink", report_symlink): + with self.assertRaisesRegex(HarneloopError, "symlinks"): + read_benchmark(self.unit, "pilot") + + def test_comparison_pairs_exact_heldout_trials_and_counts_unavailable(self) -> None: + for task_id in ("test-1", "test-2"): + for trial in range(2): + self._record("baseline", task_id, trial, 0.0 if trial == 0 else 1.0) + if task_id == "test-2" and trial == 1: + self._record( + "candidate", + task_id, + trial, + None, + status="unavailable", + reason="executor timed out", + ) + else: + self._record("candidate", task_id, trial, 1.0) + + report = compare_benchmark(self.unit, "pilot", candidate_id=self.candidate.name) + + self.assertEqual(report["expected_pairs"], 4) + self.assertEqual(report["arms"]["baseline"]["scored"], 4) + self.assertEqual(report["arms"]["baseline"]["failed"], 2) + self.assertEqual(report["arms"]["candidate"]["unavailable"], 1) + self.assertEqual(report["paired"]["wins"], 2) + self.assertEqual(report["paired"]["losses"], 0) + self.assertEqual(report["paired"]["ties"], 1) + self.assertEqual(report["paired"]["unavailable"], 1) + self.assertIn("test-1", report["per_task"]) + + def test_comparison_rejects_missing_task_trial_rows(self) -> None: + self._record("baseline", "test-1", 0, 1.0) + self._record("candidate", "test-1", 0, 1.0) + + with self.assertRaisesRegex(HarneloopError, "incomplete"): + compare_benchmark(self.unit, "pilot", candidate_id=self.candidate.name) + + def test_result_rejects_nonfinite_score(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "invalid score", "patch") + + with self.assertRaisesRegex(HarneloopError, "finite numeric"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=float("nan"), + artifact_ids=[artifact_id], + ) + + def test_result_rejects_out_of_range_score_and_condition_mismatch(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "invalid result", "patch") + with self.assertRaisesRegex(HarneloopError, "between 0 and 1"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.1, + artifact_ids=[artifact_id], + ) + mismatched = {**self.observed_conditions, "model": "different-model"} + with self.assertRaisesRegex(HarneloopError, "do not match"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=mismatched, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + + def test_manifest_tampering_is_rejected(self) -> None: + from harneloop.yamlio import read_yaml, write_yaml + + path = self.unit / ".evolve" / "benchmarks" / "pilot" / "manifest.yaml" + manifest = read_yaml(path) + manifest["model"] = "tampered" + write_yaml(path, manifest) + + with self.assertRaisesRegex(HarneloopError, "manifest hash mismatch"): + read_benchmark(self.unit, "pilot") + + def test_result_record_tampering_is_rejected(self) -> None: + from harneloop.yamlio import read_yaml, write_yaml + + self._record("baseline", "test-1", 0, 1.0) + result_root = self.unit / ".evolve" / "benchmarks" / "pilot" / "results" + result_path = next(result_root.glob("*.yaml")) + row = read_yaml(result_path) + row["score"] = 0.0 + write_yaml(result_path, row) + + with self.assertRaisesRegex(HarneloopError, "record hash mismatch"): + compare_benchmark(self.unit, "pilot", candidate_id=self.candidate.name) + + def test_result_rejects_invalid_optional_metrics(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "invalid metric", "patch") + + with self.assertRaisesRegex(HarneloopError, "integer token count"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + metrics={"input_tokens": 2.5}, + ) + + def test_result_rejects_running_run_and_wrong_split(self) -> None: + run_root = start_run( + self.unit, + task="still running", + benchmark_id="pilot", + benchmark_task_id="test-1", + benchmark_trial=0, + benchmark_arm="baseline", + ) + artifact = self.unit.parent / "running.patch" + artifact.write_text("patch\n", encoding="utf-8") + artifact_record = add_artifact(self.unit, run_root.name, artifact, kind="patch") + + with self.assertRaisesRegex(HarneloopError, "finished"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_root.name, + score=1.0, + artifact_ids=[artifact_record["id"]], + ) + + finish_run(self.unit, run_root.name, "succeeded") + with self.assertRaisesRegex(HarneloopError, "test split"): + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="dev-1", + trial=0, + run_id=run_root.name, + score=1.0, + artifact_ids=[artifact_record["id"]], + ) + + def test_comparison_detects_mutated_artifact(self) -> None: + baseline_run = None + for task_id in ("test-1", "test-2"): + for trial in range(2): + for arm in ("baseline", "candidate"): + run_id, artifact_id = self._finished_run( + arm, f"{arm}:{task_id}:{trial}", "patch", task_id=task_id, trial=trial + ) + record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm=arm, + task_id=task_id, + trial=trial, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + ) + if arm == "baseline" and task_id == "test-1" and trial == 0: + baseline_run = run_id + baseline_record = read_run(self.unit, baseline_run) + artifact_path = self.unit / baseline_record["artifacts"][0]["stored_path"] + artifact_path.write_text("tampered", encoding="utf-8") + + with self.assertRaisesRegex(HarneloopError, "hash mismatch"): + compare_benchmark(self.unit, "pilot", candidate_id=self.candidate.name) + + def test_duplicate_result_key_is_rejected(self) -> None: + self._record("baseline", "test-1", 0, 1.0) + with self.assertRaisesRegex(HarneloopError, "already exists"): + self._record("baseline", "test-1", 0, 1.0) + + def test_metrics_are_only_reported_when_available(self) -> None: + run_id, artifact_id = self._finished_run("baseline", "metrics", "patch") + row = record_benchmark_result( + self.unit, + "pilot", + observed_conditions=self.observed_conditions, + arm="baseline", + task_id="test-1", + trial=0, + run_id=run_id, + score=1.0, + artifact_ids=[artifact_id], + metrics={ + "duration_seconds": 12.5, + "input_tokens": 800, + "output_tokens": 400, + "cost_usd": 0.0, + }, + ) + for task_id in ("test-1", "test-2"): + for trial in range(2): + if task_id == "test-1" and trial == 0: + self._record("candidate", task_id, trial, 1.0) + continue + self._record("baseline", task_id, trial, 1.0) + self._record("candidate", task_id, trial, 1.0) + + report = compare_benchmark(self.unit, "pilot", candidate_id=self.candidate.name) + + self.assertEqual(row["metrics"]["duration_seconds"], 12.5) + self.assertEqual(row["metrics"]["input_tokens"], 800) + self.assertEqual(row["metrics"]["output_tokens"], 400) + self.assertEqual(row["metrics"]["cost_usd"], 0.0) + self.assertEqual(len(report["metric_observations"]["duration_seconds"]), 1) + self.assertEqual(report["metric_observations"]["duration_seconds"][0]["value"], 12.5) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_evidence_integrity.py b/tests/test_evidence_integrity.py new file mode 100644 index 0000000..6be2334 --- /dev/null +++ b/tests/test_evidence_integrity.py @@ -0,0 +1,58 @@ +import tempfile +import unittest +from pathlib import Path + +from harneloop.candidate import create_candidate, set_candidate_status +from harneloop.errors import HarneloopError +from harneloop.evidence import add_evidence +from harneloop.intake import acknowledge_intake +from harneloop.runs import add_artifact, finish_run, start_run +from harneloop.unit import init_unit +from harneloop.versioning import promote_candidate + + +class EvidenceIntegrityTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.unit = init_unit(Path(self.temp.name) / 'unit', 'integrity', 'Integrity') + acknowledge_intake(self.unit, basis='user_delegated', note='Test setup') + self.candidate = create_candidate(self.unit, 'Improve policy') + self.policy = self.candidate / 'changes/agent-facing/policy.md' + self.policy.parent.mkdir(parents=True, exist_ok=True) + self.policy.write_text('Tested policy') + self.run = start_run(self.unit, 'Check policy', candidate_id=self.candidate.name) + result = Path(self.temp.name) / 'result.txt' + result.write_text('Observed result') + self.artifact = add_artifact(self.unit, self.run.name, result, 'test-result') + finish_run(self.unit, self.run.name, 'succeeded', 'Observed result') + + def attach(self): + return add_evidence(self.unit, self.candidate.name, 'test', 'Observed result', + run_id=self.run.name, artifact_id=self.artifact['id']) + + def test_edited_overlay_cannot_attach_old_candidate_run(self): + self.policy.write_text('Untested policy') + with self.assertRaisesRegex(HarneloopError, 'changed'): + self.attach() + + def test_promotion_rechecks_overlay_after_evidence_attachment(self): + self.attach() + set_candidate_status(self.unit, self.candidate.name, 'ready') + self.policy.write_text('Untested policy') + with self.assertRaisesRegex(HarneloopError, 'changed'): + promote_candidate(self.unit, self.candidate.name, 'v1') + self.assertFalse((self.unit / 'versions/v1').exists()) + + def test_promotion_rechecks_stored_artifact_hash(self): + self.attach() + set_candidate_status(self.unit, self.candidate.name, 'ready') + (self.unit / self.artifact['stored_path']).write_text('Changed result') + with self.assertRaisesRegex(HarneloopError, 'changed'): + promote_candidate(self.unit, self.candidate.name, 'v1') + + def test_unchanged_run_evidence_allows_normal_promotion(self): + self.attach() + set_candidate_status(self.unit, self.candidate.name, 'ready') + promote_candidate(self.unit, self.candidate.name, 'v1') + self.assertEqual((self.unit / 'agent-facing/policy.md').read_text(), 'Tested policy') diff --git a/tests/test_refinement.py b/tests/test_refinement.py new file mode 100644 index 0000000..b436bb4 --- /dev/null +++ b/tests/test_refinement.py @@ -0,0 +1,90 @@ +import tempfile +import unittest +import os +import stat +from pathlib import Path + +from harneloop.candidate import create_candidate, read_candidate +from harneloop.errors import HarneloopError +from harneloop.evidence import list_evidence +from harneloop.intake import acknowledge_intake +from harneloop.runs import start_run, add_artifact, finish_run +from harneloop.unit import init_unit +from harneloop.refinement import compose_candidates, add_lesson, list_lessons +from harneloop.yamlio import write_yaml + + +class RefinementTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.unit = init_unit(Path(self.temp.name) / "unit", "refine", "Refinement") + acknowledge_intake(self.unit, basis="user_delegated", note="Test setup delegated") + + def candidate(self, name, text): + c = create_candidate(self.unit, name, plane="target_harness", validation_tier="representative") + p = c / "changes/agent-facing" / name + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text) + return c + + def test_composition_preserves_sources_and_requires_fresh_evidence(self): + a, b = self.candidate("a.md", "A"), self.candidate("b.md", "B") + result = compose_candidates(self.unit, [a.name, b.name], "Combine orthogonal changes") + self.assertEqual((result / "changes/agent-facing/a.md").read_text(), "A") + self.assertEqual((result / "changes/agent-facing/b.md").read_text(), "B") + self.assertEqual(read_candidate(self.unit, result.name)["validation_tier"], "representative") + self.assertEqual(list_evidence(self.unit, result.name), []) + self.assertEqual(read_candidate(self.unit, a.name)["status"], "accumulating") + + def test_conflicting_composition_is_rejected_without_creating_candidate(self): + a, b = self.candidate("same.md", "A"), self.candidate("same.md", "B") + before = sorted((self.unit / "candidates").iterdir()) + with self.assertRaises(HarneloopError): + compose_candidates(self.unit, [a.name, b.name], "Conflict") + self.assertEqual(before, sorted((self.unit / "candidates").iterdir())) + + def test_protected_source_files_are_rejected(self): + a, b = self.candidate("a.md", "A"), self.candidate("b.md", "B") + (a / "changes/unit.yaml").write_text("bad: true") + with self.assertRaises(HarneloopError): + compose_candidates(self.unit, [a.name, b.name], "Invalid") + + def test_cross_plane_composition_is_rejected(self): + a = self.candidate("a.md", "A") + b = create_candidate(self.unit, "Grader change", plane="evaluation") + with self.assertRaises(HarneloopError): + compose_candidates(self.unit, [a.name, b.name], "Mixed evidence") + + @unittest.skipIf(os.name == 'nt', 'Windows does not expose POSIX executable mode') + def test_composition_preserves_executable_mode(self): + a, b = self.candidate('script.sh', '#!/bin/sh\nexit 0\n'), self.candidate('b.md', 'B') + (a / 'changes/agent-facing/script.sh').chmod(0o755) + result = compose_candidates(self.unit, [a.name, b.name], 'Keep runnable script') + self.assertEqual(stat.S_IMODE((result / 'changes/agent-facing/script.sh').stat().st_mode), 0o755) + + def test_malformed_lesson_produces_actionable_error(self): + root = self.unit / '.evolve/lessons' + root.mkdir(parents=True, exist_ok=True) + write_yaml(root / 'lesson-0001.yaml', {'evidence': None}) + with self.assertRaises(HarneloopError): + list_lessons(self.unit) + + def test_lesson_requires_finished_run_and_rechecks_evidence(self): + run = start_run(self.unit, "Observed edge case") + p = Path(self.temp.name) / "result.txt" + p.write_text("Concrete observation") + artifact = add_artifact(self.unit, run.name, p, "test-result") + with self.assertRaises(HarneloopError): + add_lesson(self.unit, "boundary inputs", "Test both sides of each boundary", run.name, artifact["id"]) + finish_run(self.unit, run.name, "succeeded", "Observed result") + lesson = add_lesson(self.unit, "boundary inputs", "Test both sides of each boundary", run.name, artifact["id"]) + self.assertEqual(list_lessons(self.unit, "boundary")[0]["id"], lesson["id"]) + self.assertEqual(list_lessons(self.unit, "unrelated"), []) + (self.unit / artifact["stored_path"]).write_text("Changed after observation") + with self.assertRaises(HarneloopError): + list_lessons(self.unit) + + +if __name__ == "__main__": + unittest.main() From 8d5b2366975be3fc2b57f38a7ea6595072b68ad4 Mon Sep 17 00:00:00 2001 From: Kristofer Jussmann Date: Tue, 20 Oct 2026 20:39:15 +0300 Subject: [PATCH 2/2] Reject evidence linked to a different candidate --- src/harneloop/evidence.py | 2 ++ tests/test_evidence_integrity.py | 15 +++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/src/harneloop/evidence.py b/src/harneloop/evidence.py index e4e919e..eed610a 100644 --- a/src/harneloop/evidence.py +++ b/src/harneloop/evidence.py @@ -55,6 +55,8 @@ def validate_evidence_references(unit_root: Path, record: dict[str, Any]) -> Non if not run_file.is_file(): raise HarneloopError(f"Run does not exist: {run_id}") run_record = read_run(unit_root, str(run_id)) + if candidate_id and run_record.get("candidate_id") not in {None, candidate_id}: + raise HarneloopError("Evidence run belongs to a different candidate") if candidate_id and run_record.get("candidate_id") == candidate_id and run_record.get("candidate_sha256"): if candidate_digest(unit_root, str(candidate_id)) != run_record["candidate_sha256"]: raise HarneloopError("Candidate overlay changed since the referenced run; rerun validation") diff --git a/tests/test_evidence_integrity.py b/tests/test_evidence_integrity.py index 6be2334..ce5b6d3 100644 --- a/tests/test_evidence_integrity.py +++ b/tests/test_evidence_integrity.py @@ -56,3 +56,18 @@ def test_unchanged_run_evidence_allows_normal_promotion(self): set_candidate_status(self.unit, self.candidate.name, 'ready') promote_candidate(self.unit, self.candidate.name, 'v1') self.assertEqual((self.unit / 'agent-facing/policy.md').read_text(), 'Tested policy') + + def test_other_candidates_run_cannot_validate_this_candidate(self): + other = create_candidate(self.unit, 'Different policy') + with self.assertRaisesRegex(HarneloopError, 'different candidate'): + add_evidence(self.unit, other.name, 'test', 'Wrong candidate result', + run_id=self.run.name, artifact_id=self.artifact['id']) + + def test_baseline_run_can_supply_comparison_context(self): + baseline = start_run(self.unit, 'Baseline observation') + result = Path(self.temp.name) / 'baseline.txt' + result.write_text('Baseline context') + artifact = add_artifact(self.unit, baseline.name, result, 'test-result') + finish_run(self.unit, baseline.name, 'succeeded', 'Baseline observation') + add_evidence(self.unit, self.candidate.name, 'comparison', 'Baseline context', + run_id=baseline.name, artifact_id=artifact['id'])