diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f03b490..eb39f2cd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,7 +4,8 @@ # python-syntax — byte-compiles every Python tree (catches syntax/import- # time breakage cheaply, no third-party deps needed). # root-pytest — the suite-level tests/ directory (registry, scoring, -# normalization, acceptance gate). Needs only pytest; +# normalization, acceptance gate) plus source-bound +# orchestrator resume/cohort contracts. Needs only pytest; # dataset-dependent cases skip gracefully when their # artifacts are absent. # bun-workspace — root bun install resolving @elizaos/* from the eliza/ @@ -71,7 +72,11 @@ jobs: - name: Run suite-level tests env: PYTHONPATH: ${{ github.workspace }} - run: python -m pytest benchmarks/tests -q + run: >- + python -m pytest benchmarks/tests + benchmarks/suites/orchestrator/tests/test_execution_identity.py + benchmarks/suites/orchestrator/tests/test_cohort.py + benchmarks/suites/orchestrator/tests/test_cohort_provider_forwarder.py -q bun-workspace: name: bun workspace (plugin + TS suites) @@ -85,6 +90,8 @@ jobs: bun run --cwd plugin-benchmarks test bun run --cwd plugin-benchmarks typecheck bun run --cwd plugin-benchmarks build + - name: Manifest exporter CLI smoke + run: bun run lifeops-bench:manifest --help - name: framework typecheck run: bun run --cwd framework/typescript typecheck - name: suite tests (keyless vitest/bun-test lanes) diff --git a/.github/workflows/lifeops-bench-python.yml b/.github/workflows/lifeops-bench-python.yml index cc1681fd..440376b1 100644 --- a/.github/workflows/lifeops-bench-python.yml +++ b/.github/workflows/lifeops-bench-python.yml @@ -197,14 +197,12 @@ jobs: run: | set -e mkdir -p lifeops_bench_results - # The `wrong` reference oracle runs entirely in-process and - # exercises the runner / scorer / executor pipeline without any - # LLM. PerfectAgent's CLI entry point currently requires a - # per-scenario constructor (see __main__._build_agent_fn) and is - # covered instead by the conformance test in unit-tests above. + # A deterministic negative control exercises five static domains. + # Its results are explicitly non-publishable conformance evidence. uv run python -m eliza_lifeops_bench \ --agent wrong \ - --mode static \ + --suite smoke \ + --offline-conformance \ --seeds 1 \ --concurrency 4 \ --max-cost-usd 0.01 \ diff --git a/suites/lifeops-bench/tests/test_manifest_export.py b/suites/lifeops-bench/tests/test_manifest_export.py index eabcf57e..31805152 100644 --- a/suites/lifeops-bench/tests/test_manifest_export.py +++ b/suites/lifeops-bench/tests/test_manifest_export.py @@ -8,25 +8,14 @@ from pathlib import Path PACKAGE_ROOT = Path(__file__).resolve().parents[1] -REPO_ROOT = PACKAGE_ROOT.parents[1] MANIFEST_PATH = PACKAGE_ROOT / "manifests" / "actions.manifest.json" SUMMARY_PATH = PACKAGE_ROOT / "manifests" / "actions.summary.md" -ROOT_PACKAGE_JSON = REPO_ROOT / "package.json" def _manifest() -> dict[str, object]: return json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) -def test_root_manifest_regeneration_script_is_registered() -> None: - root_package = json.loads(ROOT_PACKAGE_JSON.read_text(encoding="utf-8")) - command = root_package["scripts"]["lifeops-bench:manifest"] - assert "suites/lifeops-bench/scripts/export-action-manifest.ts" in command - assert "--conditions=eliza-source" in command - assert "--conditions=development" in command - assert "--import tsx" in command - - def test_manifest_has_in_tree_generator_metadata() -> None: manifest = _manifest() assert manifest["schemaVersion"] == 1 diff --git a/suites/orchestrator/README.md b/suites/orchestrator/README.md index fa7177b0..2bf4ea0a 100644 --- a/suites/orchestrator/README.md +++ b/suites/orchestrator/README.md @@ -66,7 +66,7 @@ python3 -m benchmarks.orchestrator run \ Idempotent behavior: -- Existing successful signatures are skipped automatically. +- Existing successful signatures are skipped only when the request, repository revisions, runtime and benchmark/harness source, corpus inputs, and root lockfiles match. Changed inputs require a new run. Subscription checkpoint namespaces use the same source-sensitive execution contract; old namespaces are not reused. Cross-harness comparison grouping remains separate. - `--rerun-failed` reruns only signatures whose latest run failed. - `--force` always creates a fresh run. diff --git a/suites/orchestrator/cohort.py b/suites/orchestrator/cohort.py index 7ad4732b..8b2b3f2e 100644 --- a/suites/orchestrator/cohort.py +++ b/suites/orchestrator/cohort.py @@ -9,7 +9,6 @@ from __future__ import annotations -import hashlib import json import shutil import threading @@ -41,6 +40,7 @@ repair_nonzero_returncode_statuses, resume_run_group, ) +from .execution_identity import PhaseExecutionIdentity, build_phase_execution_identity from .locking import campaign_execution_lock from .provider_forwarder import ( ProviderForwarderProcess, @@ -88,49 +88,6 @@ DEFAULT_STORAGE_MIN_FREE_BYTES = 8 * 1024**3 DEFAULT_STORAGE_EXPECTED_HEADROOM_BYTES = 4 * 1024**3 DEFAULT_STORAGE_CHECK_INTERVAL_SECONDS = 30.0 -_STORAGE_CONTROL_KEYS = frozenset( - { - "campaign_storage_min_free_bytes", - "campaign_storage_expected_headroom_bytes", - "campaign_storage_check_interval_s", - } -) -_NAMESPACE_RUNTIME_KEYS = _STORAGE_CONTROL_KEYS | frozenset( - { - "_replace_adapter_defaults", - "campaign_silent_timeout_s", - "claude_subscription_gateway_url", - "eliza_bench_http_timeout_s", - "hermes_timeout_s", - "hl_bench_command_timeout_s", - "openclaw_timeout_s", - "timeout_s", - } -) -_FINGERPRINT_EXCLUDED_PARTS = frozenset( - { - ".cache", - ".git", - ".mypy_cache", - ".pytest_cache", - ".ruff_cache", - ".turbo", - ".venv", - "__pycache__", - "artifacts", - "benchmark_results", - "build", - "coverage", - "dist", - "node_modules", - "outputs", - "results", - "runs", - "venv", - } -) - - class BenchmarkCohortStatus(StrEnum): SUCCEEDED = "succeeded" FAILED = "failed" @@ -139,13 +96,6 @@ class BenchmarkCohortStatus(StrEnum): UNSUPPORTED = "unsupported" -@dataclass(frozen=True) -class PhaseExecutionIdentity: - namespace: str - contract: Mapping[str, object] - checkpoint_relpath: str - - @dataclass(frozen=True) class StoragePreflight: checked_at: str @@ -213,124 +163,6 @@ def _canonical_json(value: object) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) -def build_phase_execution_identity( - *, - workspace_root: Path, - adapter: BenchmarkAdapter, - request: RunRequest, - harnesses: tuple[str, ...], - repo_meta: Mapping[str, str | None], -) -> PhaseExecutionIdentity: - """Build the PID/run-group-independent replay identity for one phase.""" - - extra = dict(request.extra_config) - dataset_config = { - key: value - for key, value in extra.items() - if key not in _NAMESPACE_RUNTIME_KEYS - and key - not in { - "campaign_corpus_sha256", - "campaign_phase", - "campaign_profile", - "reasoning_effort", - } - } - source_fingerprint = _relevant_source_fingerprint( - workspace_root=workspace_root, - adapter=adapter, - ) - explicit_corpus = extra.get("campaign_corpus_sha256") - if explicit_corpus is not None and ( - not isinstance(explicit_corpus, str) or not explicit_corpus.strip() - ): - raise ValueError("campaign_corpus_sha256 must be a non-empty string") - corpus_sha256 = ( - explicit_corpus.strip() - if isinstance(explicit_corpus, str) - else hashlib.sha256( - _canonical_json( - { - "adapter_directory": adapter.directory, - "benchmarks_commit": repo_meta.get("benchmarks_commit"), - "dataset_config": dataset_config, - "source_fingerprint_sha256": source_fingerprint, - } - ).encode("utf-8") - ).hexdigest() - ) - contract: dict[str, object] = { - "schema_version": 1, - "campaign_profile": str(extra.get("campaign_profile") or "ad-hoc"), - "benchmark_id": adapter.id, - "benchmark_directory": adapter.directory, - "phase": str(extra.get("campaign_phase") or "single"), - "dataset_config": dataset_config, - "corpus_sha256": corpus_sha256, - "source_fingerprint_sha256": source_fingerprint, - "model": request.model.strip(), - "provider": request.provider.strip().lower(), - "reasoning_effort": str(extra.get("reasoning_effort") or "").strip().lower(), - "harnesses": list(harnesses), - } - digest = hashlib.sha256(_canonical_json(contract).encode("utf-8")).hexdigest() - namespace = f"benchmark-phase-v1-{digest}" - return PhaseExecutionIdentity( - namespace=namespace, - contract=contract, - checkpoint_relpath=( - f".subscription-checkpoints/{namespace}/responses.jsonl" - ), - ) - - -def _relevant_source_fingerprint( - *, - workspace_root: Path, - adapter: BenchmarkAdapter, -) -> str: - """Hash dirty source/corpus inputs that a Git HEAD cannot represent.""" - - benchmarks_root = workspace_root / "suites" - candidate_roots = ( - benchmarks_root / adapter.directory, - benchmarks_root / "orchestrator", - benchmarks_root / "claude-subscription-gateway", - benchmarks_root.parent / "harnesses" / "eliza", - benchmarks_root.parent / "harnesses" / "hermes", - benchmarks_root.parent / "harnesses" / "openclaw", - workspace_root.parent / "package.json", - workspace_root.parent / "bun.lock", - workspace_root.parent / "bun.lockb", - ) - files: set[Path] = set() - for root in candidate_roots: - if root.is_file(): - files.add(root) - continue - if not root.is_dir(): - continue - for path in root.rglob("*"): - try: - relative_parts = path.relative_to(root).parts - except ValueError: - continue - if any(part in _FINGERPRINT_EXCLUDED_PARTS for part in relative_parts): - continue - if path.is_file() and not path.is_symlink(): - files.add(path) - digest = hashlib.sha256() - fingerprint_base = workspace_root.parent - for path in sorted(files, key=lambda value: value.as_posix()): - relative = path.relative_to(fingerprint_base).as_posix().encode("utf-8") - digest.update(len(relative).to_bytes(8, "big")) - digest.update(relative) - with path.open("rb") as handle: - while chunk := handle.read(1024 * 1024): - digest.update(chunk) - return digest.hexdigest() - - def _storage_integer(extra: Mapping[str, object], key: str, default: int) -> int: value = extra.get(key, default) if isinstance(value, bool) or not isinstance(value, int) or value < 0: @@ -777,6 +609,8 @@ def _find_reusable_subscription_cohort( harness: _signature_for( adapter, _effective_request(adapter, replace(request, agent=harness)), + workspace_root=workspace_root, + repo_meta=_repo_meta(workspace_root), ) for harness in harnesses } @@ -875,7 +709,7 @@ def _load_resumable_execution( ) -> dict[str, object] | None: db_path = ( workspace_root - + / "suites" / "benchmark_results" / "orchestrator.sqlite" ) diff --git a/suites/orchestrator/execution_identity.py b/suites/orchestrator/execution_identity.py new file mode 100644 index 00000000..804dc32c --- /dev/null +++ b/suites/orchestrator/execution_identity.py @@ -0,0 +1,196 @@ +"""Binds benchmark resume and provider checkpoints to complete execution inputs. + +Comparison grouping is deliberately separate from this source-sensitive cache +identity. Generated output and dependency caches do not change an execution; +runtime source, revisions, corpus, harnesses, and lockfiles do. +""" + +from __future__ import annotations + +import hashlib +import json +import os +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path + +from .types import BenchmarkAdapter, RunRequest + + +def _canonical_json(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +_STORAGE_CONTROL_KEYS = frozenset( + { + "campaign_storage_min_free_bytes", + "campaign_storage_expected_headroom_bytes", + "campaign_storage_check_interval_s", + } +) +_NAMESPACE_RUNTIME_KEYS = _STORAGE_CONTROL_KEYS | frozenset( + { + "_replace_adapter_defaults", + "campaign_silent_timeout_s", + "claude_subscription_gateway_url", + "eliza_bench_http_timeout_s", + "hermes_timeout_s", + "hl_bench_command_timeout_s", + "openclaw_timeout_s", + "timeout_s", + } +) +_FINGERPRINT_EXCLUDED_PARTS = frozenset( + { + ".cache", + ".git", + ".mypy_cache", + ".pytest_cache", + ".ruff_cache", + ".turbo", + ".venv", + "__pycache__", + "artifacts", + "benchmark_results", + "build", + "coverage", + "dist", + "node_modules", + "outputs", + "results", + "runs", + "venv", + } +) + + +@dataclass(frozen=True) +class PhaseExecutionIdentity: + namespace: str + contract: Mapping[str, object] + checkpoint_relpath: str + + +def build_phase_execution_identity( + *, + workspace_root: Path, + adapter: BenchmarkAdapter, + request: RunRequest, + harnesses: tuple[str, ...], + repo_meta: Mapping[str, str | None], +) -> PhaseExecutionIdentity: + """Build the PID/run-group-independent replay identity for one phase.""" + + extra = dict(request.extra_config) + dataset_config = { + key: value + for key, value in extra.items() + if key not in _NAMESPACE_RUNTIME_KEYS + and key + not in { + "campaign_corpus_sha256", + "campaign_phase", + "campaign_profile", + "reasoning_effort", + } + } + source_fingerprint = _relevant_source_fingerprint( + workspace_root=workspace_root, + adapter=adapter, + ) + explicit_corpus = extra.get("campaign_corpus_sha256") + if explicit_corpus is not None and ( + not isinstance(explicit_corpus, str) or not explicit_corpus.strip() + ): + raise ValueError("campaign_corpus_sha256 must be a non-empty string") + corpus_sha256 = ( + explicit_corpus.strip() + if isinstance(explicit_corpus, str) + else hashlib.sha256( + _canonical_json( + { + "adapter_directory": adapter.directory, + "benchmarks_commit": repo_meta.get("benchmarks_commit"), + "dataset_config": dataset_config, + "source_fingerprint_sha256": source_fingerprint, + } + ).encode("utf-8") + ).hexdigest() + ) + contract: dict[str, object] = { + "schema_version": 2, + "repositories": dict(repo_meta), + "campaign_profile": str(extra.get("campaign_profile") or "ad-hoc"), + "benchmark_id": adapter.id, + "benchmark_directory": adapter.directory, + "phase": str(extra.get("campaign_phase") or "single"), + "dataset_config": dataset_config, + "corpus_sha256": corpus_sha256, + "source_fingerprint_sha256": source_fingerprint, + "model": request.model.strip(), + "provider": request.provider.strip().lower(), + "reasoning_effort": str(extra.get("reasoning_effort") or "").strip().lower(), + "harnesses": list(harnesses), + } + digest = hashlib.sha256(_canonical_json(contract).encode("utf-8")).hexdigest() + namespace = f"benchmark-phase-v2-{digest}" + return PhaseExecutionIdentity( + namespace=namespace, + contract=contract, + checkpoint_relpath=(f".subscription-checkpoints/{namespace}/responses.jsonl"), + ) + + +def _relevant_source_fingerprint( + *, + workspace_root: Path, + adapter: BenchmarkAdapter, +) -> str: + """Hash dirty source/corpus inputs that a Git HEAD cannot represent.""" + + benchmarks_root = workspace_root / "suites" + candidate_roots = ( + benchmarks_root / adapter.directory, + benchmarks_root / "orchestrator", + benchmarks_root / "claude-subscription-gateway", + workspace_root / "harnesses", + workspace_root / "eliza", + workspace_root / "registry", + workspace_root / "framework", + workspace_root / "lib", + workspace_root / "package.json", + workspace_root / "bun.lock", + workspace_root / "bun.lockb", + *sorted(workspace_root.glob("*.py")), + ) + files: set[Path] = set() + for root in candidate_roots: + if root.is_file(): + files.add(root) + continue + if not root.is_dir(): + continue + for directory, subdirs, names in os.walk(root, followlinks=False): + subdirs[:] = sorted( + name + for name in subdirs + if name not in _FINGERPRINT_EXCLUDED_PARTS + and not (Path(directory) / name).is_symlink() + ) + for name in names: + path = Path(directory) / name + if name.startswith(".env") or path.is_symlink(): + continue + if path.is_file(): + files.add(path) + digest = hashlib.sha256() + fingerprint_base = workspace_root + for path in sorted(files, key=lambda value: value.as_posix()): + relative = path.relative_to(fingerprint_base).as_posix().encode("utf-8") + digest.update(len(relative).to_bytes(8, "big")) + digest.update(relative) + digest.update(path.stat().st_size.to_bytes(8, "big")) + with path.open("rb") as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() diff --git a/suites/orchestrator/runner.py b/suites/orchestrator/runner.py index 0d821535..3f294a96 100644 --- a/suites/orchestrator/runner.py +++ b/suites/orchestrator/runner.py @@ -46,6 +46,7 @@ replace_run_trajectories, update_run_result, ) +from .execution_identity import build_phase_execution_identity from .env_utils import ( git_head, load_env_file, @@ -135,7 +136,13 @@ def _sanitize_name(value: str) -> str: return cleaned or "item" -def _signature_for(adapter: BenchmarkAdapter, request: RunRequest) -> str: +def _signature_for( + adapter: BenchmarkAdapter, + request: RunRequest, + *, + workspace_root: Path, + repo_meta: dict[str, str | None], +) -> str: extra_config = dict(request.extra_config) if request.agent.strip().lower() in CALIBRATION_HARNESSES: extra_config["calibration_spec_version"] = CALIBRATION_SPEC_VERSION @@ -146,6 +153,13 @@ def _signature_for(adapter: BenchmarkAdapter, request: RunRequest) -> str: "provider": request.provider, "model": request.model, "extra_config": extra_config, + "execution_namespace": build_phase_execution_identity( + workspace_root=workspace_root, + adapter=adapter, + request=request, + harnesses=(request.agent,), + repo_meta=repo_meta, + ).namespace, } return hashlib.sha256( json.dumps( @@ -590,14 +604,13 @@ def _default_env(workspace_root: Path, request: RunRequest) -> dict[str, str]: def _repo_meta(workspace_root: Path) -> dict[str, str | None]: - benchmarks_root = workspace_root / "suites" eliza_root = workspace_root / "eliza" return { - "benchmarks_commit": git_head(benchmarks_root), - "eliza_commit": git_head(eliza_root), + "benchmarks_commit": git_head(workspace_root), + "eliza_commit": git_head(eliza_root) if (eliza_root / ".git").exists() else None, "eliza_version": safe_version_from_package_json(eliza_root / "package.json"), "benchmarks_version": safe_version_from_package_json( - benchmarks_root / "package.json" + workspace_root / "package.json" ), } @@ -3043,7 +3056,9 @@ def run_benchmarks( for benchmark_id in selected_ids: adapter = discovery.adapters[benchmark_id] effective_request = _effective_request(adapter, request) - signature = _signature_for(adapter, effective_request) + signature = _signature_for( + adapter, effective_request, workspace_root=workspace_root, repo_meta=repo_meta + ) # Harness/agent compatibility — if the harness is not in the adapter's # supported list, record an ``incompatible`` outcome and skip without diff --git a/suites/orchestrator/tests/test_cohort.py b/suites/orchestrator/tests/test_cohort.py index 4adcdfd3..07f42826 100644 --- a/suites/orchestrator/tests/test_cohort.py +++ b/suites/orchestrator/tests/test_cohort.py @@ -89,7 +89,7 @@ def _request(*benchmark_ids: str) -> RunRequest: def _workspace(tmp_path: Path) -> Path: workspace_root = tmp_path / "packages" - (workspace_root / "benchmarks").mkdir(parents=True) + (workspace_root / "suites").mkdir(parents=True) return workspace_root @@ -184,7 +184,7 @@ def fake_run_benchmarks( return ( shared_run_group_id, [_outcome(benchmark_id, request.agent)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -220,7 +220,7 @@ def fake_run_benchmarks( ) conn = connect_database( - workspace_root / "benchmark_results" / "orchestrator.sqlite" + workspace_root / "suites" / "benchmark_results" / "orchestrator.sqlite" ) groups = list_run_groups(conn) conn.close() @@ -270,7 +270,7 @@ def fake_run_benchmarks(**kwargs): return ( kwargs["shared_run_group_id"], [_outcome(benchmark_id, request.agent, status)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -305,7 +305,7 @@ def test_subscription_cohort_owns_one_gateway_and_scopes_worker_credentials( lifecycle: list[str] = [] worker_envs: dict[str, dict[str, str]] = {} audit_path = ( - workspace_root / "benchmark_results" / "gateway-audit.jsonl" + workspace_root / "suites" / "benchmark_results" / "gateway-audit.jsonl" ) class FakeGateway: @@ -333,7 +333,7 @@ def fake_run_benchmarks(**kwargs): return ( kwargs["shared_run_group_id"], [_outcome("alpha", request.agent)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) def attach_gateway(**kwargs) -> dict[str, str]: @@ -346,7 +346,7 @@ def finalize(**kwargs) -> Path: assert lifecycle[-1] == "audit-attached" kwargs["success_cleanup"]() lifecycle.append("published") - return workspace_root / "benchmark_results" / "viewer_data.json" + return workspace_root / "suites" / "benchmark_results" / "viewer_data.json" monkeypatch.setattr( cohort_module, "start_claude_subscription_gateway", start_gateway @@ -407,7 +407,7 @@ def test_subscription_resume_reuses_complete_cohort_without_gateway( "_find_reusable_subscription_cohort", lambda **_kwargs: ("rg-existing", outcomes), ) - viewer = workspace_root / "benchmark_results" / "viewer.json" + viewer = workspace_root / "suites" / "benchmark_results" / "viewer.json" monkeypatch.setattr( cohort_module, "_refresh_publication", @@ -771,7 +771,7 @@ def fake_run_benchmarks(**kwargs): if request.agent == "hermes": conn = connect_database( workspace_root - / "benchmarks" + / "suites" / "benchmark_results" / "orchestrator.sqlite" ) @@ -805,7 +805,7 @@ def fake_run_benchmarks(**kwargs): return ( run_group_id, [], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -821,7 +821,7 @@ def fake_run_benchmarks(**kwargs): assert set(raised.value.failures) == {"hermes"} assert settled == {"eliza", "openclaw"} conn = connect_database( - workspace_root / "benchmark_results" / "orchestrator.sqlite" + workspace_root / "suites" / "benchmark_results" / "orchestrator.sqlite" ) runs = list_runs(conn, limit=None) groups = list_run_groups(conn) @@ -836,7 +836,7 @@ def test_shared_runner_mode_defers_group_finish_and_publication( monkeypatch: pytest.MonkeyPatch, ) -> None: workspace_root = _workspace(tmp_path) - output_root = workspace_root / "benchmark_results" + output_root = workspace_root / "suites" / "benchmark_results" conn = connect_database(output_root / "orchestrator.sqlite") initialize_database(conn) create_run_group( @@ -948,7 +948,7 @@ def _seed_stored_execution( run_group_id: str = "rgc-prior", running_run_id: str | None = "run-prior-eliza", ) -> None: - output_root = workspace_root / "benchmark_results" + output_root = workspace_root / "suites" / "benchmark_results" output_root.mkdir(parents=True, exist_ok=True) conn = connect_database(output_root / "orchestrator.sqlite") initialize_database(conn) @@ -1037,7 +1037,7 @@ def test_resumed_cohort_retires_prior_running_rows_before_workers_start( def fake_run_benchmarks(**kwargs): conn = connect_database( - workspace_root / "benchmark_results" / "orchestrator.sqlite" + workspace_root / "suites" / "benchmark_results" / "orchestrator.sqlite" ) row = conn.execute( "SELECT status FROM benchmark_runs WHERE run_id = 'run-prior-eliza'" @@ -1047,7 +1047,7 @@ def fake_run_benchmarks(**kwargs): return ( kwargs["shared_run_group_id"], [_outcome("alpha", kwargs["request"].agent)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -1070,7 +1070,7 @@ def fake_run_benchmarks(**kwargs): assert results[0].status == cohort_module.BenchmarkCohortStatus.SUCCEEDED assert prior_status_at_worker_start == ["paused_unknown"] conn = connect_database( - workspace_root / "benchmark_results" / "orchestrator.sqlite" + workspace_root / "suites" / "benchmark_results" / "orchestrator.sqlite" ) rows = {row["run_id"]: row for row in list_runs(conn, limit=None)} prior = rows["run-prior-eliza"] @@ -1121,7 +1121,7 @@ def test_stored_future_retry_blocks_resume_unless_quota_reset_assumed( ) _seed_stored_execution(workspace_root, identity) retry_at = "2100-01-01T00:00:00+00:00" - output_root = workspace_root / "benchmark_results" + output_root = workspace_root / "suites" / "benchmark_results" conn = connect_database(output_root / "orchestrator.sqlite") pause_run_group( conn, @@ -1175,7 +1175,7 @@ def fake_run_benchmarks(**kwargs): return ( kwargs["shared_run_group_id"], [_outcome("alpha", kwargs["request"].agent)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -1201,7 +1201,7 @@ def test_prepare_cohort_recovers_stale_namespaceless_groups_only( tmp_path: Path, ) -> None: workspace_root = _workspace(tmp_path) - output_root = workspace_root / "benchmark_results" + output_root = workspace_root / "suites" / "benchmark_results" output_root.mkdir(parents=True, exist_ok=True) conn = connect_database(output_root / "orchestrator.sqlite") initialize_database(conn) diff --git a/suites/orchestrator/tests/test_cohort_provider_forwarder.py b/suites/orchestrator/tests/test_cohort_provider_forwarder.py index c32685b9..7b70b9ad 100644 --- a/suites/orchestrator/tests/test_cohort_provider_forwarder.py +++ b/suites/orchestrator/tests/test_cohort_provider_forwarder.py @@ -83,7 +83,7 @@ def _outcome(benchmark_id: str, harness: str) -> BenchmarkRunOutcome: def _workspace(tmp_path: Path) -> Path: workspace_root = tmp_path / "packages" - (workspace_root / "benchmarks").mkdir(parents=True) + (workspace_root / "suites").mkdir(parents=True) return workspace_root @@ -108,7 +108,7 @@ def fake_run_benchmarks(**kwargs): return ( kwargs["shared_run_group_id"], [_outcome("alpha", request.agent)], - workspace_root / "benchmark_results" / "viewer_data.json", + workspace_root / "suites" / "benchmark_results" / "viewer_data.json", ) monkeypatch.setattr(cohort_module, "run_benchmarks", fake_run_benchmarks) @@ -128,7 +128,7 @@ def _request(provider: str, extra_config: dict[str, object] | None = None) -> Ru def _group_statuses(workspace_root: Path) -> list[str | None]: conn = connect_database( - workspace_root / "benchmark_results" / "orchestrator.sqlite" + workspace_root / "suites" / "benchmark_results" / "orchestrator.sqlite" ) try: return [group.get("cohort_status") for group in list_run_groups(conn)] @@ -367,6 +367,8 @@ def run_and_capture(remote: bool, subdir: str) -> dict[str, str]: harness: runner_module._signature_for( adapter, runner_module._effective_request(adapter, request), + workspace_root=workspace_root, + repo_meta=runner_module._repo_meta(workspace_root), ) for harness, request in captured.items() } diff --git a/suites/orchestrator/tests/test_execution_identity.py b/suites/orchestrator/tests/test_execution_identity.py new file mode 100644 index 00000000..a154e238 --- /dev/null +++ b/suites/orchestrator/tests/test_execution_identity.py @@ -0,0 +1,194 @@ +"""Checks real filesystem inputs and SQLite reuse after runtime/source changes.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from benchmarks.orchestrator.cohort import build_phase_execution_identity +from benchmarks.orchestrator.db import ( + connect_database, + initialize_database, + insert_run_start, + get_latest_succeeded_run_for_signature, + update_run_result, +) +from benchmarks.orchestrator.runner import ( + _signature_for, + _comparison_signature_for, + _repo_meta, +) +from benchmarks.orchestrator.types import BenchmarkAdapter, RunRequest, ScoreSummary + + +def _inputs(root: Path): + adapter = BenchmarkAdapter( + id="context", + directory="context", + description="Context integration fixture", + cwd=".", + command_builder=lambda *_: [], + result_locator=lambda *_: None, + score_extractor=lambda _: ScoreSummary(1.0, "ratio", True), + agent_compatibility=("eliza",), + ) + request = RunRequest( + benchmarks=("context",), + agent="eliza", + provider="test", + model="test", + extra_config={"campaign_corpus_sha256": "pinned-corpus"}, + ) + return adapter, request + + +def _signature(root, adapter, request, meta): + return _signature_for(adapter, request, workspace_root=root, repo_meta=meta) + + +@pytest.mark.parametrize( + "relative", + [ + "eliza/packages/core/src/runtime.ts", + "suites/context/corpus.jsonl", + "harnesses/eliza/agent.ts", + "harnesses/codex/runner.py", + "run.py", + "registry/commands.py", + "framework/runner.py", + "lib/results.py", + "package.json", + "bun.lock", + ], +) +def test_changed_inputs_cannot_reuse_prior_success_or_checkpoint( + tmp_path: Path, relative: str +): + source = tmp_path / relative + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text("first complete input Ω\n", encoding="utf-8") + adapter, request = _inputs(tmp_path) + meta = {"benchmarks_commit": "benchmark-a", "eliza_commit": "runtime-a"} + signature = _signature(tmp_path, adapter, request, meta) + identity = build_phase_execution_identity( + workspace_root=tmp_path, + adapter=adapter, + request=request, + harnesses=("eliza",), + repo_meta=meta, + ) + comparison = _comparison_signature_for(adapter, request) + conn = connect_database(tmp_path / "suites/benchmark_results/orchestrator.sqlite") + try: + initialize_database(conn) + insert_run_start( + conn, + run_id="prior", + run_group_id="group", + benchmark_id=adapter.id, + benchmark_directory=adapter.directory, + signature=signature, + attempt=1, + agent=request.agent, + provider=request.provider, + model=request.model, + extra_config=dict(request.extra_config), + started_at="2026-09-14T00:00:00Z", + command=[], + cwd=str(tmp_path), + stdout_path="", + stderr_path="", + benchmark_version=None, + benchmarks_commit="benchmark-a", + eliza_commit="runtime-a", + eliza_version=None, + ) + update_run_result( + conn, + run_id="prior", + status="succeeded", + ended_at="2026-09-14T00:00:01Z", + duration_seconds=1, + score=None, + unit=None, + higher_is_better=None, + metrics={}, + result_json_path=None, + artifacts=[], + error=None, + high_score_label=None, + high_score_value=None, + delta_to_high_score=None, + ) + unchanged = _signature(tmp_path, adapter, request, meta) + assert get_latest_succeeded_run_for_signature(conn, unchanged).run_id == "prior" + assert ( + build_phase_execution_identity( + workspace_root=tmp_path, + adapter=adapter, + request=request, + harnesses=("eliza",), + repo_meta=meta, + ).checkpoint_relpath + == identity.checkpoint_relpath + ) + source.write_text("second complete input Ω\n", encoding="utf-8") + changed = _signature(tmp_path, adapter, request, meta) + assert get_latest_succeeded_run_for_signature(conn, changed) is None + assert ( + build_phase_execution_identity( + workspace_root=tmp_path, + adapter=adapter, + request=request, + harnesses=("eliza",), + repo_meta=meta, + ).checkpoint_relpath + != identity.checkpoint_relpath + ) + assert _comparison_signature_for(adapter, request) == comparison + finally: + conn.close() + + +def test_revision_changes_invalidate_even_with_explicit_corpus(tmp_path: Path): + adapter, request = _inputs(tmp_path) + initial = {"benchmarks_commit": "a", "eliza_commit": "b"} + previous = _signature(tmp_path, adapter, request, initial) + for key in initial: + assert ( + _signature(tmp_path, adapter, request, {**initial, key: "changed"}) + != previous + ) + + +def test_uninitialized_runtime_is_not_attributed_to_parent(tmp_path: Path): + subprocess.run(["git", "init", "-q", str(tmp_path)], check=True) + (tmp_path / "package.json").write_text('{"version":"1.2.3"}') + subprocess.run(["git", "-C", str(tmp_path), "add", "package.json"], check=True) + subprocess.run( + [ + "git", + "-C", + str(tmp_path), + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "commit", + "-qm", + "fixture", + ], + check=True, + ) + (tmp_path / "eliza").mkdir() + meta = _repo_meta(tmp_path) + assert ( + meta["benchmarks_commit"] + == subprocess.check_output( + ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], text=True + ).strip() + ) + assert meta["eliza_commit"] is None + assert meta["benchmarks_version"] == "1.2.3"