mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-11-06 08:35:21 +08:00
Andrew Hundt
e6126468fa
Move the full implementation from run-benchmark-campaign.py to run-benchmark-experiments.py and reduce the old filename to a backwards-compatible shim that loads the new module by path and re-exports every public name, so direct imports (tests/test_benchmark_campaign.py, scripts/autotune.py, scripts/summarize-benchmark-results.py) and existing invocations keep working with byte-identical behavior. - Add --experiment-root and --allow-temporary-experiment-root, keeping --campaign-root and --allow-temporary-campaign-root as accepted aliases on both entry points. - Add --candidate-ref LABEL=REF override and an upstream/main -> origin/main -> main fallback chain so --quick/--full keep working after the api-consolidation branch is merged and its pinned refs no longer resolve; explicitly requested refs remain fail-closed. - Say experiment instead of campaign in --help text, runtime error text, and generated-report boilerplate (summarize-benchmark-results.py); keep the campaign_version manifest key, the campaigns composition key, and the .worktrees/benchmark-campaign/ directory so every retained runset stays valid. - Move docs/BENCHMARK_CAMPAIGN.md content to docs/BENCHMARK_EXPERIMENTS.md and leave a pointer stub so existing links keep resolving. - Add tests/test_benchmark_experiments_shim.py (25 subtests) proving the shim re-exports the implementation and both CLIs accept old and new flags; update the temporary-root error-text assertion. Verified: ruff clean on all touched files; 171 passed, 1 skipped across the four benchmark test files; --help output identical between the two entry points except the program name. Signed-off-by: Andrew Hundt <ATHundt@gmail.com>
1968 lines
84 KiB
Python
Executable File
1968 lines
84 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Run an immutable, resumable benchmark experiment plan and retain an auditable disk trail.
|
|
|
|
This is the canonical entry point for benchmark experiments. `run-benchmark-campaign.py`
|
|
is a backwards-compatible shim with byte-identical behavior, kept so existing
|
|
invocations, automation, and retained runsets under `.worktrees/benchmark-campaign/`
|
|
keep working unchanged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SCHEMA_VERSION = 1
|
|
CAMPAIGN_DEFINITION_VERSION = 1
|
|
DEFAULT_MINIMUM_FREE_BYTES = 2 * 1024 * 1024 * 1024
|
|
DEFAULT_STALE_LOCK_SECONDS = 6 * 60 * 60
|
|
FILENAME_DATETIME_FORMAT = "%Y-%m-%d-%H%M%S.%fZ"
|
|
DEFAULT_CANDIDATE_REFS = (
|
|
("upstream-main", "upstream/main"),
|
|
("pre-today-major", "api-consolidation-stable-2026-07-16-semantic-v2"),
|
|
("pre-upstream-merge", "pre-upstream-main-merge-2026-07-19"),
|
|
("latest", "HEAD"),
|
|
)
|
|
# Fallback chain tried only for the built-in "upstream/main" baseline default, which
|
|
# requires a remote literally named "upstream". Era-pinned tags (pre-today-major,
|
|
# pre-upstream-merge) and every --candidate-ref override stay fail-closed: an
|
|
# unresolvable ref raises rather than silently substituting a different comparison
|
|
# point. Once api-consolidation merges to main and "upstream" stops existing, this
|
|
# lets --quick/--full keep working without editing DEFAULT_CANDIDATE_REFS.
|
|
UPSTREAM_MAIN_FALLBACK_REFS = ("upstream/main", "origin/main", "main")
|
|
IDENTITY_FIELDS = (
|
|
"identity_version",
|
|
"revision",
|
|
"binary_sha256",
|
|
"build",
|
|
"capabilities",
|
|
"capability_support",
|
|
"transport",
|
|
"scenario",
|
|
"repetition",
|
|
"harness_version",
|
|
"command",
|
|
"cwd",
|
|
"environment",
|
|
"parameters",
|
|
"timeout_seconds",
|
|
"accepted_exit_codes",
|
|
)
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def filename_datetime(moment: datetime | None = None) -> str:
|
|
"""Return a sortable, filename-safe UTC datetime with collision-level precision."""
|
|
current = moment or datetime.now(timezone.utc)
|
|
return current.astimezone(timezone.utc).strftime(FILENAME_DATETIME_FORMAT)
|
|
|
|
|
|
def campaign_version() -> str:
|
|
"""Return the sortable version of the campaign definition, not a run number."""
|
|
return f"v{CAMPAIGN_DEFINITION_VERSION:04d}"
|
|
|
|
|
|
def runset_identity(spec_payload: bytes) -> str:
|
|
"""Identify an immutable runset so preparing the same spec resumes in place."""
|
|
return hashlib.sha256(spec_payload).hexdigest()[:12]
|
|
|
|
|
|
def automatic_runset_identity(spec: dict[str, Any]) -> str:
|
|
"""Hash semantic inputs while allowing an identical runset to be path-remapped."""
|
|
normalized = json.loads(json.dumps(spec))
|
|
normalized.pop("runset_id", None)
|
|
normalized.pop("benchmark_script", None)
|
|
normalized.pop("cwd", None)
|
|
for background_key in ("repository_background", "quality_background"):
|
|
background = normalized.get(background_key)
|
|
if isinstance(background, dict):
|
|
background.pop("repo", None)
|
|
candidates = normalized.get("candidates")
|
|
if isinstance(candidates, list):
|
|
for candidate in candidates:
|
|
if isinstance(candidate, dict):
|
|
candidate.pop("binary", None)
|
|
return runset_identity(canonical_json(normalized))
|
|
|
|
|
|
def _validate_runset_identity(runset: str) -> str:
|
|
if len(runset) != 12 or any(char not in "0123456789abcdef" for char in runset):
|
|
raise ValueError(f"runset identity must be 12 lowercase hexadecimal characters: {runset!r}")
|
|
return runset
|
|
|
|
|
|
def automatic_campaign_name(preset: str, source: dict[str, str], runset: str) -> str:
|
|
"""Name a resumable campaign without confusing source and execution datetimes."""
|
|
if preset not in {"quick", "full"}:
|
|
raise ValueError(f"automatic preset must be quick or full: {preset!r}")
|
|
revision = source.get("revision", "")
|
|
commit_datetime = source.get("commit_datetime_slug", "")
|
|
if len(revision) != 40 or not commit_datetime:
|
|
raise ValueError("source must contain a full revision and commit_datetime_slug")
|
|
return (
|
|
f"{campaign_version()}-{preset}-commit-{commit_datetime}-{revision[:12]}-"
|
|
f"runset-{_validate_runset_identity(runset)}"
|
|
)
|
|
|
|
|
|
def automatic_spec_name(preset: str, runset: str) -> str:
|
|
if preset not in {"quick", "full"}:
|
|
raise ValueError(f"automatic preset must be quick or full: {preset!r}")
|
|
return f"spec-{campaign_version()}-{preset}-runset-{_validate_runset_identity(runset)}.json"
|
|
|
|
|
|
def generated_artifact_name(
|
|
kind: str,
|
|
runset: str,
|
|
suffix: str,
|
|
*,
|
|
preset: str | None = None,
|
|
moment: datetime | None = None,
|
|
nonce: str | None = None,
|
|
) -> str:
|
|
"""Name generated evidence while keeping its stable runset identity visible."""
|
|
if not kind or any(not (char.isalnum() or char == "-") for char in kind):
|
|
raise ValueError(f"artifact kind is not path-safe: {kind!r}")
|
|
if preset is not None and preset not in {"quick", "full", "custom"}:
|
|
raise ValueError(f"artifact preset is invalid: {preset!r}")
|
|
if not suffix.startswith(".") or "/" in suffix:
|
|
raise ValueError(f"artifact suffix is invalid: {suffix!r}")
|
|
if nonce is not None and (
|
|
not nonce or any(not (char.isalnum() or char in "-_") for char in nonce)
|
|
):
|
|
raise ValueError(f"artifact nonce is not path-safe: {nonce!r}")
|
|
parts = [kind, campaign_version()]
|
|
if preset is not None:
|
|
parts.append(preset)
|
|
parts.extend(("runset", _validate_runset_identity(runset), "generated", filename_datetime(moment)))
|
|
if nonce is not None:
|
|
parts.append(nonce)
|
|
return "-".join(parts) + suffix
|
|
|
|
|
|
def _run_text(command: list[str], *, cwd: Path) -> str:
|
|
process = subprocess.run(command, cwd=cwd, capture_output=True, text=True, check=False)
|
|
if process.returncode != 0:
|
|
detail = process.stderr.strip() or process.stdout.strip() or "no diagnostic"
|
|
raise RuntimeError(f"command failed ({process.returncode}): {' '.join(command)}: {detail}")
|
|
return process.stdout.strip()
|
|
|
|
|
|
def resolve_commit(repository: Path, ref: str) -> str:
|
|
"""Peel a branch, tag, or commit ref to the full commit object ID."""
|
|
revision = _run_text(
|
|
["git", "rev-parse", "--verify", f"{ref}^{{commit}}"],
|
|
cwd=repository,
|
|
)
|
|
if len(revision) != 40 or any(char not in "0123456789abcdef" for char in revision.lower()):
|
|
raise RuntimeError(f"git resolved {ref!r} to an invalid commit ID: {revision!r}")
|
|
return revision.lower()
|
|
|
|
|
|
def commit_identity(repository: Path, ref: str) -> dict[str, str]:
|
|
"""Return a peeled commit, its repository datetime, and its exact tree."""
|
|
revision = resolve_commit(repository, ref)
|
|
committed_at = _run_text(["git", "show", "-s", "--format=%cI", revision], cwd=repository)
|
|
try:
|
|
parsed = datetime.fromisoformat(committed_at.replace("Z", "+00:00"))
|
|
except ValueError as error:
|
|
raise RuntimeError(
|
|
f"git returned an invalid commit datetime for {revision}: {committed_at!r}"
|
|
) from error
|
|
tree = _run_text(["git", "rev-parse", "--verify", f"{revision}^{{tree}}"], cwd=repository)
|
|
return {
|
|
"revision": revision,
|
|
"committed_at": parsed.isoformat(),
|
|
"commit_datetime_slug": parsed.strftime("%Y-%m-%d-%H%M"),
|
|
"tree": tree,
|
|
}
|
|
|
|
|
|
def resolve_default_candidate_ref(repository: Path, label: str, ref: str) -> str:
|
|
"""Resolve a default candidate ref, retrying survivable baseline aliases only.
|
|
|
|
Only the built-in "upstream/main" baseline gets a fallback chain (see
|
|
UPSTREAM_MAIN_FALLBACK_REFS), because it is the one default expected to age past
|
|
a merge: the remote may be renamed or absent in a fresh clone. Era-pinned tag
|
|
defaults and explicit --candidate-ref overrides are not touched here and remain
|
|
fail-closed in materialize_candidate: an unresolvable ref raises a clear error
|
|
instead of silently running a different comparison.
|
|
"""
|
|
del label
|
|
if ref != "upstream/main":
|
|
return ref
|
|
for candidate_ref in UPSTREAM_MAIN_FALLBACK_REFS:
|
|
try:
|
|
resolve_commit(repository, candidate_ref)
|
|
except RuntimeError:
|
|
continue
|
|
return candidate_ref
|
|
return ref
|
|
|
|
|
|
def parse_candidate_ref_override(value: str) -> tuple[str, str]:
|
|
"""Parse one repeatable --candidate-ref LABEL=REF argument."""
|
|
label, separator, ref = value.partition("=")
|
|
if not separator or not label or not ref:
|
|
raise ValueError(f"--candidate-ref must be LABEL=REF: {value!r}")
|
|
known_labels = {default_label for default_label, _ in DEFAULT_CANDIDATE_REFS}
|
|
if label not in known_labels:
|
|
raise ValueError(f"--candidate-ref label must be one of {sorted(known_labels)}: {label!r}")
|
|
return label, ref
|
|
|
|
|
|
def _path_within(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(root.resolve())
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _candidate_slug(label: str) -> str:
|
|
if not label or any(not (char.isalnum() or char in "-_") for char in label):
|
|
raise ValueError(f"candidate label is not path-safe: {label!r}")
|
|
return label
|
|
|
|
|
|
def ensure_clean_tracked_worktree(repository: Path, role: str) -> None:
|
|
"""Reject tracked edits while allowing ignored retained evidence and build output."""
|
|
tracked_status = _run_text(
|
|
["git", "status", "--porcelain", "--untracked-files=no"], cwd=repository
|
|
)
|
|
if tracked_status:
|
|
raise RuntimeError(
|
|
f"{role} has tracked modifications; commit or restore them before measurement: "
|
|
f"{repository}"
|
|
)
|
|
|
|
|
|
def _registered_candidate_worktrees(repository: Path, candidate_root: Path, revision: str) -> list[Path]:
|
|
listing = _run_text(["git", "worktree", "list", "--porcelain"], cwd=repository)
|
|
matches: list[Path] = []
|
|
for block in listing.split("\n\n"):
|
|
fields: dict[str, str] = {}
|
|
for line in block.splitlines():
|
|
key, separator, value = line.partition(" ")
|
|
if separator:
|
|
fields[key] = value
|
|
path_value = fields.get("worktree")
|
|
if fields.get("HEAD") == revision and path_value:
|
|
candidate = Path(path_value).resolve()
|
|
if _path_within(candidate, candidate_root):
|
|
matches.append(candidate)
|
|
return sorted(matches)
|
|
|
|
|
|
def _compiler_identity(worktree: Path) -> str:
|
|
try:
|
|
return _run_text(["cc", "--version"], cwd=worktree).splitlines()[0]
|
|
except (OSError, RuntimeError, IndexError):
|
|
return "unknown (see datetime-named build log)"
|
|
|
|
|
|
def _production_cflags(worktree: Path) -> str:
|
|
"""Read the candidate Makefile's canonical production flags without duplicating them."""
|
|
target = "cbm-print-production-flags"
|
|
definition = f"{target}:\n\t@printf '%s\\n' '$(CFLAGS_PROD)'\n"
|
|
try:
|
|
process = subprocess.run(
|
|
[
|
|
"make",
|
|
"-s",
|
|
"-f",
|
|
"Makefile.cbm",
|
|
"-f",
|
|
"-",
|
|
target,
|
|
],
|
|
cwd=worktree,
|
|
input=definition,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except OSError:
|
|
return "unknown (see candidate Makefile.cbm and build log)"
|
|
if process.returncode != 0:
|
|
return "unknown (see candidate Makefile.cbm and build log)"
|
|
value = process.stdout.strip()
|
|
return value or "not declared by candidate Makefile.cbm"
|
|
|
|
|
|
def _candidate_capability_support(label: str) -> dict[str, bool]:
|
|
if label == "upstream-main":
|
|
return {
|
|
"rank": False,
|
|
"dependencies": False,
|
|
"similarity": True,
|
|
"semantic_edges": True,
|
|
"git_history": True,
|
|
"http_links": False,
|
|
}
|
|
return {
|
|
"rank": True,
|
|
"dependencies": True,
|
|
"similarity": True,
|
|
"semantic_edges": True,
|
|
"git_history": True,
|
|
"http_links": True,
|
|
}
|
|
|
|
|
|
def materialize_candidate(
|
|
repository: Path,
|
|
candidate_root: Path,
|
|
label: str,
|
|
ref: str,
|
|
*,
|
|
jobs: int = 2,
|
|
) -> dict[str, Any]:
|
|
"""Resolve, isolate, production-build, and hash one benchmark candidate."""
|
|
repository = repository.expanduser().resolve()
|
|
candidate_root = candidate_root.expanduser().resolve()
|
|
safe_label = _candidate_slug(label)
|
|
if jobs <= 0:
|
|
raise ValueError("build jobs must be positive")
|
|
source_identity = commit_identity(repository, ref)
|
|
revision = source_identity["revision"]
|
|
candidate_root.mkdir(parents=True, exist_ok=True)
|
|
intended = candidate_root / f"{safe_label}-{revision[:12]}"
|
|
matches = _registered_candidate_worktrees(repository, candidate_root, revision)
|
|
if intended in matches:
|
|
worktree = intended
|
|
elif matches:
|
|
worktree = matches[0]
|
|
else:
|
|
if intended.exists():
|
|
raise RuntimeError(f"candidate path exists but is not the registered {revision} worktree: {intended}")
|
|
process = subprocess.run(
|
|
["git", "worktree", "add", "--detach", str(intended), revision],
|
|
cwd=repository,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if process.returncode != 0:
|
|
detail = process.stderr.strip() or process.stdout.strip() or "no diagnostic"
|
|
raise RuntimeError(f"could not create candidate worktree {intended}: {detail}")
|
|
worktree = intended
|
|
actual_revision = resolve_commit(worktree, "HEAD")
|
|
if actual_revision != revision:
|
|
raise RuntimeError(
|
|
f"candidate worktree HEAD mismatch: expected={revision} actual={actual_revision} path={worktree}"
|
|
)
|
|
ensure_clean_tracked_worktree(worktree, "candidate worktree")
|
|
|
|
binary = worktree / "build" / "c" / "codebase-memory-mcp"
|
|
stable_build = {
|
|
"target": f"make -j{jobs} -f Makefile.cbm cbm",
|
|
"compiler": _compiler_identity(worktree),
|
|
"cflags": _production_cflags(worktree),
|
|
"source_commit_datetime": source_identity["committed_at"],
|
|
"source_tree": source_identity["tree"],
|
|
}
|
|
cache_path = (
|
|
candidate_root
|
|
/ "cache"
|
|
/ f"candidate-{campaign_version()}-{safe_label}-commit-{revision[:12]}.json"
|
|
)
|
|
if cache_path.is_file() and binary.is_file():
|
|
try:
|
|
cached = read_json_object(cache_path).get("candidate")
|
|
if (
|
|
isinstance(cached, dict)
|
|
and cached.get("label") == safe_label
|
|
and cached.get("revision") == revision
|
|
and cached.get("binary") == str(binary)
|
|
and cached.get("build") == stable_build
|
|
and cached.get("tree") == source_identity["tree"]
|
|
and cached.get("binary_sha256") == file_sha256(binary)
|
|
):
|
|
return cached
|
|
except (OSError, ValueError, json.JSONDecodeError):
|
|
pass
|
|
|
|
stamp = filename_datetime()
|
|
log_root = candidate_root / "build-logs"
|
|
log_root.mkdir(parents=True, exist_ok=True)
|
|
build_log = log_root / f"generated-{stamp}-for-{safe_label}-commit-{revision[:12]}.log"
|
|
command = ["make", f"-j{jobs}", "-f", "Makefile.cbm", "cbm"]
|
|
with build_log.open("w", encoding="utf-8") as stream:
|
|
stream.write(f"started_at_utc={utc_now()}\n")
|
|
stream.write(f"revision={revision}\n")
|
|
stream.write(f"command={' '.join(command)}\n")
|
|
stream.flush()
|
|
process = subprocess.run(
|
|
command,
|
|
cwd=worktree,
|
|
stdout=stream,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
stream.write(f"finished_at_utc={utc_now()}\n")
|
|
stream.write(f"exit_code={process.returncode}\n")
|
|
if process.returncode != 0:
|
|
raise RuntimeError(f"candidate production build failed ({process.returncode}); see {build_log}")
|
|
if not binary.is_file():
|
|
raise RuntimeError(f"candidate build did not produce {binary}; see {build_log}")
|
|
candidate = {
|
|
"label": safe_label,
|
|
"revision": revision,
|
|
"binary": str(binary),
|
|
"binary_sha256": file_sha256(binary),
|
|
"build": stable_build,
|
|
"capability_support": _candidate_capability_support(safe_label),
|
|
"commit_datetime": source_identity["committed_at"],
|
|
"tree": source_identity["tree"],
|
|
}
|
|
metadata_root = candidate_root / "metadata"
|
|
metadata_root.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(
|
|
metadata_root / f"generated-{stamp}-for-{safe_label}-commit-{revision[:12]}.json",
|
|
{
|
|
**candidate,
|
|
"ref": ref,
|
|
"worktree": str(worktree),
|
|
"build_log": str(build_log),
|
|
"recorded_at_utc": utc_now(),
|
|
},
|
|
)
|
|
atomic_write_json(
|
|
cache_path,
|
|
{
|
|
"schema_version": SCHEMA_VERSION,
|
|
"campaign_version": campaign_version(),
|
|
"candidate": candidate,
|
|
},
|
|
)
|
|
return candidate
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def artifact_manifest(root: Path) -> dict[str, Any]:
|
|
files = []
|
|
total_bytes = 0
|
|
if root.is_dir():
|
|
for path in sorted(item for item in root.rglob("*") if item.is_file()):
|
|
size = path.stat().st_size
|
|
total_bytes += size
|
|
files.append(
|
|
{
|
|
"path": path.relative_to(root).as_posix(),
|
|
"size_bytes": size,
|
|
"sha256": file_sha256(path),
|
|
}
|
|
)
|
|
return {"file_count": len(files), "total_bytes": total_bytes, "files": files}
|
|
|
|
|
|
def canonical_json(value: Any) -> bytes:
|
|
return json.dumps(value, separators=(",", ":"), sort_keys=True).encode("utf-8")
|
|
|
|
|
|
def atomic_write_json(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.parent / f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp"
|
|
payload = json.dumps(value, indent=2, sort_keys=True) + "\n"
|
|
try:
|
|
with temporary.open("w", encoding="utf-8") as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
try:
|
|
directory_fd = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
except OSError:
|
|
# Some filesystems do not support directory fsync. The file itself
|
|
# is still synced before the atomic replacement.
|
|
pass
|
|
finally:
|
|
if temporary.exists():
|
|
temporary.unlink()
|
|
|
|
|
|
def atomic_write_bytes(path: Path, payload: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.parent / f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp"
|
|
try:
|
|
with temporary.open("wb") as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if temporary.exists():
|
|
temporary.unlink()
|
|
|
|
|
|
def read_json_object(path: Path) -> dict[str, Any]:
|
|
with path.open(encoding="utf-8") as stream:
|
|
value = json.load(stream)
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"expected JSON object: {path}")
|
|
return value
|
|
|
|
|
|
def build_automatic_spec(
|
|
repository: Path,
|
|
benchmark_script: Path,
|
|
candidates: list[dict[str, Any]],
|
|
*,
|
|
preset: str,
|
|
) -> dict[str, Any]:
|
|
"""Build the canonical safe quick or repeated full capability matrix."""
|
|
if preset not in {"quick", "full"}:
|
|
raise ValueError("preset must be quick or full")
|
|
repository = repository.expanduser().resolve()
|
|
benchmark_script = benchmark_script.expanduser().resolve()
|
|
if not benchmark_script.is_file():
|
|
raise ValueError(f"benchmark script does not exist: {benchmark_script}")
|
|
expected_labels = [label for label, _ in DEFAULT_CANDIDATE_REFS]
|
|
actual_labels = [candidate.get("label") for candidate in candidates]
|
|
if actual_labels != expected_labels:
|
|
raise ValueError(f"automatic candidates must be ordered {expected_labels}, got {actual_labels}")
|
|
repository_identity = commit_identity(repository, "HEAD")
|
|
repository_revision = repository_identity["revision"]
|
|
repository_tree = repository_identity["tree"]
|
|
runner_sha = file_sha256(Path(__file__).resolve())
|
|
benchmark_sha = file_sha256(benchmark_script)
|
|
latest_labels = ["latest"]
|
|
profiles: list[dict[str, Any]] = [{"label": "default", "config_profile": "default", "capabilities": {}}]
|
|
if preset == "full":
|
|
profiles.extend(
|
|
(
|
|
{
|
|
"label": "upstream-equivalent",
|
|
"config_profile": "default",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {
|
|
"auto_index_deps": "false",
|
|
"rank_enabled": "false",
|
|
"httplinks_enabled": "false",
|
|
},
|
|
"config_overrides": {
|
|
"auto_index_deps": "false",
|
|
"rank_enabled": "false",
|
|
"httplinks_enabled": "false",
|
|
},
|
|
},
|
|
{
|
|
"label": "eager-derived-freshness",
|
|
"config_profile": "incremental_semantic_freshness_eager",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"incremental_derived_refresh": "eager"},
|
|
},
|
|
{
|
|
"label": "rank-disabled",
|
|
"config_profile": "rank_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"rank_enabled": "false"},
|
|
},
|
|
{
|
|
"label": "dependency-disabled",
|
|
"config_profile": "dependency_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"auto_index_deps": "false"},
|
|
},
|
|
{
|
|
"label": "similarity-disabled",
|
|
"config_profile": "similarity_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"similarity_enabled": "false"},
|
|
},
|
|
{
|
|
"label": "semantic-edges-disabled",
|
|
"config_profile": "semantic_edges_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"semantic_edges_enabled": "false"},
|
|
},
|
|
{
|
|
"label": "git-history-disabled",
|
|
"config_profile": "git_history_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"githistory_enabled": "false"},
|
|
},
|
|
{
|
|
"label": "http-links-disabled",
|
|
"config_profile": "http_links_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {"httplinks_enabled": "false"},
|
|
},
|
|
{
|
|
"label": "optional-graph-disabled",
|
|
"config_profile": "optional_graph_disabled",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {
|
|
"rank_enabled": "false",
|
|
"similarity_enabled": "false",
|
|
"semantic_edges_enabled": "false",
|
|
"githistory_enabled": "false",
|
|
"httplinks_enabled": "false",
|
|
},
|
|
},
|
|
{
|
|
"label": "minimal-indexing",
|
|
"config_profile": "minimal_indexing",
|
|
"candidate_labels": latest_labels,
|
|
"capabilities": {
|
|
"auto_index_deps": "false",
|
|
"rank_enabled": "false",
|
|
"similarity_enabled": "false",
|
|
"semantic_edges_enabled": "false",
|
|
"githistory_enabled": "false",
|
|
"httplinks_enabled": "false",
|
|
},
|
|
},
|
|
)
|
|
)
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"campaign_version": campaign_version(),
|
|
"identity_version": 2,
|
|
"harness_version": (f"automatic-{preset}:benchmark-{benchmark_sha}:runner-{runner_sha}"),
|
|
"benchmark_script": str(benchmark_script),
|
|
"workload": "self_dogfood",
|
|
"repository_background": {
|
|
"repo": str(repository),
|
|
"revision": repository_revision,
|
|
"tree": repository_tree,
|
|
"commit_datetime": repository_identity["committed_at"],
|
|
},
|
|
"index_mode": "fast" if preset == "quick" else "moderate",
|
|
"execution_order": "paired_interleaved",
|
|
"cwd": str(repository),
|
|
"timeout_seconds": 900,
|
|
"cell_timeout_seconds": 1800,
|
|
"accepted_exit_codes": [0, 1],
|
|
"repetitions": 1 if preset == "quick" else 3,
|
|
"transports": ["mcp"],
|
|
"candidates": candidates,
|
|
"profiles": profiles,
|
|
"scenarios": [{"name": "c_new_leaf"}],
|
|
}
|
|
|
|
|
|
def identity_document(cell: dict[str, Any]) -> dict[str, Any]:
|
|
if cell.get("identity_version") != 2:
|
|
return {key: cell.get(key) for key in IDENTITY_FIELDS if key != "identity_version"}
|
|
document = {key: cell.get(key) for key in IDENTITY_FIELDS}
|
|
|
|
command = list(document.get("command") or [])
|
|
if command:
|
|
command[0] = "{benchmark_script}"
|
|
for flag, replacement in (
|
|
("--binary", "{candidate_binary}"),
|
|
("--repo-root", "{repository_root}"),
|
|
("--quality-background-repo", "{quality_background_root}"),
|
|
):
|
|
for index, token in enumerate(command[:-1]):
|
|
if token == flag:
|
|
command[index + 1] = replacement
|
|
document["command"] = command
|
|
if document.get("cwd") is not None:
|
|
document["cwd"] = "{working_directory}"
|
|
parameters = json.loads(json.dumps(document.get("parameters") or {}))
|
|
for background_key in ("repository_background", "quality_background"):
|
|
background = parameters.get(background_key)
|
|
if isinstance(background, dict) and "repo" in background:
|
|
background["repo"] = f"{{{background_key}_root}}"
|
|
document["parameters"] = parameters
|
|
return document
|
|
|
|
|
|
def cell_identity(cell: dict[str, Any]) -> str:
|
|
return hashlib.sha256(canonical_json(identity_document(cell))).hexdigest()[:24]
|
|
|
|
|
|
def validate_cell(cell: dict[str, Any], index: int) -> None:
|
|
required = {
|
|
"label": str,
|
|
"revision": str,
|
|
"binary_sha256": str,
|
|
"build": dict,
|
|
"capabilities": dict,
|
|
"transport": str,
|
|
"scenario": str,
|
|
"repetition": int,
|
|
"harness_version": str,
|
|
"command": list,
|
|
}
|
|
for key, expected_type in required.items():
|
|
if not isinstance(cell.get(key), expected_type):
|
|
raise ValueError(f"cells[{index}].{key} must be {expected_type.__name__}")
|
|
if not cell["label"] or "=" in cell["label"]:
|
|
raise ValueError(f"cells[{index}].label must be non-empty and cannot contain '='")
|
|
if len(cell["revision"]) != 40:
|
|
raise ValueError(f"cells[{index}].revision must be a full 40-character commit hash")
|
|
if len(cell["binary_sha256"]) != 64:
|
|
raise ValueError(f"cells[{index}].binary_sha256 must be a full SHA-256")
|
|
if not cell["command"] or not all(isinstance(item, str) for item in cell["command"]):
|
|
raise ValueError(f"cells[{index}].command must be a non-empty string array")
|
|
accepted = cell.get("accepted_exit_codes", [0])
|
|
if not isinstance(accepted, list) or not accepted or not all(isinstance(code, int) for code in accepted):
|
|
raise ValueError(f"cells[{index}].accepted_exit_codes must be a non-empty integer array")
|
|
support = cell.get("capability_support")
|
|
if support is not None and (
|
|
not isinstance(support, dict)
|
|
or not all(isinstance(key, str) and isinstance(value, bool) for key, value in support.items())
|
|
):
|
|
raise ValueError(f"cells[{index}].capability_support must be a string-to-boolean object")
|
|
|
|
|
|
def validate_plan(plan: dict[str, Any]) -> list[dict[str, Any]]:
|
|
if plan.get("schema_version") != SCHEMA_VERSION:
|
|
raise ValueError(f"schema_version must be {SCHEMA_VERSION}")
|
|
cells = plan.get("cells")
|
|
if not isinstance(cells, list) or not cells:
|
|
raise ValueError("cells must be a non-empty array")
|
|
typed_cells: list[dict[str, Any]] = []
|
|
identities: set[str] = set()
|
|
for index, value in enumerate(cells):
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"cells[{index}] must be an object")
|
|
validate_cell(value, index)
|
|
identity = cell_identity(value)
|
|
if identity in identities:
|
|
raise ValueError(f"duplicate cell identity at cells[{index}]: {identity}")
|
|
identities.add(identity)
|
|
typed_cells.append(value)
|
|
return typed_cells
|
|
|
|
|
|
def _string_map(value: Any, field: str) -> dict[str, str]:
|
|
if value is None:
|
|
return {}
|
|
if not isinstance(value, dict) or not all(
|
|
isinstance(key, str) and isinstance(item, str) for key, item in value.items()
|
|
):
|
|
raise ValueError(f"{field} must be a string-to-string object")
|
|
return dict(value)
|
|
|
|
|
|
def _nonempty_list(value: Any, field: str) -> list[Any]:
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError(f"{field} must be a non-empty array")
|
|
return value
|
|
|
|
|
|
def _optional_iso_datetime(value: Any, field: str) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"{field} must be an ISO 8601 datetime string")
|
|
try:
|
|
datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as error:
|
|
raise ValueError(f"{field} must be an ISO 8601 datetime string") from error
|
|
return value
|
|
|
|
|
|
def expand_matrix_spec(spec: dict[str, Any]) -> dict[str, Any]:
|
|
"""Expand a compact benchmark grid into immutable campaign cells."""
|
|
if spec.get("schema_version") != SCHEMA_VERSION:
|
|
raise ValueError(f"schema_version must be {SCHEMA_VERSION}")
|
|
harness_version = spec.get("harness_version")
|
|
benchmark_script = spec.get("benchmark_script")
|
|
cwd = spec.get("cwd")
|
|
repetitions = spec.get("repetitions")
|
|
benchmark_timeout = spec.get("timeout_seconds", 240)
|
|
index_mode = spec.get("index_mode", "fast")
|
|
accepted_exit_codes = spec.get("accepted_exit_codes", [0])
|
|
capability_quality = spec.get("capability_quality")
|
|
workload = spec.get("workload", "matrix")
|
|
identity_version = spec.get("identity_version", 1)
|
|
execution_order = spec.get("execution_order")
|
|
quality_background = spec.get("quality_background")
|
|
repository_background = spec.get("repository_background")
|
|
if not isinstance(harness_version, str) or not harness_version:
|
|
raise ValueError("harness_version must be a non-empty string")
|
|
if not isinstance(benchmark_script, str) or not benchmark_script:
|
|
raise ValueError("benchmark_script must be a non-empty string")
|
|
if not isinstance(cwd, str) or not cwd:
|
|
raise ValueError("cwd must be a non-empty string")
|
|
if not isinstance(repetitions, int) or repetitions <= 0:
|
|
raise ValueError("repetitions must be a positive integer")
|
|
if not isinstance(benchmark_timeout, int) or benchmark_timeout <= 0:
|
|
raise ValueError("timeout_seconds must be a positive integer")
|
|
if index_mode not in {"fast", "moderate", "full"}:
|
|
raise ValueError("index_mode must be fast, moderate, or full")
|
|
if execution_order not in {None, "grouped", "paired_interleaved"}:
|
|
raise ValueError("execution_order must be grouped or paired_interleaved")
|
|
if capability_quality is not None and (
|
|
not isinstance(capability_quality, str) or not capability_quality or "=" in capability_quality
|
|
):
|
|
raise ValueError("capability_quality must be a non-empty argument value")
|
|
if workload not in {"matrix", "self_dogfood"}:
|
|
raise ValueError("workload must be matrix or self_dogfood")
|
|
if identity_version not in {1, 2}:
|
|
raise ValueError("identity_version must be 1 or 2")
|
|
if capability_quality is not None and workload != "matrix":
|
|
raise ValueError("capability_quality cannot be combined with a self_dogfood workload")
|
|
if quality_background is not None:
|
|
if capability_quality not in {"similarity", "semantic_edges"}:
|
|
raise ValueError("quality_background requires capability_quality similarity or semantic_edges")
|
|
if not isinstance(quality_background, dict):
|
|
raise ValueError("quality_background must be an object")
|
|
background_repo = quality_background.get("repo")
|
|
background_revision = quality_background.get("revision")
|
|
background_tree = quality_background.get("tree")
|
|
background_datetime = _optional_iso_datetime(
|
|
quality_background.get("commit_datetime"), "quality_background.commit_datetime"
|
|
)
|
|
if not isinstance(background_repo, str) or not Path(background_repo).is_dir():
|
|
raise ValueError("quality_background.repo must be an existing directory")
|
|
if not isinstance(background_revision, str) or len(background_revision) != 40:
|
|
raise ValueError("quality_background.revision must be a full commit hash")
|
|
if not isinstance(background_tree, str) or len(background_tree) != 40:
|
|
raise ValueError("quality_background.tree must be a full tree hash")
|
|
quality_background = {
|
|
"repo": str(Path(background_repo).expanduser().resolve()),
|
|
"revision": background_revision,
|
|
"tree": background_tree,
|
|
}
|
|
if background_datetime is not None:
|
|
quality_background["commit_datetime"] = background_datetime
|
|
if repository_background is not None:
|
|
if workload != "self_dogfood":
|
|
raise ValueError("repository_background requires workload self_dogfood")
|
|
if not isinstance(repository_background, dict):
|
|
raise ValueError("repository_background must be an object")
|
|
background_repo = repository_background.get("repo")
|
|
background_revision = repository_background.get("revision")
|
|
background_tree = repository_background.get("tree")
|
|
background_datetime = _optional_iso_datetime(
|
|
repository_background.get("commit_datetime"),
|
|
"repository_background.commit_datetime",
|
|
)
|
|
if not isinstance(background_repo, str) or not Path(background_repo).is_dir():
|
|
raise ValueError("repository_background.repo must be an existing directory")
|
|
if not isinstance(background_revision, str) or len(background_revision) != 40:
|
|
raise ValueError("repository_background.revision must be a full commit hash")
|
|
if not isinstance(background_tree, str) or len(background_tree) != 40:
|
|
raise ValueError("repository_background.tree must be a full tree hash")
|
|
repository_background = {
|
|
"repo": str(Path(background_repo).expanduser().resolve()),
|
|
"revision": background_revision,
|
|
"tree": background_tree,
|
|
}
|
|
if background_datetime is not None:
|
|
repository_background["commit_datetime"] = background_datetime
|
|
elif workload == "self_dogfood":
|
|
raise ValueError("workload self_dogfood requires repository_background")
|
|
if (
|
|
not isinstance(accepted_exit_codes, list)
|
|
or not accepted_exit_codes
|
|
or not all(isinstance(code, int) and not isinstance(code, bool) for code in accepted_exit_codes)
|
|
):
|
|
raise ValueError("accepted_exit_codes must be a non-empty integer array")
|
|
cell_timeout = spec.get("cell_timeout_seconds", benchmark_timeout * 4)
|
|
if not isinstance(cell_timeout, int) or cell_timeout <= 0:
|
|
raise ValueError("cell_timeout_seconds must be a positive integer")
|
|
benchmark_path = Path(benchmark_script).expanduser().resolve()
|
|
if not benchmark_path.is_file():
|
|
raise ValueError(f"benchmark_script does not exist: {benchmark_path}")
|
|
benchmark_sha256 = file_sha256(benchmark_path)
|
|
|
|
candidates = _nonempty_list(spec.get("candidates"), "candidates")
|
|
candidate_labels = {item.get("label") for item in candidates if isinstance(item, dict)}
|
|
profiles = _nonempty_list(spec.get("profiles"), "profiles")
|
|
scenarios = (
|
|
[{"name": f"{capability_quality}_quality"}]
|
|
if capability_quality is not None
|
|
else _nonempty_list(spec.get("scenarios"), "scenarios")
|
|
)
|
|
transports = _nonempty_list(spec.get("transports"), "transports")
|
|
if not all(isinstance(item, str) and item for item in transports):
|
|
raise ValueError("transports must contain non-empty strings")
|
|
common_environment = _string_map(spec.get("environment"), "environment")
|
|
|
|
cells: list[dict[str, Any]] = []
|
|
for candidate_index, candidate in enumerate(candidates):
|
|
if not isinstance(candidate, dict):
|
|
raise ValueError(f"candidates[{candidate_index}] must be an object")
|
|
candidate_label = candidate.get("label")
|
|
revision = candidate.get("revision")
|
|
binary_value = candidate.get("binary")
|
|
build = candidate.get("build")
|
|
if not isinstance(candidate_label, str) or not candidate_label or "=" in candidate_label:
|
|
raise ValueError(f"candidates[{candidate_index}].label is invalid")
|
|
if not isinstance(revision, str) or len(revision) != 40:
|
|
raise ValueError(f"candidates[{candidate_index}].revision must be a full commit hash")
|
|
if not isinstance(binary_value, str) or not binary_value:
|
|
raise ValueError(f"candidates[{candidate_index}].binary must be a path string")
|
|
if not isinstance(build, dict):
|
|
raise ValueError(f"candidates[{candidate_index}].build must be an object")
|
|
binary = Path(binary_value).expanduser().resolve()
|
|
if not binary.is_file():
|
|
raise ValueError(f"candidate binary does not exist: {binary}")
|
|
binary_sha = file_sha256(binary)
|
|
declared_sha = candidate.get("binary_sha256")
|
|
if declared_sha is not None and declared_sha != binary_sha:
|
|
raise ValueError(f"candidates[{candidate_index}].binary_sha256 does not match {binary}")
|
|
candidate_environment = _string_map(candidate.get("environment"), f"candidates[{candidate_index}].environment")
|
|
candidate_support = candidate.get("capability_support")
|
|
if candidate_support is not None and (
|
|
not isinstance(candidate_support, dict)
|
|
or not all(isinstance(key, str) and isinstance(value, bool) for key, value in candidate_support.items())
|
|
):
|
|
raise ValueError(f"candidates[{candidate_index}].capability_support must be a string-to-boolean object")
|
|
|
|
for profile_index, profile in enumerate(profiles):
|
|
if not isinstance(profile, dict):
|
|
raise ValueError(f"profiles[{profile_index}] must be an object")
|
|
profile_label = profile.get("label")
|
|
config_profile = profile.get("config_profile")
|
|
capabilities = profile.get("capabilities")
|
|
if not isinstance(profile_label, str) or not profile_label or "=" in profile_label:
|
|
raise ValueError(f"profiles[{profile_index}].label is invalid")
|
|
if not isinstance(config_profile, str) or not config_profile:
|
|
raise ValueError(f"profiles[{profile_index}].config_profile is invalid")
|
|
if not isinstance(capabilities, dict):
|
|
raise ValueError(f"profiles[{profile_index}].capabilities must be an object")
|
|
scoped_candidates = profile.get("candidate_labels")
|
|
if scoped_candidates is not None:
|
|
if (
|
|
not isinstance(scoped_candidates, list)
|
|
or not scoped_candidates
|
|
or not all(isinstance(item, str) and item for item in scoped_candidates)
|
|
):
|
|
raise ValueError(f"profiles[{profile_index}].candidate_labels must be a non-empty string array")
|
|
unknown_candidates = set(scoped_candidates) - candidate_labels
|
|
if unknown_candidates:
|
|
raise ValueError(
|
|
f"profiles[{profile_index}].candidate_labels contains unknown candidates: "
|
|
f"{', '.join(sorted(unknown_candidates))}"
|
|
)
|
|
if candidate_label not in scoped_candidates:
|
|
continue
|
|
overrides = _string_map(
|
|
profile.get("config_overrides"),
|
|
f"profiles[{profile_index}].config_overrides",
|
|
)
|
|
if config_profile == "default":
|
|
for key, claimed_value in capabilities.items():
|
|
disabled = claimed_value is False or (
|
|
isinstance(claimed_value, str) and claimed_value.strip().lower() == "false"
|
|
)
|
|
if disabled and overrides.get(key, "").strip().lower() != "false":
|
|
raise ValueError(
|
|
f"profiles[{profile_index}].capabilities claims {key}=false "
|
|
"but the default profile does not apply that setting; add the "
|
|
"same value to config_overrides"
|
|
)
|
|
if "incremental_exact_max_affected_paths" in overrides:
|
|
raise ValueError("exact cap belongs in scenarios[].exact_caps, not profile overrides")
|
|
profile_environment = _string_map(profile.get("environment"), f"profiles[{profile_index}].environment")
|
|
|
|
for scenario_index, scenario in enumerate(scenarios):
|
|
if not isinstance(scenario, dict):
|
|
raise ValueError(f"scenarios[{scenario_index}] must be an object")
|
|
scenario_name = scenario.get("name")
|
|
if not isinstance(scenario_name, str) or not scenario_name:
|
|
raise ValueError(f"scenarios[{scenario_index}].name is invalid")
|
|
if capability_quality is not None or workload == "self_dogfood":
|
|
frontier_values: list[int | None] = [None]
|
|
cap_values: list[int | None] = [None]
|
|
else:
|
|
frontier_values = _nonempty_list(
|
|
scenario.get("frontier_files"),
|
|
f"scenarios[{scenario_index}].frontier_files",
|
|
)
|
|
cap_values = _nonempty_list(
|
|
scenario.get("exact_caps"),
|
|
f"scenarios[{scenario_index}].exact_caps",
|
|
)
|
|
if not all(isinstance(item, int) and item > 0 for item in frontier_values):
|
|
raise ValueError("frontier_files must contain positive integers")
|
|
if not all(
|
|
item is None or (isinstance(item, int) and not isinstance(item, bool) and item > 0)
|
|
for item in cap_values
|
|
):
|
|
raise ValueError("exact_caps must contain positive integers or null")
|
|
|
|
for transport_index, transport in enumerate(transports):
|
|
for frontier_files in frontier_values:
|
|
for exact_cap in cap_values:
|
|
effective_capabilities = dict(capabilities)
|
|
effective_capabilities.update(overrides)
|
|
if capability_quality is not None:
|
|
command = [
|
|
str(benchmark_path),
|
|
"--binary",
|
|
str(binary),
|
|
"--capability-quality",
|
|
capability_quality,
|
|
"--transport",
|
|
transport,
|
|
"--config-profile",
|
|
config_profile,
|
|
"--index-mode",
|
|
index_mode,
|
|
]
|
|
if quality_background is not None:
|
|
command.extend(
|
|
(
|
|
"--quality-background-repo",
|
|
quality_background["repo"],
|
|
"--quality-background-revision",
|
|
quality_background["revision"],
|
|
)
|
|
)
|
|
elif workload == "self_dogfood":
|
|
assert repository_background is not None
|
|
command = [
|
|
str(benchmark_path),
|
|
"--binary",
|
|
str(binary),
|
|
"--self-dogfood",
|
|
"--repo-root",
|
|
repository_background["repo"],
|
|
"--repo-revision",
|
|
repository_background["revision"],
|
|
"--self-dogfood-scenarios",
|
|
scenario_name,
|
|
"--transport",
|
|
transport,
|
|
"--config-profile",
|
|
config_profile,
|
|
"--index-mode",
|
|
index_mode,
|
|
]
|
|
else:
|
|
command = [
|
|
str(benchmark_path),
|
|
"--binary",
|
|
str(binary),
|
|
"--matrix",
|
|
"--matrix-scenarios",
|
|
scenario_name,
|
|
"--frontier-files",
|
|
str(frontier_files),
|
|
"--transport",
|
|
transport,
|
|
"--config-profile",
|
|
config_profile,
|
|
"--index-mode",
|
|
index_mode,
|
|
]
|
|
cap_label = "default"
|
|
if isinstance(exact_cap, int):
|
|
effective_capabilities["incremental_exact_max_affected_paths"] = str(exact_cap)
|
|
command.extend(
|
|
(
|
|
"--config",
|
|
f"incremental_exact_max_affected_paths={exact_cap}",
|
|
)
|
|
)
|
|
cap_label = str(exact_cap)
|
|
for key, value in sorted(overrides.items()):
|
|
command.extend(("--config", f"{key}={value}"))
|
|
if capability_quality is not None or workload == "self_dogfood":
|
|
command.append("--include-logs")
|
|
command.extend(
|
|
(
|
|
"--timeout",
|
|
str(benchmark_timeout),
|
|
"--out",
|
|
"{result_path}",
|
|
)
|
|
)
|
|
parameters = {
|
|
"config_profile": config_profile,
|
|
"config_overrides": dict(sorted(overrides.items())),
|
|
"benchmark_script_sha256": benchmark_sha256,
|
|
"index_mode": index_mode,
|
|
}
|
|
if capability_quality is not None:
|
|
parameters["capability_quality"] = capability_quality
|
|
if quality_background is not None:
|
|
parameters["quality_background"] = quality_background
|
|
label = f"{candidate_label}.{profile_label}.{transport}.{scenario_name}"
|
|
elif workload == "self_dogfood":
|
|
assert repository_background is not None
|
|
parameters["repository_background"] = repository_background
|
|
label = f"{candidate_label}.{profile_label}.{transport}.{scenario_name}"
|
|
else:
|
|
parameters["frontier_files"] = frontier_files
|
|
parameters["exact_cap"] = exact_cap
|
|
label = (
|
|
f"{candidate_label}.{profile_label}.{transport}."
|
|
f"{scenario_name}.f{frontier_files}.cap{cap_label}"
|
|
)
|
|
environment = {
|
|
**common_environment,
|
|
**candidate_environment,
|
|
**profile_environment,
|
|
}
|
|
for repetition in range(1, repetitions + 1):
|
|
cell = {
|
|
"label": label,
|
|
"revision": revision,
|
|
"binary_sha256": binary_sha,
|
|
"build": build,
|
|
"capabilities": effective_capabilities,
|
|
"transport": transport,
|
|
"scenario": scenario_name,
|
|
"repetition": repetition,
|
|
"harness_version": harness_version,
|
|
"command": command,
|
|
"cwd": str(Path(cwd).expanduser().resolve()),
|
|
"parameters": parameters,
|
|
"timeout_seconds": cell_timeout,
|
|
"accepted_exit_codes": list(accepted_exit_codes),
|
|
}
|
|
if identity_version == 2:
|
|
cell["identity_version"] = 2
|
|
if environment:
|
|
cell["environment"] = environment
|
|
if isinstance(candidate_support, dict):
|
|
cell["capability_support"] = dict(sorted(candidate_support.items()))
|
|
cell["_design"] = {
|
|
"candidate_index": candidate_index,
|
|
"profile_index": profile_index,
|
|
"scenario_index": scenario_index,
|
|
"transport_index": transport_index,
|
|
"grouped_position": len(cells),
|
|
}
|
|
cells.append(cell)
|
|
if execution_order == "paired_interleaved":
|
|
cells.sort(
|
|
key=lambda cell: (
|
|
cell["repetition"],
|
|
cell["_design"]["scenario_index"],
|
|
cell["_design"]["transport_index"],
|
|
cell["_design"]["candidate_index"],
|
|
cell["_design"]["profile_index"],
|
|
cell["_design"]["grouped_position"],
|
|
)
|
|
)
|
|
for position, cell in enumerate(cells, start=1):
|
|
cell["parameters"] = {
|
|
**cell["parameters"],
|
|
"execution_order": execution_order,
|
|
"execution_block": cell["repetition"],
|
|
"execution_position": position,
|
|
}
|
|
for cell in cells:
|
|
cell.pop("_design", None)
|
|
plan = {"schema_version": SCHEMA_VERSION, "cells": cells}
|
|
campaign_definition = spec.get("campaign_version")
|
|
if campaign_definition is not None:
|
|
if campaign_definition != campaign_version():
|
|
raise ValueError(f"campaign_version must be {campaign_version()}")
|
|
plan["campaign_version"] = campaign_definition
|
|
runset = spec.get("runset_id")
|
|
if runset is not None:
|
|
plan["runset_id"] = _validate_runset_identity(runset)
|
|
if execution_order is not None:
|
|
plan["execution_order"] = execution_order
|
|
validate_plan(plan)
|
|
return plan
|
|
|
|
|
|
def ensure_disk_space(root: Path, minimum_free_bytes: int) -> None:
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
free = shutil.disk_usage(root).free
|
|
if free < minimum_free_bytes:
|
|
raise RuntimeError(f"insufficient experiment disk space: free={free} required={minimum_free_bytes} root={root}")
|
|
|
|
|
|
def resource_snapshot(path: Path) -> dict[str, Any]:
|
|
disk = shutil.disk_usage(path)
|
|
try:
|
|
load_average: list[float] | None = [round(value, 6) for value in os.getloadavg()]
|
|
except (AttributeError, OSError):
|
|
load_average = None
|
|
physical_memory_bytes: int | None = None
|
|
try:
|
|
pages = int(os.sysconf("SC_PHYS_PAGES"))
|
|
page_size = int(os.sysconf("SC_PAGE_SIZE"))
|
|
if pages > 0 and page_size > 0:
|
|
physical_memory_bytes = pages * page_size
|
|
except (AttributeError, OSError, TypeError, ValueError):
|
|
pass
|
|
return {
|
|
"captured_at_utc": utc_now(),
|
|
"hostname": socket.gethostname(),
|
|
"load_average": load_average,
|
|
"cpu_count": os.cpu_count(),
|
|
"physical_memory_bytes": physical_memory_bytes,
|
|
"disk": {
|
|
"path": str(path.resolve()),
|
|
"total_bytes": disk.total,
|
|
"used_bytes": disk.used,
|
|
"free_bytes": disk.free,
|
|
},
|
|
}
|
|
|
|
|
|
def validate_campaign_root(root: Path, *, allow_temporary: bool = False, temporary_root: Path | None = None) -> Path:
|
|
"""Require retained campaign state to live outside the OS temporary tree."""
|
|
resolved = root.expanduser().resolve()
|
|
temp = (temporary_root or Path(tempfile.gettempdir())).expanduser().resolve()
|
|
if not allow_temporary and (resolved == temp or temp in resolved.parents):
|
|
raise ValueError(
|
|
f"experiment root is temporary and may be lost after a crash or reboot: {resolved}; "
|
|
"choose a durable ignored path, or pass --allow-temporary-experiment-root only "
|
|
"for disposable tests"
|
|
)
|
|
return resolved
|
|
|
|
|
|
def process_is_live(pid: int) -> bool:
|
|
if pid <= 0:
|
|
return False
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
return True
|
|
return True
|
|
|
|
|
|
def acquire_lock(cell_root: Path, stale_after_seconds: int) -> tuple[Path, dict[str, Any] | None]:
|
|
cell_root.mkdir(parents=True, exist_ok=True)
|
|
lock_path = cell_root / "running.lock"
|
|
stale_record: dict[str, Any] | None = None
|
|
if lock_path.exists():
|
|
try:
|
|
existing = read_json_object(lock_path)
|
|
except (OSError, ValueError, json.JSONDecodeError):
|
|
existing = {"invalid": True}
|
|
try:
|
|
started_epoch = float(existing.get("started_epoch", 0.0))
|
|
pid = int(existing.get("pid", -1))
|
|
except (TypeError, ValueError):
|
|
started_epoch = 0.0
|
|
pid = -1
|
|
age = time.time() - started_epoch
|
|
same_host = existing.get("hostname") == socket.gethostname()
|
|
live = same_host and process_is_live(pid)
|
|
if live or age < stale_after_seconds:
|
|
raise RuntimeError(f"benchmark cell is already locked: {lock_path}")
|
|
stale_record = {"recovered_at_utc": utc_now(), "previous_lock": existing}
|
|
stale_path = cell_root / f"stale-lock-{filename_datetime()}-{uuid.uuid4().hex[:8]}.json"
|
|
atomic_write_json(stale_path, stale_record)
|
|
lock_path.unlink()
|
|
document = {
|
|
"pid": os.getpid(),
|
|
"hostname": socket.gethostname(),
|
|
"started_at_utc": utc_now(),
|
|
"started_epoch": time.time(),
|
|
}
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
|
descriptor = os.open(lock_path, flags, 0o600)
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
|
stream.write(json.dumps(document, indent=2, sort_keys=True) + "\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
return lock_path, stale_record
|
|
|
|
|
|
def resolve_result_path(cell_root: Path, completion: dict[str, Any]) -> Path:
|
|
relative = completion.get("result_path")
|
|
if not isinstance(relative, str):
|
|
raise ValueError("completion result_path is missing")
|
|
candidate = (cell_root / relative).resolve()
|
|
if cell_root.resolve() not in candidate.parents:
|
|
raise ValueError("completion result_path escapes the cell directory")
|
|
return candidate
|
|
|
|
|
|
def validate_result(path: Path, cell: dict[str, Any]) -> dict[str, Any]:
|
|
result = read_json_object(path)
|
|
metadata = result.get("binary_metadata")
|
|
actual_sha = metadata.get("sha256") if isinstance(metadata, dict) else None
|
|
if actual_sha != cell["binary_sha256"]:
|
|
raise ValueError(f"result binary SHA-256 mismatch: expected={cell['binary_sha256']} actual={actual_sha}")
|
|
if result.get("error"):
|
|
raise ValueError(f"benchmark result contains an error: {result['error']}")
|
|
derived = result.get("derived")
|
|
if not isinstance(derived, dict) or not isinstance(derived.get("passed"), bool):
|
|
raise ValueError("benchmark result must contain derived.passed as a boolean")
|
|
cases = result.get("cases")
|
|
measurements = result.get("measurements")
|
|
if not (isinstance(cases, list) and cases) and not isinstance(measurements, dict):
|
|
raise ValueError("benchmark result must contain non-empty cases or measurements")
|
|
expected_background = cell.get("parameters", {}).get("quality_background")
|
|
if expected_background is not None:
|
|
first_case = cases[0] if isinstance(cases, list) and cases else None
|
|
actual_background = first_case.get("background_repository") if isinstance(first_case, dict) else None
|
|
if not isinstance(actual_background, dict):
|
|
raise ValueError("benchmark result is missing background_repository identity")
|
|
for key in ("revision", "tree"):
|
|
if actual_background.get(key) != expected_background.get(key):
|
|
raise ValueError(
|
|
f"background repository {key} mismatch: "
|
|
f"expected={expected_background.get(key)} actual={actual_background.get(key)}"
|
|
)
|
|
expected_repository = cell.get("parameters", {}).get("repository_background")
|
|
if expected_repository is not None:
|
|
actual_repository = result.get("repository_background")
|
|
if not isinstance(actual_repository, dict):
|
|
raise ValueError("benchmark result is missing repository_background identity")
|
|
for key in ("revision", "tree"):
|
|
if actual_repository.get(key) != expected_repository.get(key):
|
|
raise ValueError(
|
|
f"repository background {key} mismatch: "
|
|
f"expected={expected_repository.get(key)} "
|
|
f"actual={actual_repository.get(key)}"
|
|
)
|
|
return result
|
|
|
|
|
|
def validate_attempt_artifacts(cell_root: Path, completion: dict[str, Any]) -> None:
|
|
"""Re-hash a completed attempt's archived evidence before trusting its audit status."""
|
|
attempt_id = completion.get("attempt")
|
|
if attempt_id is None:
|
|
# Historical hand-authored plans may predate per-attempt evidence. Their
|
|
# result hash remains validated, but there is no artifact claim to check.
|
|
return
|
|
if (
|
|
not isinstance(attempt_id, str)
|
|
or not attempt_id
|
|
or Path(attempt_id).name != attempt_id
|
|
or attempt_id in {".", ".."}
|
|
):
|
|
raise ValueError("completion attempt identifier is invalid")
|
|
attempt_root = cell_root / "attempts" / attempt_id
|
|
attempt = read_json_object(attempt_root / "attempt.json")
|
|
if attempt.get("cell_identity") != completion.get("cell_identity"):
|
|
raise ValueError("attempt cell identity does not match the completion")
|
|
if attempt.get("status") != "completed":
|
|
raise ValueError("completed cell references a non-completed attempt")
|
|
expected = attempt.get("artifacts")
|
|
if not isinstance(expected, dict):
|
|
raise ValueError("completed attempt artifact manifest is missing")
|
|
actual = artifact_manifest(attempt_root / "artifacts")
|
|
if actual != expected:
|
|
raise ValueError("completed attempt artifact manifest does not match retained files")
|
|
|
|
|
|
def valid_completion(cell_root: Path, cell: dict[str, Any]) -> dict[str, Any] | None:
|
|
completion_path = cell_root / "complete.json"
|
|
if not completion_path.is_file():
|
|
return None
|
|
completion = read_json_object(completion_path)
|
|
if completion.get("cell_identity") != cell_identity(cell):
|
|
raise ValueError("completion cell identity does not match the plan")
|
|
result_path = resolve_result_path(cell_root, completion)
|
|
validate_result(result_path, cell)
|
|
if file_sha256(result_path) != completion.get("result_sha256"):
|
|
raise ValueError("completion result SHA-256 does not match the retained result")
|
|
validate_attempt_artifacts(cell_root, completion)
|
|
return completion
|
|
|
|
|
|
def expanded_command(command: list[str], attempt_root: Path, result_path: Path) -> list[str]:
|
|
replacements = {
|
|
"{attempt_dir}": str(attempt_root),
|
|
"{result_path}": str(result_path),
|
|
}
|
|
return [replacements.get(item, item) for item in command]
|
|
|
|
|
|
def cell_process_group_options() -> dict[str, Any]:
|
|
if os.name == "nt":
|
|
return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
|
|
return {"start_new_session": True}
|
|
|
|
|
|
def stop_cell_process_tree(
|
|
process: subprocess.Popen[bytes], initial_signal: int, grace_seconds: float = 30.0
|
|
) -> int | None:
|
|
"""Stop an isolated benchmark process group, allowing harness cleanup first."""
|
|
if process.poll() is not None:
|
|
return process.returncode
|
|
try:
|
|
if os.name == "nt":
|
|
process.send_signal(signal.CTRL_BREAK_EVENT)
|
|
else:
|
|
os.killpg(process.pid, initial_signal)
|
|
except (OSError, ProcessLookupError):
|
|
pass
|
|
try:
|
|
return process.wait(timeout=grace_seconds)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
if os.name == "nt":
|
|
subprocess.run(
|
|
["taskkill", "/PID", str(process.pid), "/T", "/F"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
else:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except (OSError, ProcessLookupError):
|
|
pass
|
|
try:
|
|
return process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
return process.poll()
|
|
|
|
|
|
def run_cell(
|
|
campaign_root: Path,
|
|
cell: dict[str, Any],
|
|
*,
|
|
minimum_free_bytes: int = DEFAULT_MINIMUM_FREE_BYTES,
|
|
stale_lock_seconds: int = DEFAULT_STALE_LOCK_SECONDS,
|
|
) -> dict[str, Any]:
|
|
validate_cell(cell, 0)
|
|
ensure_disk_space(campaign_root, minimum_free_bytes)
|
|
identity = cell_identity(cell)
|
|
cell_root = campaign_root / "runs" / identity
|
|
try:
|
|
completion = valid_completion(cell_root, cell)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
return {
|
|
"cell_identity": identity,
|
|
"label": cell["label"],
|
|
"status": "corrupt",
|
|
"error": str(exc),
|
|
}
|
|
if completion is not None:
|
|
return {"cell_identity": identity, "label": cell["label"], "status": "resumed"}
|
|
|
|
lock_path, stale_record = acquire_lock(cell_root, stale_lock_seconds)
|
|
try:
|
|
attempt_id = filename_datetime() + f"-{uuid.uuid4().hex[:8]}"
|
|
attempt_root = cell_root / "attempts" / attempt_id
|
|
attempt_root.mkdir(parents=True)
|
|
artifact_root = attempt_root / "artifacts"
|
|
result_path = attempt_root / "result.json"
|
|
command = expanded_command(cell["command"], attempt_root, result_path)
|
|
cwd = Path(cell.get("cwd") or Path.cwd()).expanduser().resolve()
|
|
environment = dict(os.environ)
|
|
overrides = cell.get("environment", {})
|
|
if not isinstance(overrides, dict) or not all(
|
|
isinstance(key, str) and isinstance(value, str) for key, value in overrides.items()
|
|
):
|
|
raise ValueError("cell environment must be a string-to-string object")
|
|
environment.update(overrides)
|
|
environment["CBM_BENCHMARK_ARTIFACT_DIR"] = str(artifact_root)
|
|
command_record = {
|
|
"cell_identity": identity,
|
|
"identity": identity_document(cell),
|
|
"label": cell["label"],
|
|
"command": command,
|
|
"cwd": str(cwd),
|
|
"environment_overrides": overrides,
|
|
"artifact_directory": "artifacts",
|
|
"started_at_utc": utc_now(),
|
|
"stale_lock_recovered": stale_record is not None,
|
|
"resource_before": resource_snapshot(campaign_root),
|
|
}
|
|
atomic_write_json(attempt_root / "command.json", command_record)
|
|
started = time.monotonic()
|
|
returncode: int | None = None
|
|
error: str | None = None
|
|
interrupted = False
|
|
except Exception:
|
|
if lock_path.exists():
|
|
lock_path.unlink()
|
|
raise
|
|
try:
|
|
with (
|
|
(attempt_root / "stdout.log").open("wb") as stdout,
|
|
(attempt_root / "stderr.log").open("wb") as stderr,
|
|
):
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
cwd=cwd,
|
|
env=environment,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
**cell_process_group_options(),
|
|
)
|
|
returncode = process.wait(timeout=cell.get("timeout_seconds"))
|
|
except subprocess.TimeoutExpired as exc:
|
|
error = f"command timed out after {exc.timeout} seconds"
|
|
returncode = stop_cell_process_tree(process, signal.SIGTERM)
|
|
except KeyboardInterrupt:
|
|
error = "command interrupted by SIGINT"
|
|
interrupted = True
|
|
returncode = stop_cell_process_tree(process, signal.SIGINT)
|
|
accepted_codes = cell.get("accepted_exit_codes", [0])
|
|
if error is None and returncode not in accepted_codes:
|
|
error = f"command exited with {returncode}; accepted={accepted_codes}"
|
|
result: dict[str, Any] | None = None
|
|
if error is None:
|
|
try:
|
|
result = validate_result(result_path, cell)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
error = str(exc)
|
|
attempt_record = {
|
|
**command_record,
|
|
"finished_at_utc": utc_now(),
|
|
"elapsed_seconds": round(time.monotonic() - started, 6),
|
|
"returncode": returncode,
|
|
"status": "completed" if error is None else "failed",
|
|
"error": error,
|
|
"resource_after": resource_snapshot(campaign_root),
|
|
"artifacts": artifact_manifest(artifact_root),
|
|
}
|
|
atomic_write_json(attempt_root / "attempt.json", attempt_record)
|
|
if interrupted:
|
|
raise KeyboardInterrupt
|
|
if error is not None:
|
|
return {
|
|
"cell_identity": identity,
|
|
"label": cell["label"],
|
|
"status": "failed",
|
|
"error": error,
|
|
"attempt": attempt_id,
|
|
}
|
|
assert result is not None
|
|
derived = result.get("derived")
|
|
benchmark_passed = derived.get("passed") if isinstance(derived, dict) else None
|
|
completion = {
|
|
"cell_identity": identity,
|
|
"label": cell["label"],
|
|
"completed_at_utc": utc_now(),
|
|
"attempt": attempt_id,
|
|
"result_path": str(result_path.relative_to(cell_root)),
|
|
"result_sha256": file_sha256(result_path),
|
|
"returncode": returncode,
|
|
"benchmark_passed": benchmark_passed,
|
|
}
|
|
atomic_write_json(cell_root / "complete.json", completion)
|
|
return {
|
|
"cell_identity": identity,
|
|
"label": cell["label"],
|
|
"status": "completed",
|
|
}
|
|
finally:
|
|
if lock_path.exists():
|
|
lock_path.unlink()
|
|
|
|
|
|
def scan_campaign(campaign_root: Path, cells: list[dict[str, Any]]) -> dict[str, Any]:
|
|
expected = {cell_identity(cell): cell for cell in cells}
|
|
entries: list[dict[str, Any]] = []
|
|
counts = {
|
|
"complete": 0,
|
|
"missing": 0,
|
|
"corrupt": 0,
|
|
"duplicate_attempts": 0,
|
|
"unplanned": 0,
|
|
}
|
|
for identity, cell in expected.items():
|
|
cell_root = campaign_root / "runs" / identity
|
|
attempts_root = cell_root / "attempts"
|
|
attempt_count = sum(1 for path in attempts_root.iterdir() if path.is_dir()) if attempts_root.is_dir() else 0
|
|
if attempt_count > 1:
|
|
counts["duplicate_attempts"] += attempt_count - 1
|
|
status = "missing"
|
|
error = None
|
|
try:
|
|
if valid_completion(cell_root, cell) is not None:
|
|
status = "complete"
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
status = "corrupt"
|
|
error = str(exc)
|
|
counts[status] += 1
|
|
entries.append(
|
|
{
|
|
"cell_identity": identity,
|
|
"label": cell["label"],
|
|
"status": status,
|
|
"attempts": attempt_count,
|
|
"error": error,
|
|
}
|
|
)
|
|
runs_root = campaign_root / "runs"
|
|
actual = {path.name for path in runs_root.iterdir() if path.is_dir()} if runs_root.is_dir() else set()
|
|
unplanned = sorted(actual - set(expected))
|
|
counts["unplanned"] = len(unplanned)
|
|
return {"counts": counts, "cells": entries, "unplanned": unplanned}
|
|
|
|
|
|
def environment_snapshot(plan_path: Path) -> dict[str, Any]:
|
|
return {
|
|
"captured_at_utc": utc_now(),
|
|
"plan_path": str(plan_path.resolve()),
|
|
"plan_sha256": file_sha256(plan_path),
|
|
"hostname": socket.gethostname(),
|
|
"platform": platform.platform(),
|
|
"python": sys.version,
|
|
"cpu_count": os.cpu_count(),
|
|
"resources": resource_snapshot(plan_path.parent),
|
|
}
|
|
|
|
|
|
def completed_report_inputs(campaign_root: Path, cells: list[dict[str, Any]]) -> list[tuple[str, Path]]:
|
|
inputs: list[tuple[str, Path]] = []
|
|
for cell in cells:
|
|
cell_root = campaign_root / "runs" / cell_identity(cell)
|
|
completion = valid_completion(cell_root, cell)
|
|
if completion is not None:
|
|
result_path = resolve_result_path(cell_root, completion)
|
|
inputs.append(
|
|
(
|
|
cell["label"],
|
|
materialize_report_input(campaign_root, cell, result_path),
|
|
)
|
|
)
|
|
return inputs
|
|
|
|
|
|
def materialize_report_input(campaign_root: Path, cell: dict[str, Any], result_path: Path) -> Path:
|
|
"""Create a deterministic derived input with candidate metadata beside immutable raw results."""
|
|
document = read_json_object(result_path)
|
|
parameters = document.get("parameters")
|
|
if not isinstance(parameters, dict):
|
|
parameters = {}
|
|
document["parameters"] = parameters
|
|
support = cell.get("capability_support")
|
|
if isinstance(support, dict):
|
|
parameters["capability_support"] = dict(sorted(support.items()))
|
|
cell_parameters = cell.get("parameters")
|
|
if isinstance(cell_parameters, dict):
|
|
for key in ("execution_order", "execution_block", "execution_position"):
|
|
if key in cell_parameters:
|
|
parameters[key] = cell_parameters[key]
|
|
source_sha = file_sha256(result_path)
|
|
identity = cell_identity(cell)
|
|
document["campaign_provenance"] = {
|
|
"cell_identity": identity,
|
|
"source_result": str(result_path),
|
|
"source_result_sha256": source_sha,
|
|
}
|
|
output = campaign_root / "reports" / "inputs" / f"{identity}-{source_sha[:12]}.json"
|
|
atomic_write_json(output, document)
|
|
return output
|
|
|
|
|
|
def generate_report(campaign_root: Path, cells: list[dict[str, Any]], output: Path) -> dict[str, Any]:
|
|
inputs = completed_report_inputs(campaign_root, cells)
|
|
if not inputs:
|
|
raise RuntimeError("cannot generate a report without completed campaign cells")
|
|
summarizer = Path(__file__).resolve().with_name("summarize-benchmark-results.py")
|
|
command = [sys.executable, str(summarizer)]
|
|
for label, result_path in inputs:
|
|
command.extend(("--input", f"{label}={result_path}"))
|
|
command.extend(("--out", str(output)))
|
|
process = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
if process.returncode != 0:
|
|
raise RuntimeError(f"report generator exited with {process.returncode}: {process.stderr.strip()}")
|
|
return {
|
|
"path": str(output),
|
|
"sha256": file_sha256(output),
|
|
"input_count": len(inputs),
|
|
"generator": str(summarizer),
|
|
}
|
|
|
|
|
|
def write_manifest(
|
|
campaign_root: Path,
|
|
plan_path: Path,
|
|
cells: list[dict[str, Any]],
|
|
report: dict[str, Any] | None = None,
|
|
*,
|
|
runset: str | None = None,
|
|
) -> Path:
|
|
manifest = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"generated_at_utc": utc_now(),
|
|
"plan_sha256": file_sha256(plan_path),
|
|
"audit": scan_campaign(campaign_root, cells),
|
|
"generated_report": report,
|
|
}
|
|
effective_runset = runset or file_sha256(plan_path)[:12]
|
|
name = generated_artifact_name(
|
|
"manifest",
|
|
effective_runset,
|
|
".json",
|
|
nonce=uuid.uuid4().hex[:8],
|
|
)
|
|
path = campaign_root / "manifests" / name
|
|
atomic_write_json(path, manifest)
|
|
return path
|
|
|
|
|
|
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
source = parser.add_mutually_exclusive_group()
|
|
source.add_argument("--plan", type=Path, help="Fully expanded immutable experiment plan.")
|
|
source.add_argument(
|
|
"--matrix-spec",
|
|
type=Path,
|
|
help="Compact deterministic grid expanded and archived before execution.",
|
|
)
|
|
source.add_argument(
|
|
"--quick",
|
|
dest="preset",
|
|
action="store_const",
|
|
const="quick",
|
|
help="Automatically prepare and run the safe one-repetition smoke (default).",
|
|
)
|
|
source.add_argument(
|
|
"--full",
|
|
dest="preset",
|
|
action="store_const",
|
|
const="full",
|
|
help="Automatically prepare and run the repeated capability matrix.",
|
|
)
|
|
parser.add_argument(
|
|
"--experiment-root",
|
|
"--campaign-root",
|
|
dest="campaign_root",
|
|
type=Path,
|
|
help=(
|
|
"Durable result root (--campaign-root is a backwards-compatible alias). "
|
|
"Automatic modes default to a versioned, commit-qualified, "
|
|
"content-addressed runset directory under "
|
|
".worktrees/benchmark-campaign (path kept for backwards compatibility "
|
|
"with existing retained runsets)."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--candidate-root",
|
|
type=Path,
|
|
help=("Automatic candidate worktree/build root (default: .worktrees/benchmark-candidates)."),
|
|
)
|
|
parser.add_argument("--build-jobs", type=int, default=2)
|
|
parser.add_argument(
|
|
"--allow-temporary-experiment-root",
|
|
"--allow-temporary-campaign-root",
|
|
dest="allow_temporary_campaign_root",
|
|
action="store_true",
|
|
help="Allow disposable experiment state under the OS temporary directory.",
|
|
)
|
|
parser.add_argument(
|
|
"--candidate-ref",
|
|
dest="candidate_ref_overrides",
|
|
action="append",
|
|
default=[],
|
|
metavar="LABEL=REF",
|
|
help=(
|
|
"Override one automatic candidate's git ref by label (repeatable), e.g. "
|
|
"--candidate-ref upstream-main=origin/main. Valid labels: "
|
|
+ ", ".join(label for label, _ in DEFAULT_CANDIDATE_REFS)
|
|
+ ". Only applies to --quick/--full; explicit --plan/--matrix-spec already "
|
|
"accept any resolvable ref directly in the spec. An override is an explicit "
|
|
"request and stays fail-closed: an unresolvable override ref raises rather "
|
|
"than falling back."
|
|
),
|
|
)
|
|
parser.add_argument("--minimum-free-gb", type=float, default=2.0)
|
|
parser.add_argument("--stale-lock-hours", type=float, default=6.0)
|
|
parser.add_argument("--audit-only", action="store_true")
|
|
parser.add_argument(
|
|
"--report-out",
|
|
type=Path,
|
|
help="Generated Markdown path (default: versioned runset report under EXPERIMENT_ROOT/reports).",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
if args.plan is None and args.matrix_spec is None and args.preset is None:
|
|
args.preset = "quick"
|
|
if args.preset is None and args.campaign_root is None:
|
|
parser.error("--experiment-root (or --campaign-root) is required with --plan or --matrix-spec")
|
|
if args.build_jobs <= 0:
|
|
parser.error("--build-jobs must be positive")
|
|
candidate_ref_overrides: dict[str, str] = {}
|
|
for value in args.candidate_ref_overrides:
|
|
try:
|
|
label, ref = parse_candidate_ref_override(value)
|
|
except ValueError as error:
|
|
parser.error(str(error))
|
|
candidate_ref_overrides[label] = ref
|
|
if candidate_ref_overrides and args.preset is None:
|
|
parser.error("--candidate-ref only applies to --quick/--full")
|
|
args.candidate_ref = candidate_ref_overrides
|
|
return args
|
|
|
|
|
|
def _commit_datetime_slug(repository: Path, revision: str) -> str:
|
|
return commit_identity(repository, revision)["commit_datetime_slug"]
|
|
|
|
|
|
def prepare_automatic_campaign(
|
|
args: argparse.Namespace,
|
|
) -> tuple[Path, Path]:
|
|
repository = Path(__file__).resolve().parents[1]
|
|
ensure_clean_tracked_worktree(repository, "benchmark source worktree")
|
|
candidate_root = (
|
|
args.candidate_root.expanduser().resolve()
|
|
if args.candidate_root
|
|
else repository / ".worktrees" / "benchmark-candidates"
|
|
)
|
|
ensure_disk_space(candidate_root, max(0, int(args.minimum_free_gb * 1024**3)))
|
|
candidate_ref_overrides: dict[str, str] = getattr(args, "candidate_ref", {}) or {}
|
|
effective_candidate_refs = [
|
|
(label, candidate_ref_overrides[label])
|
|
if label in candidate_ref_overrides
|
|
else (label, resolve_default_candidate_ref(repository, label, ref))
|
|
for label, ref in DEFAULT_CANDIDATE_REFS
|
|
]
|
|
candidates = [
|
|
materialize_candidate(
|
|
repository,
|
|
candidate_root,
|
|
label,
|
|
ref,
|
|
jobs=args.build_jobs,
|
|
)
|
|
for label, ref in effective_candidate_refs
|
|
]
|
|
benchmark_script = repository / "scripts" / "benchmark-incremental-speed.py"
|
|
spec = build_automatic_spec(
|
|
repository,
|
|
benchmark_script,
|
|
candidates,
|
|
preset=args.preset,
|
|
)
|
|
revision = spec["repository_background"]["revision"]
|
|
tree = spec["repository_background"]["tree"]
|
|
commit_datetime = _commit_datetime_slug(repository, revision)
|
|
runset = automatic_runset_identity(spec)
|
|
spec["runset_id"] = runset
|
|
spec_payload = (json.dumps(spec, indent=2, sort_keys=True) + "\n").encode("utf-8")
|
|
source_identity = {
|
|
"revision": revision,
|
|
"commit_datetime_slug": commit_datetime,
|
|
"tree": tree,
|
|
}
|
|
campaign_root = (
|
|
args.campaign_root.expanduser().resolve()
|
|
if args.campaign_root
|
|
else repository
|
|
/ ".worktrees"
|
|
/ "benchmark-campaign"
|
|
/ automatic_campaign_name(args.preset, source_identity, runset)
|
|
)
|
|
campaign_root = validate_campaign_root(
|
|
campaign_root,
|
|
allow_temporary=args.allow_temporary_campaign_root,
|
|
)
|
|
spec_path = campaign_root / "inputs" / automatic_spec_name(args.preset, runset)
|
|
if spec_path.exists():
|
|
if spec_path.read_bytes() != spec_payload:
|
|
raise RuntimeError(f"automatic spec path contains different bytes: {spec_path}")
|
|
else:
|
|
atomic_write_bytes(spec_path, spec_payload)
|
|
return campaign_root, spec_path
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_arguments(argv)
|
|
|
|
if args.preset is not None:
|
|
campaign_root, matrix_spec = prepare_automatic_campaign(args)
|
|
args.matrix_spec = matrix_spec
|
|
else:
|
|
assert args.campaign_root is not None
|
|
campaign_root = validate_campaign_root(
|
|
args.campaign_root,
|
|
allow_temporary=args.allow_temporary_campaign_root,
|
|
)
|
|
|
|
minimum_free_bytes = max(0, int(args.minimum_free_gb * 1024**3))
|
|
stale_lock_seconds = max(1, int(args.stale_lock_hours * 3600))
|
|
ensure_disk_space(campaign_root, minimum_free_bytes)
|
|
if args.matrix_spec:
|
|
spec_path = args.matrix_spec.expanduser().resolve()
|
|
spec = read_json_object(spec_path)
|
|
plan = expand_matrix_spec(spec)
|
|
plan["matrix_spec_sha256"] = file_sha256(spec_path)
|
|
archived_spec = campaign_root / "specs" / f"{file_sha256(spec_path)}.json"
|
|
if not archived_spec.exists():
|
|
atomic_write_bytes(archived_spec, spec_path.read_bytes())
|
|
plan_payload = (json.dumps(plan, indent=2, sort_keys=True) + "\n").encode("utf-8")
|
|
plan_digest = hashlib.sha256(plan_payload).hexdigest()
|
|
plan_path = campaign_root / "plans" / f"{plan_digest}.json"
|
|
if not plan_path.exists():
|
|
atomic_write_bytes(plan_path, plan_payload)
|
|
else:
|
|
plan_path = args.plan.expanduser().resolve()
|
|
plan = read_json_object(plan_path)
|
|
archived_plan = campaign_root / "plans" / f"{file_sha256(plan_path)}.json"
|
|
if not archived_plan.exists():
|
|
atomic_write_bytes(archived_plan, plan_path.read_bytes())
|
|
plan_path = archived_plan
|
|
cells = validate_plan(plan)
|
|
runset = plan.get("runset_id", file_sha256(plan_path)[:12])
|
|
runset = _validate_runset_identity(runset)
|
|
snapshot_name = generated_artifact_name("environment", runset, ".json")
|
|
atomic_write_json(
|
|
campaign_root / "environments" / snapshot_name,
|
|
environment_snapshot(plan_path),
|
|
)
|
|
|
|
failures = 0
|
|
if not args.audit_only:
|
|
for cell in cells:
|
|
outcome = run_cell(
|
|
campaign_root,
|
|
cell,
|
|
minimum_free_bytes=minimum_free_bytes,
|
|
stale_lock_seconds=stale_lock_seconds,
|
|
)
|
|
print(json.dumps(outcome, sort_keys=True), flush=True)
|
|
failures += int(outcome["status"] in {"failed", "corrupt"})
|
|
audit = scan_campaign(campaign_root, cells)
|
|
report_metadata = None
|
|
if audit["counts"]["complete"]:
|
|
report_path = (
|
|
args.report_out.expanduser().resolve()
|
|
if args.report_out
|
|
else campaign_root
|
|
/ "reports"
|
|
/ generated_artifact_name(
|
|
"report",
|
|
runset,
|
|
".md",
|
|
preset=args.preset or "custom",
|
|
)
|
|
)
|
|
report_metadata = generate_report(campaign_root, cells, report_path)
|
|
manifest_path = write_manifest(
|
|
campaign_root,
|
|
plan_path,
|
|
cells,
|
|
report_metadata,
|
|
runset=runset,
|
|
)
|
|
print(json.dumps({"manifest": str(manifest_path), "audit": audit}, indent=2, sort_keys=True))
|
|
return 1 if failures or audit["counts"]["missing"] or audit["counts"]["corrupt"] else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|