From ef7035197ef9a466176547fbe92105cf265a1a62 Mon Sep 17 00:00:00 2001 From: Arjun Sharma Date: Mon, 19 Oct 2026 03:46:39 +0000 Subject: [PATCH 01/32] verify: fc-verify with status, classify, report, review and automatic not_approved rules --- pyproject.toml | 1 + src/datasmith/verify/__init__.py | 6 + src/datasmith/verify/cli.py | 89 ++++++++++++++ src/datasmith/verify/noise.py | 198 +++++++++++++++++++++++++++++++ src/datasmith/verify/report.py | 162 +++++++++++++++++++++++++ src/datasmith/verify/review.py | 100 ++++++++++++++++ src/datasmith/verify/status.py | 23 ++++ src/datasmith/verify/survey.py | 92 ++++++++++++++ tests/verify/__init__.py | 0 tests/verify/conftest.py | 50 ++++++++ tests/verify/test_verify.py | 97 +++++++++++++++ 11 files changed, 818 insertions(+) create mode 100644 src/datasmith/verify/__init__.py create mode 100644 src/datasmith/verify/cli.py create mode 100644 src/datasmith/verify/noise.py create mode 100644 src/datasmith/verify/report.py create mode 100644 src/datasmith/verify/review.py create mode 100644 src/datasmith/verify/status.py create mode 100644 src/datasmith/verify/survey.py create mode 100644 tests/verify/__init__.py create mode 100644 tests/verify/conftest.py create mode 100644 tests/verify/test_verify.py diff --git a/pyproject.toml b/pyproject.toml index 26803406..53b1b72e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ [project.scripts] fc-data = "datasmith.update.cli:main" fc-render-tasks = "datasmith.harbor_adapter.regen:main" +fc-verify = "datasmith.verify.cli:main" [project.urls] Homepage = "https://formula-code.github.io/datasmith/" diff --git a/src/datasmith/verify/__init__.py b/src/datasmith/verify/__init__.py new file mode 100644 index 00000000..1422de74 --- /dev/null +++ b/src/datasmith/verify/__init__.py @@ -0,0 +1,6 @@ +"""Task approval and repair: the per-task stages behind the ``fc-verify`` command. + +A survey folder holds one state file per task (``state/.json``), the oracle records (``oracle/.json``), +the pinned task folders (``pinned//``), the noise rule (``threshold_rule.json``) and the review file +(``manual_review.csv``). +""" diff --git a/src/datasmith/verify/cli.py b/src/datasmith/verify/cli.py new file mode 100644 index 00000000..429b045a --- /dev/null +++ b/src/datasmith/verify/cli.py @@ -0,0 +1,89 @@ +"""fc-verify: per-task approval and repair stages on a survey folder. + +fc-verify status SURVEY +fc-verify classify SURVEY [TASK ...] oracle up/down per task on the noise rule +fc-verify report SURVEY [TASK ...] review summaries (default: measured tasks pending review) +fc-verify review SURVEY TASK DECISION [--note] +fc-verify auto SURVEY [--dry-run] automatic not_approved rules +fc-verify add SURVEY TASK ... pending rows for new tasks +fc-verify failure LOG ... error class of failed image builds +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from datasmith.verify import report, review, status +from datasmith.verify.noise import classify_all, load_rule +from datasmith.verify.survey import DECISIONS, Survey, task_id + + +def _tasks(ids: list[str]) -> list[str]: + return [task_id(t) if "#" in t else t for t in ids] + + +def _status(survey: Survey, a: argparse.Namespace) -> None: + print(json.dumps(status.counts(survey), indent=1)) + + +def _classify(survey: Survey, a: argparse.Namespace) -> None: + stats = classify_all(survey.records(), load_rule(survey.rule_path)) + for t in _tasks(a.tasks) or sorted(stats): + if t in stats: + print(json.dumps({k: v for k, v in stats[t].items() if k not in ("up", "down")})) + + +def _report(survey: Survey, a: argparse.Namespace) -> None: + for s in report.summaries(survey, load_rule(survey.rule_path), _tasks(a.tasks) or None): + print(report.render(s), end="\n\n") + + +def _review(survey: Survey, a: argparse.Namespace) -> None: + print(",".join(review.set_decision(survey, a.task, a.decision, a.note, a.data).values())) + + +def _auto(survey: Survey, a: argparse.Namespace) -> None: + picked = review.apply_auto(survey, dry_run=a.dry_run) + for d in picked: + print(f"{d['task']}: {d['detail']}") + print(f"{'would mark' if a.dry_run else 'marked'} {len(picked)} not_approved") + + +def _add(survey: Survey, a: argparse.Namespace) -> None: + print("\n".join(review.add_pending(survey, _tasks(a.tasks)))) + + +COMMANDS = {"status": _status, "classify": _classify, "report": _report, "review": _review, "auto": _auto, "add": _add} + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser( + prog="fc-verify", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + sub = ap.add_subparsers(dest="cmd", required=True) + for name in ("status", "classify", "report", "review", "auto", "add"): + p = sub.add_parser(name) + p.add_argument("survey", type=Path) + if name in ("classify", "report", "add"): + p.add_argument("tasks", nargs="*") + sub.choices["review"].add_argument("task") + sub.choices["review"].add_argument("decision", choices=DECISIONS) + sub.choices["review"].add_argument("--note", default="") + sub.choices["review"].add_argument("--data", default="phase 1") + sub.choices["auto"].add_argument("--dry-run", action="store_true") + sub.add_parser("failure").add_argument("logs", nargs="+", type=Path) + a = ap.parse_args(argv) + + if a.cmd == "failure": + for log in a.logs: + print(json.dumps({"log": str(log), **report.failure_reason(log)})) + return 0 + COMMANDS[a.cmd](Survey(a.survey), a) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/datasmith/verify/noise.py b/src/datasmith/verify/noise.py new file mode 100644 index 00000000..c4db4318 --- /dev/null +++ b/src/datasmith/verify/noise.py @@ -0,0 +1,198 @@ +"""Noise rule and the per-benchmark oracle classification built on it. + +All values are natural-log speedups. The no-change spread is s = 1.4826 * median |ln s|, centered at 0: a no-change run +that lands off 1.00x counts whether it is biased or scattered. Per benchmark, sigma_b^2 = (n_b·s_b^2 + n0·s_repo^2)/(n_b + n0) +and theta_b = clip(k·sigma_b, theta_min, theta_max). A benchmark is up (down) when at least 2 oracle runs pass +theta_b (-theta_b). +""" + +from __future__ import annotations + +import json +import math +import statistics +from pathlib import Path +from typing import Any + +from datasmith.verify.survey import repo_of + +NEEDS = {"global": ("theta",), "per_benchmark": ("k", "n0", "theta_min", "theta_max")} +# An oracle run that fails more than this share of the tests that pass without the patch cannot be the reference. +ORACLE_MAX_BROKEN = 0.5 +ORACLE_MIN_RUNS = 2 + + +def load_rule(path: Path) -> dict[str, Any]: + rule = json.loads(Path(path).read_text()) + for repo in [None, *rule.get("repos", {})]: + r = rule_for(rule, repo) + missing = [k for k in NEEDS.get(r.get("form"), ("form",)) if k not in r] + if missing or r.get("bias_check", "aggregate") not in {"aggregate", "per_benchmark"}: + raise ValueError(f"{path}: bad rule for repo {repo}: missing {missing} or bad bias_check") + return rule + + +def rule_for(rule: dict, repo: str | None) -> dict: + return {**rule, **rule.get("repos", {}).get(repo, {})} + + +def robust_sd(xs: list[float], center: float | None = None) -> float | None: + """1.4826 * median |x - center|; with center None the sample median is used and at least 2 values are needed.""" + if not xs or (center is None and len(xs) < 2): + return None + m = statistics.median(xs) if center is None else center + return 1.4826 * statistics.median(abs(x - m) for x in xs) + + +def nop_logs(record: dict) -> dict[str, list[float]]: + out: dict[str, list[float]] = {} + for run in record["runs"]: + if run["role"] == "nop": + for b, v in run["log_speedups"].items(): + out.setdefault(b, []).append(v) + return out + + +def repo_spreads(records: list[dict]) -> dict[str, float]: + """Per repo: the no-change spread around 0 over all its no-change log speedups.""" + values: dict[str, list[float]] = {} + for record in records: + for xs in nop_logs(record).values(): + values.setdefault(repo_of(record["task"]), []).extend(xs) + return {repo: s for repo, xs in values.items() if (s := robust_sd(xs, center=0.0)) is not None} + + +def sigmas(rule: dict, record: dict, spreads: dict[str, float]) -> dict[str, tuple[float | None, float]]: + """(s_b, sigma_b) per benchmark of the record.""" + r = rule_for(rule, repo_of(record["task"])) + nops, s_repo = nop_logs(record), spreads.get(repo_of(record["task"])) + out = {} + for b in {b for run in record["runs"] for b in run["log_speedups"]}: + s_b = robust_sd(nops.get(b, []), center=0.0) + parts = [(len(nops[b]), s_b)] if s_b is not None else [] + parts += [(r["n0"], s_repo)] if s_repo is not None else [] + n = sum(w for w, _ in parts) + out[b] = (s_b, math.sqrt(sum(w * s * s for w, s in parts) / n) if n else math.inf) + return out + + +def thetas(rule: dict, record: dict, spreads: dict[str, float]) -> dict[str, float]: + r = rule_for(rule, repo_of(record["task"])) + if r["form"] == "global": + return {b: r["theta"] for run in record["runs"] for b in run["log_speedups"]} + return { + b: min(max(r["k"] * sigma, r["theta_min"]), r["theta_max"]) + for b, (_, sigma) in sigmas(rule, record, spreads).items() + } + + +def moved(values: list[float | None], bar: float, need: int = ORACLE_MIN_RUNS) -> tuple[bool, bool]: + """(up, down) for one benchmark: at least `need` runs past +bar / -bar.""" + xs = [v for v in values if v is not None] + need = min(need, len(xs)) if xs else 1 + return sum(v > bar for v in xs) >= need, sum(v < -bar for v in xs) >= need + + +def oracle_problem(run: dict, max_broken: float = ORACLE_MAX_BROKEN) -> str | None: + """Why an oracle run cannot be the reference (crashed, measured nothing, broke most tests), else None.""" + tests = run.get("tests") or {} + if run.get("exception"): + return run["exception"] + if not run["log_speedups"]: + return "no_benchmarks" + if tests.get("n_base_pass") and (tests["n_tests"] == 0 or tests["n_regressed"] > max_broken * tests["n_base_pass"]): + return "tests_broken" + return None + + +def classify(record: dict, rule: dict, spreads: dict[str, float]) -> dict[str, Any]: + """Per-task counts on the k·sigma_b bar: oracle up, oracle down, no-change values past the bar, and the best benchmarks. + + The bar is k·sigma_b (no clip), the same bar the review page shows; `verify` below uses the clipped theta_b. + """ + k = float(rule_for(rule, repo_of(record["task"])).get("k", 2)) + orc = sorted((r for r in record["runs"] if r["role"] != "nop"), key=lambda r: r["run"]) + nop = sorted((r for r in record["runs"] if r["role"] == "nop"), key=lambda r: r["run"]) + up, down, alarms, nop_values, medians = [], [], 0, 0, [] + for b, (_, sigma) in sorted(sigmas(rule, record, spreads).items()): + bar = k * sigma + o = [r["log_speedups"].get(b) for r in orc] + for v in (r["log_speedups"].get(b) for r in nop): + if v is not None: + nop_values += 1 + alarms += abs(v) > bar + if o and None not in o: + u, d = moved(o, bar) + up += [b] if u else [] + down += [b] if d else [] + medians.append((statistics.median(o), b)) + best = [{"benchmark": b, "speedup": math.exp(m)} for m, b in sorted(medians, reverse=True)[:3]] + return { + "task": record["task"], + "n_benchmarks": len(sigmas(rule, record, spreads)), + "oracle_up": len(up), + "oracle_down": len(down), + "up": up, + "down": down, + "false_alarms": alarms, + "n_nop_values": nop_values, + "best": best, + } + + +def verify(record: dict, rule: dict, spreads: dict[str, float]) -> dict[str, Any]: + """All oracle runs present and usable, on a reused baseline; at least one benchmark up beyond theta_b in 2 runs and none + moving both ways below that count; no-change bias on the improved benchmarks within theta_b.""" + theta = thetas(rule, record, spreads) + r_task = rule_for(rule, repo_of(record["task"])) + oracle = [r for r in record["runs"] if r["role"] != "nop"] + problems = { + r["run"]: p for r in oracle if (p := oracle_problem(r, r_task.get("oracle_max_broken", ORACLE_MAX_BROKEN))) + } + runs = [r["log_speedups"] for r in oracle] + common = set.intersection(*map(set, runs)) if runs else set() + need = min(int(r_task.get("oracle_min_runs", ORACLE_MIN_RUNS)), len(runs)) + n_up = {b: sum(r[b] > theta[b] for r in runs) for b in common} + n_down = {b: sum(r[b] < -theta[b] for r in runs) for b in common} + up = sorted(b for b in common if n_up[b] >= need) + conflicts = sorted(b for b in common if n_up[b] and n_down[b] and max(n_up[b], n_down[b]) < need) + median = {b: statistics.median(r[b] for r in runs) for b in sorted(common)} + nops = nop_logs(record) + bias = {b: statistics.median(nops[b]) if nops.get(b) else None for b, m in median.items() if m > theta[b]} + ratios = [v / theta[b] for b, v in bias.items() if v is not None] + if r_task.get("bias_check", "aggregate") == "aggregate": + unbiased = bool(ratios) and abs(statistics.mean(ratios)) < 1 + else: + unbiased = bool(ratios) and len(ratios) == len(bias) and all(abs(x) < 1 for x in ratios) + checks = { + "oracle_ok": len(oracle) == record["n_oracle"] and not problems, + "baseline_reused": all(r.get("baseline_reused") is not False for r in record["runs"]), + "speedup": bool(up) and not conflicts, + "unbiased": unbiased, + } + entry = {b: {"oracle_log_speedup": m, "oracle_speedup": math.exp(m), "theta": theta[b]} for b, m in median.items()} + failing = sorted({t for r in oracle for t in (r.get("tests") or {}).get("failing", [])}) + return { + "task": record["task"], + "verified": all(checks.values()), + "checks": checks, + "rule": rule, + "up": up, + "sign_conflicts": conflicts, + "nop_bias": bias, + "oracle_problems": problems, + "entry": {"benchmarks": entry, "oracle_failing_tests": failing}, + } + + +def margin(agent_logs: dict[str, float], verdict: dict) -> float | None: + """Mean over oracle-improved benchmarks of (attempt - no-change bias) / theta_b; below 1 is within noise of no change.""" + e = verdict["entry"]["benchmarks"] + xs = [ + (agent_logs[b] - v) / e[b]["theta"] for b, v in verdict["nop_bias"].items() if v is not None and b in agent_logs + ] + return statistics.mean(xs) if xs else None + + +def classify_all(records: dict[str, dict], rule: dict) -> dict[str, dict[str, Any]]: + spreads = repo_spreads(list(records.values())) + return {t: classify(rec, rule, spreads) for t, rec in records.items()} diff --git a/src/datasmith/verify/report.py b/src/datasmith/verify/report.py new file mode 100644 index 00000000..e8b4d4d3 --- /dev/null +++ b/src/datasmith/verify/report.py @@ -0,0 +1,162 @@ +"""Review summaries for measured tasks, and the build error behind a failed image.""" + +from __future__ import annotations + +import collections +import math +import re +import statistics +from pathlib import Path +from typing import Any + +from datasmith.verify.noise import moved, repo_spreads, rule_for, sigmas +from datasmith.verify.survey import Survey, display_id, repo_of + +_SOLUTION = re.compile(r"<< '__SOLUTION__'\n(.*?)\n__SOLUTION__", re.DOTALL) + + +def oracle_diff(task_dir: Path) -> str: + """The oracle diff inside solution/solve.sh.""" + path = task_dir / "solution" / "solve.sh" + m = _SOLUTION.search(path.read_text()) if path.exists() else None + return m.group(1) if m else "" + + +def files_in_diff(diff: str) -> list[str]: + files = [line[6:].strip() for line in diff.splitlines() if line.startswith("+++ b/")] + return list(dict.fromkeys(files)) + + +def prompt_task(instruction: str) -> str: + """The task description section of instruction.md.""" + text = instruction.split("**Task Description**", 1)[-1].split("## Task Requirements", 1)[0] + lines = [ln for ln in text.strip().splitlines() if ln.strip() and not ln.startswith("Your main goal")] + return " ".join(lines).replace("Use the following information if needed to understand the problem:", "").strip() + + +def touched_packages(files: list[str]) -> set[str]: + """Top-level package names under the source root that the diff changes, tests excluded.""" + out = set() + for p in files: + if "/tests/" in p or p.startswith("tests/"): + continue + parts = p.split("/") + out.add(parts[1] if len(parts) > 2 else parts[-1].rsplit(".", 1)[0]) + return out + + +def summarize(survey: Survey, task: str, record: dict, rule: dict, spreads: dict[str, float]) -> dict[str, Any]: + """Facts for review: prompt, patch, oracle and no-change moves on the k·sigma_b bar, by suite. No verdict.""" + k = float(rule_for(rule, repo_of(task)).get("k", 2)) + orc = sorted((r for r in record["runs"] if r["role"] != "nop"), key=lambda r: r["run"]) + nop = sorted((r for r in record["runs"] if r["role"] == "nop"), key=lambda r: r["run"]) + up, down, nop_up, nop_down, bars = [], [], 0, 0, [] + for b, (_, sigma) in sigmas(rule, record, spreads).items(): + bar = k * sigma + bars.append(bar) + o = [r["log_speedups"].get(b) for r in orc] + if o and None not in o: + u, d = moved(o, bar) + m = math.exp(statistics.median(o)) + up += [(m, b)] if u else [] + down += [(m, b)] if d else [] + n = [r["log_speedups"].get(b) for r in nop] + if any(v is not None for v in n): + u, d = moved(n, bar) + nop_up, nop_down = nop_up + u, nop_down + d + task_dir = survey.task_dir(task) + diff = oracle_diff(task_dir) + files = files_in_diff(diff) + pkgs = touched_packages(files) + instruction = (task_dir / "instruction.md").read_text() if (task_dir / "instruction.md").exists() else "" + return { + "task": task, + "prompt": prompt_task(instruction), + "diff_lines": diff.count("\n"), + "files": files, + "n_benchmarks": len(bars), + "bar_median": math.exp(statistics.median(bars)) if bars else None, + "oracle_up": len(up), + "oracle_down": len(down), + "nop_up": nop_up, + "nop_down": nop_down, + "up_in_changed_packages": sum(any(p in b.split(".")[0] for p in pkgs) for _, b in up), + "up_by_suite": collections.Counter(b.split(".")[0] for _, b in up).most_common(8), + "top_up": sorted(up, reverse=True)[:5], + "top_down": sorted(down)[:3], + } + + +def render(s: dict[str, Any]) -> str: + lines = [ + f"### {s['task']}", + f"- Prompt: {s['prompt'][:400]}", + f"- Patch: {s['diff_lines']} diff lines in {len(s['files'])} files: {', '.join(s['files'][:10])}", + f"- Benchmarks: {s['n_benchmarks']}; noise bar median {s['bar_median']:.3f}x" + if s["bar_median"] + else "- Benchmarks: 0", + f"- Oracle up {s['oracle_up']}, down {s['oracle_down']}; no-change runs, same rule: up {s['nop_up']}, down {s['nop_down']}", + f"- Up benchmarks in suites named after a changed package: {s['up_in_changed_packages']} of {s['oracle_up']}", + f"- Up by suite: {', '.join(f'{n} {c}' for n, c in s['up_by_suite']) or 'none'}", + ] + lines += [f" - up {m:.3f}x {b}" for m, b in s["top_up"]] + lines += [f" - down {m:.3f}x {b}" for m, b in s["top_down"]] + return "\n".join(lines) + + +def summaries(survey: Survey, rule: dict, tasks: list[str] | None = None) -> list[dict[str, Any]]: + """Summaries for the given tasks, or for every measured task still pending review.""" + records = survey.records() + spreads = repo_spreads(list(records.values())) + if tasks is None: + pending = {r["task"].lower() for r in survey.reviews() if r["decision"] in ("", "pending")} + states = survey.states() + tasks = [ + t for t in records if states.get(t, {}).get("phase") == "measured" and display_id(t).lower() in pending + ] + return [summarize(survey, t, records[t], rule, spreads) for t in tasks if t in records] + + +_ERROR = re.compile( + r"(ModuleNotFoundError|ImportError|AttributeError|NameError|TypeError|ValueError|NoBenchmarksError|ASVError" + r"|UserError: No __init__|CancelledError|no asv config resolved)[^\n]*" +) +_SKIP = re.compile(r"benchmarks\.json missing|During handling|Failed to build the project") +_MISSING = re.compile( + r"No module named '([\w.]+)'|Missing optional dependency '([\w.-]+)'|[Pp]lease install '([\w.-]+)'" + r"|require either `([\w.-]+)`" +) +FAILURE_CLASSES = ( + ("missing_module", re.compile(r"ModuleNotFoundError|Missing optional dependency|[Pp]lease install|require either")), + ("syntax_error", re.compile(r"SyntaxError|IndentationError")), + ( + "version_api_mismatch", + re.compile( + r"cannot import name|unexpected keyword|has no attribute|unsupported operand" + r"|dtype size changed|separate project" + ), + ), + ("missing_system_library", re.compile(r"\.so[.\d]*: cannot open shared object")), + ("no_benchmark_package", re.compile(r"No __init__\.py")), + ("no_benchmarks", re.compile(r"NoBenchmarksError")), + ("dictionary_changed", re.compile(r"dictionary changed size during iteration")), +) + + +def failure_reason(log: Path) -> dict[str, str]: + """The last error line of an image build log, its class and the first missing module after benchmark discovery.""" + lines = log.read_text(errors="replace").splitlines() + hits = [m.group(0).strip() for line in lines if not _SKIP.search(line) for m in [_ERROR.search(line)] if m] + hits += [line.strip() for line in lines if "dictionary changed size during iteration" in line] + why = hits[-1][:200] if hits else "" + start = max( + (i for i, line in enumerate(lines) if "Error running" in line and "benchmark.py discover" in line), + default=len(lines), + ) + found = next((m for line in lines[start:] for m in [_MISSING.search(line)] if m), None) + missing = next((g for g in found.groups() if g), "") if found else "" + cls = next( + (name for name, rx in FAILURE_CLASSES if rx.search(why) or (name == "missing_module" and missing)), + "other" if why else "no_error_line", + ) + return {"reason": why, "class": cls, "missing": missing} diff --git a/src/datasmith/verify/review.py b/src/datasmith/verify/review.py new file mode 100644 index 00000000..1eb9d863 --- /dev/null +++ b/src/datasmith/verify/review.py @@ -0,0 +1,100 @@ +"""Review decisions in the survey's manual_review.csv, and the automatic not_approved rules. + +Automatic rules only turn a pending task into not_approved; they never approve and never change a decided row. +""" + +from __future__ import annotations + +import datetime +from typing import Any + +from datasmith.verify.noise import classify_all, load_rule +from datasmith.verify.survey import DECISIONS, UNMEASURED, Survey, display_id, task_id + +NOTE_UP_ZERO = "auto: no oracle benchmark above noise (oracle up 0)" +NOTE_UNMEASURED = { + "uncovered": "uncovered: no benchmark runs the code the patch changes", + "base_edited": "base edited: the patch changes the benchmark files", + "patch_failed": "patch failed: the patch did not apply in the image", + "no_benchmarks": "no benchmarks: benchmark suite empty or failed to load at the base commit", + "image_failed": "image build failed", + "error": "build failed (state error)", + "too_slow": "too slow: the oracle measurement exceeded the time limit", +} + + +def set_decision( + survey: Survey, task: str, decision: str, note: str = "", data: str = "phase 1", reviewer: str = "user" +) -> dict[str, str]: + """Record one decision; `task` is owner/repo#N or owner__repo__N.""" + if decision not in DECISIONS: + raise ValueError(f"decision must be one of {DECISIONS}") + key = (display_id(task) if "__" in task else task).lower() + rows = survey.reviews() + hit = [r for r in rows if r["task"].lower() == key] + if not hit: + raise KeyError(f"{task}: not in {survey.review_path}") + today = datetime.date.today().isoformat() + hit[0].update( + decision=decision, reviewer=reviewer, date=today, data=data, note=note or f"{reviewer} review {today}" + ) + survey.write_reviews(rows) + return hit[0] + + +def add_pending(survey: Survey, tasks: list[str]) -> list[str]: + """Add a pending row for every task (owner__repo__N) that has none; returns the added display ids.""" + rows = survey.reviews() + have = {r["task"].lower() for r in rows} + added = [display_id(t) for t in tasks if display_id(t).lower() not in have] + if added: + rows += [{"task": t, "decision": "pending", "reviewer": "", "date": "", "data": "", "note": ""} for t in added] + survey.write_reviews(rows) + return added + + +def auto_decisions(survey: Survey) -> list[dict[str, Any]]: + """Pending tasks that an automatic rule marks not_approved, with the rule's note. + + Rules: a measured task with no oracle benchmark up beyond noise; a task in an unmeasured end state. + """ + states = survey.states() + records = survey.records() + stats = classify_all(records, load_rule(survey.rule_path)) if records and survey.rule_path.exists() else {} + out = [] + for row in survey.reviews(): + if row["decision"] not in ("", "pending"): + continue + task = task_id(row["task"]) + phase = states.get(task, {}).get("phase") + if phase == "measured" and task in stats and stats[task]["oracle_up"] == 0: + s = stats[task] + out.append({ + "task": task, + "rule": "oracle_up_0", + "note": NOTE_UP_ZERO, + "detail": f"oracle up 0, down {s['oracle_down']}, benchmarks {s['n_benchmarks']}", + }) + elif phase in UNMEASURED: + out.append({ + "task": task, + "rule": phase, + "note": f"auto: no usable measurement; {NOTE_UNMEASURED[phase]}", + "detail": phase, + }) + return out + + +def apply_auto(survey: Survey, dry_run: bool = False) -> list[dict[str, Any]]: + picked = auto_decisions(survey) + if picked and not dry_run: + notes = {display_id(d["task"]).lower(): d["note"] for d in picked} + today = datetime.date.today().isoformat() + rows = survey.reviews() + for r in rows: + if r["task"].lower() in notes: + r.update( + decision="not_approved", reviewer="auto", date=today, data="phase 1", note=notes[r["task"].lower()] + ) + survey.write_reviews(rows) + return picked diff --git a/src/datasmith/verify/status.py b/src/datasmith/verify/status.py new file mode 100644 index 00000000..99335654 --- /dev/null +++ b/src/datasmith/verify/status.py @@ -0,0 +1,23 @@ +"""Survey counts: task phases from the state files and decisions from the review file.""" + +from __future__ import annotations + +import collections +from typing import Any + +from datasmith.verify.survey import Survey, task_id + + +def counts(survey: Survey) -> dict[str, Any]: + states = survey.states() + reviews = survey.reviews() + phases = collections.Counter(s.get("phase") or "not_started" for s in states.values()) + reviewed = {task_id(r["task"]) for r in reviews} + if reviewed - set(states): + phases["not_started"] += len(reviewed - set(states)) + return { + "tasks": len(reviewed | set(states)), + "phases": dict(phases.most_common()), + "review": dict(collections.Counter(r["decision"] or "pending" for r in reviews).most_common()), + "nodes": dict(collections.Counter(str(s.get("node") or "unknown") for s in states.values()).most_common()), + } diff --git a/src/datasmith/verify/survey.py b/src/datasmith/verify/survey.py new file mode 100644 index 00000000..143d9f7a --- /dev/null +++ b/src/datasmith/verify/survey.py @@ -0,0 +1,92 @@ +"""Survey folder layout: task ids, state files, oracle records, task folders and the review file.""" + +from __future__ import annotations + +import csv +import json +import re +import shutil +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +REVIEW_FIELDS = ("task", "decision", "reviewer", "date", "data", "note") +DECISIONS = ("approved", "not_approved", "pending") +# End states of a task that has no usable oracle measurement. +UNMEASURED = ("uncovered", "base_edited", "patch_failed", "no_benchmarks", "image_failed", "error", "too_slow") + +_DISPLAY = re.compile(r"([\w.-]+)/([\w.-]+)#(\d+)") + + +def task_id(display: str) -> str: + """``owner/repo#N`` -> ``owner__repo__N``.""" + m = _DISPLAY.fullmatch(display.strip()) + if not m: + raise ValueError(f"not an owner/repo#N id: {display!r}") + return "__".join(m.groups()) + + +def display_id(task: str) -> str: + """``owner__repo__N`` -> ``owner/repo#N``.""" + owner, rest = task.split("__", 1) + repo, num = rest.rsplit("__", 1) + return f"{owner}/{repo}#{num}" + + +def repo_of(task: str) -> str: + return task.split("__")[1] + + +@dataclass(frozen=True) +class Survey: + root: Path + + @property + def review_path(self) -> Path: + return self.root / "manual_review.csv" + + @property + def rule_path(self) -> Path: + return self.root / "threshold_rule.json" + + def state(self, task: str) -> dict[str, Any]: + path = self.root / "state" / f"{task}.json" + return json.loads(path.read_text()) if path.exists() else {} + + def states(self) -> dict[str, dict[str, Any]]: + return {p.stem: json.loads(p.read_text()) for p in sorted((self.root / "state").glob("*.json"))} + + def records(self) -> dict[str, dict[str, Any]]: + """Oracle records with at least one measured run, keyed by task.""" + out = {} + for path in sorted((self.root / "oracle").glob("*.json")): + try: + rec = json.loads(path.read_text()) + except json.JSONDecodeError: + continue + if any(run.get("log_speedups") for run in rec.get("runs", [])): + out[rec["task"]] = rec + return out + + def task_dir(self, task: str) -> Path: + return self.root / "pinned" / task + + def reviews(self) -> list[dict[str, str]]: + if not self.review_path.exists(): + return [] + with self.review_path.open() as f: + return [{k: row.get(k, "") or "" for k in REVIEW_FIELDS} for row in csv.DictReader(f)] + + def write_reviews(self, rows: list[dict[str, str]]) -> None: + """Write the review file, keeping one timestamped backup of the previous version.""" + if self.review_path.exists(): + shutil.copy2( + self.review_path, self.review_path.with_name(f"manual_review.csv.bak-{time.strftime('%Y%m%d-%H%M%S')}") + ) + tmp = self.review_path.with_suffix(".csv.tmp") + with tmp.open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=REVIEW_FIELDS, lineterminator="\n") + w.writeheader() + w.writerows(rows) + tmp.replace(self.review_path) diff --git a/tests/verify/__init__.py b/tests/verify/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/verify/conftest.py b/tests/verify/conftest.py new file mode 100644 index 00000000..3c567dc6 --- /dev/null +++ b/tests/verify/conftest.py @@ -0,0 +1,50 @@ +import csv +import json +import math +from pathlib import Path + +import pytest + +RULE = {"form": "per_benchmark", "k": 2, "n0": 10, "theta_min": 0.0296, "theta_max": 0.3, "bias_check": "aggregate"} +FIELDS = ["task", "decision", "reviewer", "date", "data", "note"] + + +def run(role: str, n: int, logs: dict[str, float]) -> dict: + return {"role": role, "run": n, "log_speedups": logs, "baseline_reused": True, "tests": {}} + + +def record(task: str, oracle: list[dict[str, float]], nop: list[dict[str, float]]) -> dict: + runs = [run("oracle_baseline", i + 1, x) for i, x in enumerate(oracle)] + runs += [run("nop", i + 1, x) for i, x in enumerate(nop)] + return {"task": task, "n_oracle": len(oracle), "n_nop": len(nop), "runs": runs} + + +@pytest.fixture +def survey(tmp_path: Path) -> Path: + """Three tasks: one with a clear gain, one measured with no gain, one uncovered; all pending.""" + for d in ("state", "oracle", "pinned"): + (tmp_path / d).mkdir() + (tmp_path / "threshold_rule.json").write_text(json.dumps(RULE)) + quiet = [ + {"a.time_x": 0.001, "a.time_y": -0.002}, + {"a.time_x": -0.001, "a.time_y": 0.0}, + {"a.time_x": 0.0, "a.time_y": 0.001}, + ] + fast = record("o__r__1", [{"a.time_x": math.log(1.5), "a.time_y": 0.0}] * 3, quiet) + flat = record("o__r__2", [{"a.time_x": 0.001, "a.time_y": -0.001}] * 3, quiet) + for rec in (fast, flat): + (tmp_path / "oracle" / f"{rec['task']}.json").write_text(json.dumps(rec)) + for task, phase in (("o__r__1", "measured"), ("o__r__2", "measured"), ("o__r__3", "uncovered")): + (tmp_path / "state" / f"{task}.json").write_text(json.dumps({"task": task, "phase": phase})) + td = tmp_path / "pinned" / "o__r__1" + (td / "solution").mkdir(parents=True) + (td / "solution" / "solve.sh").write_text( + "cat > p.diff << '__SOLUTION__'\n--- a/pkg/a/core.py\n+++ b/pkg/a/core.py\n@@ -1 +1 @@\n-x\n+y\n__SOLUTION__\n" + ) + (td / "instruction.md").write_text("**Task Description**\n\nThe code is slow.\n\n## Task Requirements\n- x\n") + with (tmp_path / "manual_review.csv").open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=FIELDS) + w.writeheader() + for t in ("o/r#1", "o/r#2", "o/r#3"): + w.writerow({"task": t, "decision": "pending"}) + return tmp_path diff --git a/tests/verify/test_verify.py b/tests/verify/test_verify.py new file mode 100644 index 00000000..9d737f15 --- /dev/null +++ b/tests/verify/test_verify.py @@ -0,0 +1,97 @@ +import json +import math + +from datasmith.verify import report, review, status +from datasmith.verify.cli import main +from datasmith.verify.noise import classify_all, load_rule, moved, robust_sd +from datasmith.verify.survey import Survey, display_id, task_id + + +def test_ids_round_trip(): + assert task_id("scikit-learn/scikit-learn#123") == "scikit-learn__scikit-learn__123" + assert display_id("pydata__xarray__3588") == "pydata/xarray#3588" + + +def test_moved_needs_two_of_three_runs(): + assert moved([0.2, 0.2, 0.0], 0.1) == (True, False) + assert moved([0.2, 0.0, 0.0], 0.1) == (False, False) + assert moved([-0.2, -0.2, None], 0.1) == (False, True) + + +def test_robust_sd_is_centered_at_zero(): + assert robust_sd([0.1, -0.1, 0.1], center=0.0) == 1.4826 * 0.1 + assert robust_sd([], center=0.0) is None + + +def test_classify_counts_gain_and_no_gain(survey): + s = Survey(survey) + stats = classify_all(s.records(), load_rule(s.rule_path)) + assert stats["o__r__1"]["oracle_up"] == 1 + assert stats["o__r__1"]["best"][0]["benchmark"] == "a.time_x" + assert math.isclose(stats["o__r__1"]["best"][0]["speedup"], 1.5) + assert stats["o__r__2"]["oracle_up"] == 0 + + +def test_auto_rules_pick_no_gain_and_unmeasured(survey): + s = Survey(survey) + picked = {d["task"]: d["rule"] for d in review.auto_decisions(s)} + assert picked == {"o__r__2": "oracle_up_0", "o__r__3": "uncovered"} + review.apply_auto(s) + decided = {r["task"]: r["decision"] for r in s.reviews()} + assert decided == {"o/r#1": "pending", "o/r#2": "not_approved", "o/r#3": "not_approved"} + assert review.auto_decisions(s) == [] + + +def test_auto_rules_never_change_a_decided_row(survey): + s = Survey(survey) + review.set_decision(s, "o/r#2", "approved", note="kept by hand") + assert "o__r__2" not in {d["task"] for d in review.auto_decisions(s)} + + +def test_set_decision_keeps_a_backup(survey): + s = Survey(survey) + review.set_decision(s, "o__r__1", "approved", note="gold") + assert list(survey.glob("manual_review.csv.bak-*")) + assert next(r for r in s.reviews() if r["task"] == "o/r#1")["note"] == "gold" + + +def test_add_pending_adds_only_new_rows(survey): + s = Survey(survey) + assert review.add_pending(s, ["o__r__1", "o__r__9"]) == ["o/r#9"] + assert len(s.reviews()) == 4 + + +def test_report_summarizes_pending_measured_tasks(survey): + s = Survey(survey) + out = {x["task"]: x for x in report.summaries(s, load_rule(s.rule_path))} + assert set(out) == {"o__r__1", "o__r__2"} + assert out["o__r__1"]["files"] == ["pkg/a/core.py"] + assert out["o__r__1"]["prompt"] == "The code is slow." + assert out["o__r__1"]["up_in_changed_packages"] == 1 + assert "### o__r__1" in report.render(out["o__r__1"]) + + +def test_failure_reason_classes(tmp_path): + log = tmp_path / "t.log" + log.write_text("Error running benchmark.py discover\nModuleNotFoundError: No module named 'numba'\n") + assert report.failure_reason(log) == { + "reason": "ModuleNotFoundError: No module named 'numba'", + "class": "missing_module", + "missing": "numba", + } + log.write_text("RuntimeError: dictionary changed size during iteration\n") + assert report.failure_reason(log)["class"] == "dictionary_changed" + + +def test_status_counts(survey): + c = status.counts(Survey(survey)) + assert c["tasks"] == 3 + assert c["phases"] == {"measured": 2, "uncovered": 1} + + +def test_cli_status_and_dry_run(survey, capsys): + assert main(["status", str(survey)]) == 0 + assert json.loads(capsys.readouterr().out)["review"] == {"pending": 3} + assert main(["auto", str(survey), "--dry-run"]) == 0 + assert "would mark 2 not_approved" in capsys.readouterr().out + assert {r["decision"] for r in Survey(survey).reviews()} == {"pending"} From 7fcb5eb33c7bc10ed09df06343608ae97914c686 Mon Sep 17 00:00:00 2001 From: Arjun Sharma Date: Mon, 19 Oct 2026 03:48:27 +0000 Subject: [PATCH 02/32] verify: precheck stage (patch applies, base edits, benchmark coverage) with the in-image check --- pyproject.toml | 6 +- src/datasmith/verify/cli.py | 34 +++++- src/datasmith/verify/in_image/__init__.py | 0 .../verify/in_image/coverage_check.py | 64 +++++++++++ src/datasmith/verify/precheck.py | 100 ++++++++++++++++++ src/datasmith/verify/survey.py | 9 ++ tests/verify/test_precheck.py | 71 +++++++++++++ 7 files changed, 277 insertions(+), 7 deletions(-) create mode 100644 src/datasmith/verify/in_image/__init__.py create mode 100644 src/datasmith/verify/in_image/coverage_check.py create mode 100644 src/datasmith/verify/precheck.py create mode 100644 tests/verify/test_precheck.py diff --git a/pyproject.toml b/pyproject.toml index 53b1b72e..2a46d92a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,7 +80,7 @@ packages = ["src/datasmith"] [tool.mypy] files = ["src"] -exclude = ["src/datasmith/agents/templates/", "src/datasmith/docker/templates/", "src/datasmith/harbor_adapter/template/"] +exclude = ["src/datasmith/agents/templates/", "src/datasmith/docker/templates/", "src/datasmith/harbor_adapter/template/", "src/datasmith/verify/in_image/"] disallow_untyped_defs = true disallow_any_unimported = true no_implicit_optional = true @@ -106,7 +106,7 @@ markers = [ target-version = "py312" line-length = 120 fix = true -extend-exclude = ["archive/", "dataset/", "scratch/", "src/datasmith/docker/templates/", "src/datasmith/agents/templates/", "src/datasmith/harbor_adapter/template/"] +extend-exclude = ["archive/", "dataset/", "scratch/", "src/datasmith/docker/templates/", "src/datasmith/agents/templates/", "src/datasmith/harbor_adapter/template/", "src/datasmith/verify/in_image/"] force-exclude = true [tool.ruff.lint] @@ -184,7 +184,7 @@ branch = true source = ["src"] [tool.deptry] -exclude = ["src/datasmith/docker/templates/", "src/datasmith/agents/templates/", "src/datasmith/harbor_adapter/template/"] +exclude = ["src/datasmith/docker/templates/", "src/datasmith/agents/templates/", "src/datasmith/harbor_adapter/template/", "src/datasmith/verify/in_image/"] ignore = ["DEP003"] [tool.deptry.per_rule_ignores] DEP001 = ["tomllib", "tomli"] diff --git a/src/datasmith/verify/cli.py b/src/datasmith/verify/cli.py index 429b045a..d7a73e73 100644 --- a/src/datasmith/verify/cli.py +++ b/src/datasmith/verify/cli.py @@ -6,6 +6,7 @@ fc-verify review SURVEY TASK DECISION [--note] fc-verify auto SURVEY [--dry-run] automatic not_approved rules fc-verify add SURVEY TASK ... pending rows for new tasks +fc-verify precheck SURVEY TASK ... [--measure-uncovered always_affected,compiled] [--rerun] fc-verify failure LOG ... error class of failed image builds """ @@ -16,7 +17,7 @@ import sys from pathlib import Path -from datasmith.verify import report, review, status +from datasmith.verify import precheck, report, review, status from datasmith.verify.noise import classify_all, load_rule from datasmith.verify.survey import DECISIONS, Survey, task_id @@ -56,7 +57,28 @@ def _add(survey: Survey, a: argparse.Namespace) -> None: print("\n".join(review.add_pending(survey, _tasks(a.tasks)))) -COMMANDS = {"status": _status, "classify": _classify, "report": _report, "review": _review, "auto": _auto, "add": _add} +def _precheck(survey: Survey, a: argparse.Namespace) -> None: + kinds = set(filter(None, a.measure_uncovered.split(","))) + for t in _tasks(a.tasks): + st = survey.state(t) + if not st.get("tag"): + print(json.dumps({"task": t, "error": "no image tag in state"})) + continue + end, result = precheck.precheck(survey, t, st["tag"], kinds, rerun=a.rerun) + if end: + survey.save_state(t, {**st, "phase": end}) + print(json.dumps({"task": t, "end": end, **{k: result.get(k) for k in ("by_deps", "patch_failed", "error")}})) + + +COMMANDS = { + "status": _status, + "classify": _classify, + "report": _report, + "review": _review, + "auto": _auto, + "add": _add, + "precheck": _precheck, +} def main(argv: list[str] | None = None) -> int: @@ -64,16 +86,20 @@ def main(argv: list[str] | None = None) -> int: prog="fc-verify", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) sub = ap.add_subparsers(dest="cmd", required=True) - for name in ("status", "classify", "report", "review", "auto", "add"): + for name in ("status", "classify", "report", "review", "auto", "add", "precheck"): p = sub.add_parser(name) p.add_argument("survey", type=Path) - if name in ("classify", "report", "add"): + if name in ("classify", "report", "add", "precheck"): p.add_argument("tasks", nargs="*") sub.choices["review"].add_argument("task") sub.choices["review"].add_argument("decision", choices=DECISIONS) sub.choices["review"].add_argument("--note", default="") sub.choices["review"].add_argument("--data", default="phase 1") sub.choices["auto"].add_argument("--dry-run", action="store_true") + sub.choices["precheck"].add_argument( + "--measure-uncovered", default="", help=",".join(precheck.MEASURE_UNCOVERED_KINDS) + ) + sub.choices["precheck"].add_argument("--rerun", action="store_true") sub.add_parser("failure").add_argument("logs", nargs="+", type=Path) a = ap.parse_args(argv) diff --git a/src/datasmith/verify/in_image/__init__.py b/src/datasmith/verify/in_image/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/datasmith/verify/in_image/coverage_check.py b/src/datasmith/verify/in_image/coverage_check.py new file mode 100644 index 00000000..2bf709f2 --- /dev/null +++ b/src/datasmith/verify/in_image/coverage_check.py @@ -0,0 +1,64 @@ +"""Runs inside a task image (Python 3.8 or newer): apply the oracle patch, run LSV's selection on the image's deps DB, print one JSON line. + +by_deps = benchmarks selected through a dependency on a file the patch changes (always_affected ones excluded). +patch_failed = files where `patch` rejected a hunk (the oracle is then not the PR's change). +edited_before = tracked files the image already has edited, before the patch (some base images ship part of the fix). +edited_overlap = those files whose edited lines touch an oracle hunk (its changed or context lines). +""" +import json +import os +import re +import subprocess + +DB = "/opt/lsv/cache/results/.lightspeed_deps.db" +REPO = "/workspace/repo" +HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? ") + + +def old_ranges(diff): + """{file: [(first, last)]} old-side line ranges of each hunk; a pure insertion covers the line it follows.""" + out, cur = {}, None + for line in diff.splitlines(): + if line.startswith("+++ "): + cur = line[6:] if line.startswith("+++ b/") else None + if cur is not None: + out.setdefault(cur, []) + continue + m = HUNK.match(line) + if m and cur is not None: + start, n = int(m.group(1)), int(m.group(2) or 1) + out[cur].append((start, start + max(n, 1) - 1) if n else (start, start + 1)) + return out + + +def overlapping(image_diff, patch_text): + """Files whose image edits touch a hunk of the oracle patch.""" + edits, patch = old_ranges(image_diff), old_ranges(patch_text) + return sorted(f for f, rs in edits.items() if any(a <= d and c <= b for a, b in rs for c, d in patch.get(f, []))) + + +def main(): + from asv.contrib.lightspeed.deps_db import LightspeedDB + from asv.contrib.lightspeed.fingerprint import changed_files_with_fingerprints + + edited_before = subprocess.check_output(["git", "diff", "--name-only", "HEAD"], cwd=REPO, text=True).split() + image_diff = subprocess.check_output(["git", "diff", "-U0", "HEAD"], cwd=REPO, text=True, errors="replace") + with open("/tmp/solve.sh", errors="replace") as f: + edited_overlap = overlapping(image_diff, f.read()) + # Trials diff against a commit of the starting tree (setup.sh), so the image's own edits are not changed files. + subprocess.run("git add -A && git -c user.name=fc -c user.email=fc@local commit -q --no-verify --allow-empty -m fc-baseline", + shell=True, cwd=REPO, check=True, capture_output=True) + subprocess.run(["bash", "/tmp/solve.sh"], cwd=REPO, capture_output=True) + rejected = [p[:-4] for p in subprocess.check_output(["git", "ls-files", "--others", "--", "*.rej"], cwd=REPO, text=True).split()] + changed = [os.path.join(REPO, p) for p in subprocess.check_output(["git", "diff", "--name-only"], cwd=REPO, text=True).split()] + db = LightspeedDB(DB) + always = {r[0] for r in db._con.execute("select benchmark_id from benchmark_meta where always_affected = 1")} + total = db._con.execute("select count(*) from (select benchmark_id from benchmark_dep union select benchmark_id from benchmark_meta)").fetchone()[0] + affected = {str(b) for b in db.get_affected_benchmark_ids(changed_files_with_fingerprints(changed, db.get_stored_fshas()))} + print(json.dumps({"changed_files": len(changed), "benchmarks": total, "always_affected": len(always), + "selected": len(affected), "by_deps": len(affected - always), "patch_failed": rejected, + "edited_before": edited_before, "edited_overlap": edited_overlap})) + + +if __name__ == "__main__": + main() diff --git a/src/datasmith/verify/precheck.py b/src/datasmith/verify/precheck.py new file mode 100644 index 00000000..44ca652f --- /dev/null +++ b/src/datasmith/verify/precheck.py @@ -0,0 +1,100 @@ +"""Static checks before timing: does the patch apply, did the image already carry part of it, does a benchmark cover it. + +The check runs `in_image/coverage_check.py` in the task image without network; its result is kept in +`/coverage/.json` and reused while it has no error. +""" + +from __future__ import annotations + +import json +import re +import subprocess +from pathlib import Path +from typing import Any + +from datasmith.verify.survey import Survey + +CHECK_SCRIPT = Path(__file__).parent / "in_image" / "coverage_check.py" +NON_SOURCE = re.compile(r"(^|/)(tests?|docs?)/|CHANGELOG|\.(md|rst|txt)$") +COMPILED = re.compile(r"\.(pyx|pxd|pxi|c|cc|cpp|cxx|h|hpp|f|f90|rs|cu)$") +# Kinds of uncovered task that are measured anyway when asked for. +MEASURE_UNCOVERED_KINDS = ("always_affected", "compiled") + + +def patch_files(solve_sh: Path) -> set[str]: + """Files the oracle patch in solve.sh changes.""" + return set(re.findall(r"^\+\+\+ b/(\S+)", solve_sh.read_text(), re.M)) if solve_sh.exists() else set() + + +def source(files: list[str] | set[str]) -> set[str]: + return {f for f in files if not NON_SOURCE.search(f)} + + +def run_check(tag: str, solve_sh: Path, timeout: int = 1800) -> dict[str, Any]: + script = ( + 'eval "$(micromamba shell hook --shell=bash)"; micromamba activate "$ENV_NAME"; ' + "cd /workspace/repo && python /tmp/check.py" + ) + p = subprocess.run( + [ + "docker", + "run", + "--rm", + "--network", + "none", + "-v", + f"{solve_sh}:/tmp/solve.sh:ro", + "-v", + f"{CHECK_SCRIPT}:/tmp/check.py:ro", + "--entrypoint", + "bash", + tag, + "-lc", + script, + ], + capture_output=True, + text=True, + timeout=timeout, + ) + last = (p.stdout.strip().splitlines() or [""])[-1] + return json.loads(last) if p.returncode == 0 and last.startswith("{") else {"error": p.stderr[-800:]} + + +def decide(result: dict[str, Any], changed: set[str], measure_uncovered: set[str] = frozenset()) -> str | None: + """`base_edited`, `patch_failed`, `uncovered`, or None when the task should be measured. + + base_edited: the image already has edits that touch an oracle hunk, or that make a hunk fail. Results without + `edited_overlap` count any edit in a file the patch changes. A check that errored returns None. + """ + if "error" in result: + return None + edited = source(result.get("edited_before", [])) & changed + if "edited_overlap" in result: + edited = source(result["edited_overlap"]) | (edited & set(result.get("patch_failed", []))) + if edited: + return "base_edited" + if source(result.get("patch_failed", [])): + return "patch_failed" + if result.get("by_deps", 0) > 0: + return None + benchmarks = result.get("benchmarks") + if "always_affected" in measure_uncovered and benchmarks and result.get("always_affected") == benchmarks: + return None + src = source(changed) + if "compiled" in measure_uncovered and src and all(COMPILED.search(f) for f in src): + return None + return "uncovered" + + +def precheck( + survey: Survey, task: str, tag: str, measure_uncovered: set[str] = frozenset(), rerun: bool = False +) -> tuple[str | None, dict[str, Any]]: + """Run (or reuse) the in-image check for one task and return (end state or None, check result).""" + path = survey.root / "coverage" / f"{task}.json" + solve = survey.task_dir(task) / "solution" / "solve.sh" + result = json.loads(path.read_text()) if path.exists() else {"error": "not run"} + if rerun or "error" in result: + result = run_check(tag, solve) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(result)) + return decide(result, patch_files(solve), measure_uncovered), result diff --git a/src/datasmith/verify/survey.py b/src/datasmith/verify/survey.py index 143d9f7a..5fda3fa2 100644 --- a/src/datasmith/verify/survey.py +++ b/src/datasmith/verify/survey.py @@ -54,6 +54,15 @@ def state(self, task: str) -> dict[str, Any]: path = self.root / "state" / f"{task}.json" return json.loads(path.read_text()) if path.exists() else {} + def save_state(self, task: str, state: dict[str, Any]) -> None: + path = self.root / "state" / f"{task}.json" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.tmp") + tmp.write_text( + json.dumps({**state, "task": task, "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}) + ) + tmp.replace(path) + def states(self) -> dict[str, dict[str, Any]]: return {p.stem: json.loads(p.read_text()) for p in sorted((self.root / "state").glob("*.json"))} diff --git a/tests/verify/test_precheck.py b/tests/verify/test_precheck.py new file mode 100644 index 00000000..bf81fa7d --- /dev/null +++ b/tests/verify/test_precheck.py @@ -0,0 +1,71 @@ +from datasmith.verify.in_image import coverage_check as cc +from datasmith.verify.precheck import decide + +PATCH = """cat > /workspace/repo/solution_patch.diff << '__SOLUTION__' +diff --git a/pkg/m.py b/pkg/m.py +--- a/pkg/m.py ++++ b/pkg/m.py +@@ -10,7 +10,7 @@ def f(): + a +-b ++c +diff --git a/pkg/new.py b/pkg/new.py +--- /dev/null ++++ b/pkg/new.py +@@ -0,0 +1,2 @@ ++x +__SOLUTION__ +""" + + +def image_diff(start, n, path="pkg/m.py"): + return f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -{start},{n} +{start},{n} @@\n-x\n+y\n" + + +def test_edit_inside_or_next_to_a_hunk_overlaps(): + assert cc.overlapping(image_diff(12, 1), PATCH) == ["pkg/m.py"] + assert cc.overlapping(image_diff(16, 3), PATCH) == ["pkg/m.py"] + assert cc.overlapping(image_diff(9, 0), PATCH) == ["pkg/m.py"] + + +def test_edit_elsewhere_or_in_another_file_does_not_overlap(): + assert cc.overlapping(image_diff(200, 4), PATCH) == [] + assert cc.overlapping(image_diff(3, 6), PATCH) == [] + assert cc.overlapping(image_diff(12, 1, "pkg/other.py"), PATCH) == [] + + +CHANGED = {"pkg/m.py", "tests/test_m.py"} + + +def result(**kw): + return { + "benchmarks": 10, + "always_affected": 0, + "by_deps": 3, + "patch_failed": [], + "edited_before": [], + "edited_overlap": [], + **kw, + } + + +def test_decide_end_states(): + assert decide(result(), CHANGED) is None + assert decide(result(edited_overlap=["pkg/m.py"]), CHANGED) == "base_edited" + assert decide(result(edited_before=["pkg/m.py"]), CHANGED) is None + assert decide(result(patch_failed=["pkg/m.py"]), CHANGED) == "patch_failed" + assert decide(result(patch_failed=["tests/test_m.py"]), CHANGED) is None + assert decide(result(by_deps=0), CHANGED) == "uncovered" + assert decide({"error": "boom"}, CHANGED) is None + + +def test_decide_without_overlap_field_counts_any_edit_in_a_changed_file(): + r = result(edited_before=["pkg/m.py"]) + del r["edited_overlap"] + assert decide(r, CHANGED) == "base_edited" + + +def test_uncovered_kinds_measured_on_request(): + assert decide(result(by_deps=0, always_affected=10), CHANGED, {"always_affected"}) is None + assert decide(result(by_deps=0), {"pkg/_fast.pyx"}, {"compiled"}) is None + assert decide(result(by_deps=0), CHANGED, {"compiled"}) == "uncovered" From 61f41a8a8e9c8a9da513cd616121ee9de73aa197 Mon Sep 17 00:00:00 2001 From: Arjun Sharma Date: Mon, 19 Oct 2026 03:53:08 +0000 Subject: [PATCH 03/32] verify: build, pin, scrub and publish stages for task images --- src/datasmith/verify/cli.py | 81 ++++++- src/datasmith/verify/images.py | 383 +++++++++++++++++++++++++++++++ src/datasmith/verify/pin.py | 63 +++++ src/datasmith/verify/registry.py | 168 ++++++++++++++ src/datasmith/verify/scrub.py | 142 ++++++++++++ tests/verify/test_images.py | 146 ++++++++++++ 6 files changed, 978 insertions(+), 5 deletions(-) create mode 100644 src/datasmith/verify/images.py create mode 100644 src/datasmith/verify/pin.py create mode 100644 src/datasmith/verify/registry.py create mode 100644 src/datasmith/verify/scrub.py create mode 100644 tests/verify/test_images.py diff --git a/src/datasmith/verify/cli.py b/src/datasmith/verify/cli.py index d7a73e73..4a27fd28 100644 --- a/src/datasmith/verify/cli.py +++ b/src/datasmith/verify/cli.py @@ -7,6 +7,9 @@ fc-verify auto SURVEY [--dry-run] automatic not_approved rules fc-verify add SURVEY TASK ... pending rows for new tasks fc-verify precheck SURVEY TASK ... [--measure-uncovered always_affected,compiled] [--rerun] +fc-verify build TASKS_ROOT OUT [--ids FILE ...] [--workers N] [--retry-failed] +fc-verify publish TASKS_ROOT MANIFEST OUT [--region R] [--repo NAME] [--until-idle] +fc-verify pins TASKS_ROOT MANIFEST pin status of every task folder fc-verify failure LOG ... error class of failed image builds """ @@ -14,10 +17,11 @@ import argparse import json +import re import sys from pathlib import Path -from datasmith.verify import precheck, report, review, status +from datasmith.verify import images, pin, precheck, registry, report, review, status from datasmith.verify.noise import classify_all, load_rule from datasmith.verify.survey import DECISIONS, Survey, task_id @@ -81,6 +85,56 @@ def _precheck(survey: Survey, a: argparse.Namespace) -> None: } +def _ids(files: list[Path], root: Path) -> list[str]: + out = [] + for f in files: + for line in f.read_text().splitlines(): + if m := re.search(r"([\w.-]+)/([\w.-]+)#(\d+)", line): + out.append("__".join(m.groups())) + elif re.fullmatch(r"[\w.-]+__[\w.-]+__\d+", line.strip()): + out.append(line.strip()) + return out or sorted(p.name for p in root.iterdir() if (p / "environment").is_dir()) + + +def _build(a: argparse.Namespace) -> int: + images.build( + a.tasks_root, + _ids(a.ids, a.tasks_root), + a.out, + a.workers, + images.Limits(keep_failed=a.keep_failed), + retry_failed=a.retry_failed, + ) + return 0 + + +def _publish(a: argparse.Namespace) -> int: + def building() -> bool: + return images.sh(["pgrep", "-f", "[f]c-verify build"]).returncode == 0 + + registry.publish(a.tasks_root, a.manifest, a.out, registry.EcrRegistry(a.region, a.repo), a.until_idle, building) + return 0 + + +def _pins(a: argparse.Namespace) -> int: + rows, bad = images.rows_by_task(a.manifest), 0 + for task_dir in sorted(p for p in a.tasks_root.iterdir() if (p / "task.toml").is_file()): + state, tag = pin.plan(task_dir, rows.get(task_dir.name)) + if state not in ("ok", "unbuilt"): + bad += state in ("drift", "env-mismatch", "unverified-pinned") + print(f"{task_dir.name}: {state}" + (f" -> {tag}" if tag else "")) + return 1 if bad else 0 + + +def _failure(a: argparse.Namespace) -> int: + for log in a.logs: + print(json.dumps({"log": str(log), **report.failure_reason(log)})) + return 0 + + +NO_SURVEY = {"build": _build, "publish": _publish, "pins": _pins, "failure": _failure} + + def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( prog="fc-verify", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter @@ -101,12 +155,29 @@ def main(argv: list[str] | None = None) -> int: ) sub.choices["precheck"].add_argument("--rerun", action="store_true") sub.add_parser("failure").add_argument("logs", nargs="+", type=Path) + b = sub.add_parser("build") + b.add_argument("tasks_root", type=Path) + b.add_argument("out", type=Path) + b.add_argument( + "--ids", type=Path, action="append", default=[], help="files of owner/repo#N or owner__repo__N lines" + ) + b.add_argument("--workers", type=int, default=1) + b.add_argument("--retry-failed", action="store_true") + b.add_argument("--keep-failed", action="store_true") + pb = sub.add_parser("publish") + pb.add_argument("tasks_root", type=Path) + pb.add_argument("manifest", type=Path) + pb.add_argument("out", type=Path) + pb.add_argument("--region", default="us-east-1") + pb.add_argument("--repo", default="formulacode") + pb.add_argument("--until-idle", action="store_true", help="keep publishing while a build runs") + pp = sub.add_parser("pins") + pp.add_argument("tasks_root", type=Path) + pp.add_argument("manifest", type=Path) a = ap.parse_args(argv) + if a.cmd in NO_SURVEY: + return NO_SURVEY[a.cmd](a) - if a.cmd == "failure": - for log in a.logs: - print(json.dumps({"log": str(log), **report.failure_reason(log)})) - return 0 COMMANDS[a.cmd](Survey(a.survey), a) return 0 diff --git a/src/datasmith/verify/images.py b/src/datasmith/verify/images.py new file mode 100644 index 00000000..9cd41400 --- /dev/null +++ b/src/datasmith/verify/images.py @@ -0,0 +1,383 @@ +"""Build, check, save and record one image per rendered task. + +Each built image gets a manifest row (`manifest.jsonl`, newest row per task wins); a task that fails gets a row in +`failures.jsonl`. A build is stopped when it prints nothing for `FC_BUILD_IDLE_S` seconds (default 4 h) or when the +root disk runs low; a disk stop is not a task failure. Several workers share one task list; a task is claimed by +creating `queue/claimed/`. +""" + +from __future__ import annotations + +import functools +import hashlib +import json +import os +import re +import shutil +import subprocess +import threading +import time +from collections.abc import Iterable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +GIB = 1024**3 +QWEN_CODE_VERSION = "0.22.3" +VERIFY_SH = r""" +set -u +export NVM_DIR=/root/.nvm; set +u; . "$NVM_DIR/nvm.sh" >/dev/null; set -u +echo "QWEN=$(qwen --version)" +set +u; . /etc/profile.d/asv_build_vars.sh; set -u +eval "$(micromamba shell hook --shell=bash)" +micromamba run -n "$ENV_NAME" python -c "import asv_runner, coverage, jinja2, asv.contrib.lightspeed" && echo IMPORTS=ok +micromamba run -n "$ENV_NAME" snapshot-tool --help >/dev/null 2>&1 && echo SNAPSHOT_TOOL=ok +echo "HEAD=$(git -C /workspace/repo rev-parse HEAD 2>/dev/null)" +echo "LSV_INIT=$(cat /opt/lsv/cache/lsv_init_results.json 2>/dev/null | tr -d '\n')" +echo "ADDED_DEPS=$(cat /opt/lsv/cache/added_deps.json 2>/dev/null | tr -d '\n')" +""" +LOW_GRACE_S, HARD_FLOOR_GB = 180, 3.0 + + +@dataclass +class Limits: + min_free_gb: float = 12.0 + need_free_gb: float = 24.0 + out_min_free_gb: float = 50.0 + keep_if_free_gb: float = 60.0 + idle_s: int = int(os.environ.get("FC_BUILD_IDLE_S", "14400")) + keep_failed: bool = False + + +def sh(cmd: list[str], **kw: Any) -> subprocess.CompletedProcess: + return subprocess.run(cmd, text=True, capture_output=True, **kw) + + +def read_jsonl(path: Path) -> list[dict]: + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + + +_append_lock = threading.Lock() + + +def append_jsonl(path: Path, row: dict) -> None: + with _append_lock, path.open("a") as f: + f.write(json.dumps(row, sort_keys=True) + "\n") + + +def rows_by_task(manifest: Path) -> dict[str, dict]: + """A rebuild appends a row; the last row of a task wins.""" + return {r["task_id"]: r for r in read_jsonl(manifest)} + + +def image_exists(ref: str) -> bool: + return sh(["docker", "image", "inspect", ref]).returncode == 0 + + +def root_free_gb() -> float: + return shutil.disk_usage("/").free / GIB + + +@functools.lru_cache(maxsize=1) +def docker_root() -> str: + """Docker's data dir; / when the daemon does not answer.""" + try: + out = sh(["docker", "info", "-f", "{{.DockerRootDir}}"], timeout=20).stdout.strip() + except (OSError, subprocess.TimeoutExpired): + out = "" + return out if out and os.path.isdir(out) else "/" + + +def docker_free() -> int: + return shutil.disk_usage(docker_root()).free + + +def env_digest(env_dir: Path) -> str: + """sha256 over the build context's relative paths and bytes; the image tag carries its first 12 hex digits.""" + h = hashlib.sha256() + for p in sorted(q for q in env_dir.rglob("*") if q.is_file()): + h.update(str(p.relative_to(env_dir)).encode() + b"\0" + p.read_bytes() + b"\0") + return h.hexdigest() + + +def image_tag(task_dir: Path) -> str: + return f"fc-task/{task_dir.name.lower()}:{env_digest(task_dir / 'environment')[:12]}" + + +def run_guarded(cmd: list[str], log_path: Path, limits: Limits, env: dict | None = None) -> tuple[int, str]: + """Run cmd with its output in log_path. Kill it when the log stops growing for idle_s, when root free space stays + below min_free_gb for LOW_GRACE_S (short spikes from other users recover), or below HARD_FLOOR_GB at once.""" + with log_path.open("a") as lf: + lf.write(f"$ {' '.join(cmd)}\n") + lf.flush() + proc = subprocess.Popen(cmd, stdout=lf, stderr=subprocess.STDOUT, env=env) + reason, low_since, size, active = "", None, -1, time.monotonic() + while proc.poll() is None: + if log_path.stat().st_size != size: + size, active = log_path.stat().st_size, time.monotonic() + elif time.monotonic() - active >= limits.idle_s: + reason = f"no output for {limits.idle_s}s; killed" + proc.kill() + proc.wait() + break + free = root_free_gb() + low_since = (low_since or time.monotonic()) if free < limits.min_free_gb else None + if free < HARD_FLOOR_GB or (low_since is not None and time.monotonic() - low_since >= LOW_GRACE_S): + reason = f"root free {free:.1f}G < {limits.min_free_gb}G; killed" + proc.kill() + proc.wait() + break + time.sleep(10) + return proc.returncode, reason + + +def tail(path: Path, n: int = 40) -> str: + return "\n".join(path.read_text(errors="replace").splitlines()[-n:]) + + +def _json_or(text: str | None, default: Any) -> Any: + try: + return json.loads(text or "null") or default + except json.JSONDecodeError: + return default + + +def check_image(tag: str) -> tuple[dict, list[str]]: + """Facts about a built image and the problems that keep it from being measured.""" + r = sh(["docker", "run", "--rm", "--network", "none", "--entrypoint", "bash", tag, "-c", VERIFY_SH], timeout=600) + out = dict(line.split("=", 1) for line in r.stdout.splitlines() if "=" in line and line.split("=", 1)[0].isupper()) + problems = [] + if out.get("QWEN", "").strip() != QWEN_CODE_VERSION: + problems.append(f"qwen-code version {out.get('QWEN')!r} != {QWEN_CODE_VERSION}") + if out.get("IMPORTS") != "ok": + problems.append("import asv_runner/coverage/jinja2/asv.contrib.lightspeed failed") + if out.get("SNAPSHOT_TOOL") != "ok": + problems.append("snapshot-tool missing") + lsv = _json_or(out.get("LSV_INIT"), {}) + info = { + "head": out.get("HEAD", ""), + "env_fingerprint": lsv.get("env_fingerprint"), + "baseline_sha": lsv.get("baseline_sha"), + "bake_cpus_bounded": lsv.get("bake_cpus_bounded"), + "benchmarks_discovered": len(lsv.get("benchmarks_discovered") or []), + "added_deps": _json_or(out.get("ADDED_DEPS"), None), + } + if not lsv.get("baseline_sha"): + problems.append("no stored baseline in /opt/lsv/cache/lsv_init_results.json") + elif lsv["baseline_sha"] != info["head"]: + problems.append(f"baseline_sha {lsv['baseline_sha']} != HEAD {info['head']}") + if not info["env_fingerprint"]: + problems.append("no env_fingerprint in the stored baseline") + if not r.stdout.strip(): + problems.append(f"verify container failed: {r.stderr.strip()[-500:]}") + return info, problems + + +def save_and_check(tag: str, tar: Path) -> tuple[int, str]: + """docker save | zstd into tar, check that it lists, and return (bytes, sha256).""" + tmp = tar.with_suffix(tar.suffix + ".partial") + r = sh(["bash", "-c", f"set -o pipefail; docker save {tag} | zstd -T8 -3 -q -f -o {tmp}"]) + if r.returncode != 0: + tmp.unlink(missing_ok=True) + raise RuntimeError(f"docker save failed: {r.stderr[-500:]}") + listing = sh(["bash", "-c", f"set -o pipefail; zstd -dc {tmp} | tar -tf -"]) + if listing.returncode != 0 or "manifest.json" not in listing.stdout.split(): + tmp.unlink(missing_ok=True) + raise RuntimeError(f"saved tar failed to list: {listing.stderr[-500:]}") + tmp.rename(tar) + h = hashlib.sha256() + with tar.open("rb") as f: + while chunk := f.read(1 << 24): + h.update(chunk) + return tar.stat().st_size, h.hexdigest() + + +def build_one(task_dir: Path, out: Path, limits: Limits, preexisting: set[str]) -> tuple[dict | None, dict | None]: + """(manifest row or None, failure row or None) for one task.""" + task_id = task_dir.name + stamp = json.loads((task_dir / "render_stamp.json").read_text()) + base = stamp["params"]["base_image"] + digest = env_digest(task_dir / "environment") + tag = f"fc-task/{task_id.lower()}:{digest[:12]}" + tar = out / f"{tag.replace('/', '_').replace(':', '__')}.tar.zst" + blog = out / "logs" / f"{task_id}.log" + blog.parent.mkdir(parents=True, exist_ok=True) + + def fail(stage: str, err: str, disk: bool = False) -> tuple[None, dict]: + return None, { + "task_id": task_id, + "tag": tag, + "stage": stage, + "error": err, + "disk": disk, + "log": str(blog), + "at": time.strftime("%FT%T"), + } + + pulled_base = False + if not image_exists(base): + rc, why = run_guarded(["docker", "pull", base], blog, limits) + if rc != 0: + return fail("pull", why or tail(blog), disk=why.startswith("root free")) + pulled_base = True + log_offset = blog.stat().st_size if blog.exists() else 0 + t0 = time.time() + rc, why = run_guarded( + ["docker", "build", "--rm", "--force-rm", "--network", "host", "-t", tag, str(task_dir / "environment")], + blog, + limits, + env=dict(os.environ, DOCKER_BUILDKIT="0"), + ) + build_s = round(time.time() - t0, 1) + try: + if rc != 0: + # The legacy builder keeps a failed build's last intermediate image; drop it unless it is the base. + with blog.open() as f: + f.seek(log_offset) + layers = re.findall(r"^ ---> ([0-9a-f]{12})$", f.read(), re.M) + base_id = sh(["docker", "image", "inspect", "-f", "{{.Id}}", base]).stdout.strip() + if layers and not base_id.startswith("sha256:" + layers[-1]): + sh(["docker", "rmi", layers[-1]]) + return fail("build", why or tail(blog), disk=why.startswith("root free")) + info, problems = check_image(tag) + size, sha = save_and_check(tag, tar) + image_id = sh(["docker", "image", "inspect", "-f", "{{.Id}}", tag]).stdout.strip() + img_size = int(sh(["docker", "image", "inspect", "-f", "{{.Size}}", tag]).stdout.strip() or 0) + row = { + "task_id": task_id, + "tag": tag, + "image_id": image_id, + "image_bytes": img_size, + "base_image": base, + "tar_path": str(tar), + "tar_bytes": size, + "tar_sha256": sha, + "bake_seconds": build_s, + "verified": not problems, + "problems": problems, + "render_source_sha256": stamp["render_source_sha256"], + "datasmith_commit": stamp["datasmith"]["commit"], + "env_digest": digest, + "built_at": time.strftime("%FT%T"), + **info, + } + return row, (fail("verify", "; ".join(problems))[1] if problems else None) + except Exception as e: + return fail("save", str(e)) + finally: + if image_exists(tag) and (root_free_gb() < limits.keep_if_free_gb or rc != 0 or not tar.exists()): + sh(["docker", "rmi", tag]) + if pulled_base and base not in preexisting and image_exists(base) and root_free_gb() < limits.keep_if_free_gb: + sh(["docker", "rmi", base]) + + +def _say(msg: str) -> None: + print(f"[build {time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) + + +def _blocked(out: Path, limits: Limits) -> str | None: + if (out / "STOP").exists(): + return "STOP file present" + if root_free_gb() < limits.need_free_gb: + return f"root free {root_free_gb():.1f}G < {limits.need_free_gb}G" + if shutil.disk_usage(out).free / GIB < limits.out_min_free_gb: + return f"{out} free below {limits.out_min_free_gb}G" + return None + + +def _build_claimed(task_dir: Path, out: Path, limits: Limits, preexisting: set[str]) -> str: + """Build one claimed task and record the result: `ok`, `failed` or `disk`.""" + task_id = task_dir.name + if not (task_dir / "environment").is_dir(): + append_jsonl( + out / "failures.jsonl", + {"task_id": task_id, "stage": "render", "error": f"no task dir {task_dir}", "at": time.strftime("%FT%T")}, + ) + return "failed" + _say(f"{task_id}: start") + row, err = build_one(task_dir, out, limits, preexisting) + if row: + # An image that fails its check is never measured; its archive only fills the disk. + if not row["verified"] and not limits.keep_failed: + Path(row["tar_path"]).unlink(missing_ok=True) + append_jsonl(out / "manifest.jsonl", row) + _say(f"{task_id}: saved {row['tar_bytes'] / GIB:.2f}G in {row['bake_seconds']}s verified={row['verified']}") + if err and err.get("disk"): + _say(f"{task_id}: stopped for disk at {err['stage']}: {err['error']}") + return "disk" + if err: + append_jsonl(out / "failures.jsonl", err) + _say(f"{task_id}: FAILED at {err['stage']}: {err['error'][-300:]}") + return "failed" + return "ok" + + +def build( + tasks_root: Path, + task_ids: Iterable[str], + out: Path, + workers: int = 1, + limits: Limits | None = None, + retry_failed: bool = False, +) -> None: + """Build every task not yet in the manifest (or failures, unless retry_failed), `workers` at a time. + + Stops starting new tasks when `out/STOP` exists or a disk limit is reached. + """ + limits = limits or Limits() + out.mkdir(parents=True, exist_ok=True) + claimed = out / "queue" / "claimed" + claimed.mkdir(parents=True, exist_ok=True) + manifest, failures = out / "manifest.jsonl", out / "failures.jsonl" + done = {r["task_id"] for r in read_jsonl(manifest)} + if not retry_failed: + done |= {r["task_id"] for r in read_jsonl(failures)} + todo = [t for t in dict.fromkeys(task_ids) if t not in done] + preexisting = set(sh(["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"]).stdout.split()) + stop = threading.Event() + _say(f"{len(todo)} task(s) to build with {workers} worker(s); root free {root_free_gb():.1f}G") + + def worker(task_id: str) -> None: + if stop.is_set(): + return + if why := _blocked(out, limits): + _say(f"{why}; stopping") + stop.set() + return + try: + (claimed / task_id).mkdir() + except FileExistsError: + return + if _build_claimed(tasks_root / task_id, out, limits, preexisting) == "disk": + # Not a task failure: release the claim so a later run builds it. + (claimed / task_id).rmdir() + stop.set() + + with ThreadPoolExecutor(max(1, workers)) as pool: + list(pool.map(worker, todo)) + + +def load(row: dict, ecr_repo: str | None = None, min_free_gb: float = 12.0) -> str: + """Make the image of a manifest row present on the daemon (from its archive, else from the registry) and check its id.""" + tag, want = row["tag"], row["image_id"] + have = sh(["docker", "image", "inspect", "-f", "{{.Id}}", tag]).stdout.strip() + if not have: + if docker_free() < row.get("image_bytes", 0) + min_free_gb * GIB: + raise RuntimeError(f"{row['task_id']}: Docker disk too full to load {tag}") + tar = row.get("tar_path") + if tar and Path(tar).exists(): + r = sh(["docker", "load", "-i", tar]) + elif ecr_repo: + uri = f"{ecr_repo}:{tag.removeprefix('fc-task/').replace(':', '__')}" + r = sh(["bash", "-c", f"docker pull -q {uri} && docker tag {uri} {tag} && docker rmi {uri}"]) + else: + raise RuntimeError(f"{row['task_id']}: {tag} is not loaded and has no archive or registry") + if r.returncode: + raise RuntimeError(f"{row['task_id']}: load failed: {r.stderr.strip()[-300:]}") + have = sh(["docker", "image", "inspect", "-f", "{{.Id}}", tag]).stdout.strip() + if have != want: + raise RuntimeError(f"{row['task_id']}: {tag} is {have or 'missing'}, manifest image_id is {want}") + return tag diff --git a/src/datasmith/verify/pin.py b/src/datasmith/verify/pin.py new file mode 100644 index 00000000..612873bc --- /dev/null +++ b/src/datasmith/verify/pin.py @@ -0,0 +1,63 @@ +"""Pin a built image into a task's task.toml (`[environment] docker_image`). + +A row is pinned only when the task's environment/ still has the digest the image was built from. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from datasmith.verify.images import env_digest + +SECTION = re.compile(r"^\s*\[environment\]\s*$") +PIN = re.compile(r"""^\s*docker_image\s*=\s*["']([^"']*)["']""") + + +def _env_span(lines: list[str]) -> tuple[int, int] | None: + start = next((i for i, line in enumerate(lines) if SECTION.match(line)), None) + if start is None: + return None + end = next((i for i in range(start + 1, len(lines)) if lines[i].lstrip().startswith("[")), len(lines)) + return start, end + + +def read_pin(toml_text: str) -> str | None: + lines = toml_text.splitlines() + start, end = _env_span(lines) or (0, 0) + return next((m[1] for line in lines[start + 1 : end] if (m := PIN.match(line))), None) + + +def set_pin(toml_text: str, tag: str) -> str: + lines, new = toml_text.splitlines(), f'docker_image = "{tag}"' + span = _env_span(lines) + if span is None: + return toml_text.rstrip("\n") + f"\n\n[environment]\n{new}\n" + i = next((i for i in range(span[0] + 1, span[1]) if PIN.match(lines[i])), None) + if i is None: + lines.insert(span[0] + 1, new) + else: + lines[i] = new + return "\n".join(lines) + "\n" + + +def plan(task_dir: Path, row: dict | None) -> tuple[str, str | None]: + """(status, tag to pin or None) for one task and its newest manifest row.""" + pin = read_pin((task_dir / "task.toml").read_text()) + if row is None: + return ("unbuilt" if pin is None else "foreign-pin"), None + if not row.get("verified"): + return ("unverified-pinned" if pin == row["tag"] else "unverified"), None + if env_digest(task_dir / "environment") != row.get("env_digest"): + return "env-mismatch", None + if pin == row["tag"]: + return "ok", None + return ("unpinned" if pin is None else "drift"), row["tag"] + + +def write_pin(task_dir: Path, tag: str) -> None: + toml = task_dir / "task.toml" + tmp = toml.with_suffix(".toml.tmp") + tmp.write_text(set_pin(toml.read_text(), tag)) + # Rename, so a hard-linked copy never writes through to the render. + tmp.replace(toml) diff --git a/src/datasmith/verify/registry.py b/src/datasmith/verify/registry.py new file mode 100644 index 00000000..4f6d37e9 --- /dev/null +++ b/src/datasmith/verify/registry.py @@ -0,0 +1,168 @@ +"""Publish built images: pin, remove git history, check for leaks, push to a registry, and record each step. + +`published.jsonl` holds the -scrub rows (no archive); `pushed.jsonl` holds the registry records. A rerun skips +tasks already published. +""" + +from __future__ import annotations + +import json +import time +from collections.abc import Callable +from pathlib import Path +from typing import Protocol + +from datasmith.verify import images, pin, scrub +from datasmith.verify.images import append_jsonl, read_jsonl, sh + +LOGIN_EVERY_S = 6 * 3600 + + +class Registry(Protocol): + def push(self, task_id: str, tag: str) -> dict: ... + + +class EcrRegistry: + """One ECR repository; an image `fc-task/:` is pushed as `:__`.""" + + def __init__(self, region: str = "us-east-1", repo: str = "formulacode") -> None: + self.region, self.repo = region, repo + account = sh(["aws", "sts", "get-caller-identity", "--query", "Account", "--output", "text"], check=True) + self.registry = f"{account.stdout.strip()}.dkr.ecr.{region}.amazonaws.com" + self.uri = f"{self.registry}/{repo}" + self.logged_in = 0.0 + + def login(self) -> None: + if time.time() - self.logged_in < LOGIN_EVERY_S: + return + password = sh(["aws", "ecr", "get-login-password", "--region", self.region], check=True).stdout + sh(["docker", "login", "--username", "AWS", "--password-stdin", self.registry], input=password, check=True) + self.logged_in = time.time() + + def ensure_repo(self) -> None: + have = sh(["aws", "ecr", "describe-repositories", "--region", self.region, "--repository-names", self.repo]) + if have.returncode: + sh( + [ + "aws", + "ecr", + "create-repository", + "--region", + self.region, + "--repository-name", + self.repo, + "--tags", + "Key=project,Value=formulacode", + ], + check=True, + ) + + def push(self, task_id: str, tag: str) -> dict: + name, ident = tag.split(":", 1) + ecr_tag = f"{name.split('/')[-1]}__{ident}" + uri = f"{self.uri}:{ecr_tag}" + self.login() + self.ensure_repo() + try: + sh(["docker", "tag", tag, uri], check=True) + sh(["docker", "push", "--quiet", uri], check=True) + digest = sh( + [ + "aws", + "ecr", + "describe-images", + "--region", + self.region, + "--repository-name", + self.repo, + "--image-ids", + f"imageTag={ecr_tag}", + "--query", + "imageDetails[0].imageDigest", + "--output", + "text", + ], + check=True, + ).stdout.strip() + finally: + sh(["docker", "rmi", uri]) + return { + "task_id": task_id, + "tag": tag, + "uri": uri, + "digest": digest, + "pushed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + + +def gt_hash(task_dir: Path) -> str: + cfg = task_dir / "tests" / "config.json" + return json.loads(cfg.read_text()).get("gt_hash", "") if cfg.exists() else "" + + +def publish_one(task_dir: Path, row: dict, registry: Registry, out: Path) -> str: + """Pin the built image, load it, scrub it, probe the result, push it; returns the pushed tag.""" + task = task_dir.name + status, tag = pin.plan(task_dir, row) + if status not in ("ok", "unpinned", "drift"): + raise RuntimeError(f"{task}: cannot pin ({status})") + if tag: + pin.write_pin(task_dir, tag) + images.load(row) + new, res = scrub.scrub(row["tag"]) + if not new: + raise RuntimeError(f"{task}: {res.get('error')}") + check = scrub.probe(task, new, gt_hash(task_dir)) + if check.get("status") != "clean": + sh(["docker", "rmi", new]) + raise RuntimeError(f"{task}: scrubbed image still leaks: {check.get('leaks') or check.get('error')}") + info = json.loads(sh(["docker", "image", "inspect", new]).stdout)[0] + scrubbed = { + **row, + "tag": new, + "image_id": info["Id"], + "image_bytes": info["Size"], + "scrubbed_from": row["tag"], + "scrubbed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "tar_path": None, + "scrub_check": { + **{k: check.get(k) for k in ("unreachable_newer", "refs_ahead", "pr_commits", "remotes")}, + "nested_git_removed": res["nested_git_removed"], + "nested_tree_same": res["nested_tree_same"], + }, + } + for k in ("tar_bytes", "tar_sha256"): + scrubbed.pop(k, None) + append_jsonl(out / "pushed.jsonl", registry.push(task, new)) + pin.write_pin(task_dir, new) + append_jsonl(out / "published.jsonl", scrubbed) + for t in (new, row["tag"]): + sh(["docker", "rmi", t]) + return new + + +def publish( + tasks_root: Path, + manifest: Path, + out: Path, + registry: Registry, + until_idle: bool = False, + building: Callable[[], bool] = lambda: False, +) -> None: + """Publish every verified, unpublished manifest row; with until_idle, keep going until `building()` is False.""" + out.mkdir(parents=True, exist_ok=True) + tried: set[str] = set() + while True: + done = {r["task_id"] for r in read_jsonl(out / "published.jsonl")} + todo = [t for t, r in images.rows_by_task(manifest).items() if r.get("verified") and t not in done | tried] + for task in todo: + tried.add(task) + start = time.time() + try: + result = publish_one(tasks_root / task, images.rows_by_task(manifest)[task], registry, out) + except Exception as error: + result = f"error: {type(error).__name__}: {error}" + print(f"[{time.strftime('%H:%M:%S')}] {task}: {result} ({time.time() - start:.0f} s)", flush=True) + if not until_idle or (not building() and not todo): + return + time.sleep(60) diff --git a/src/datasmith/verify/scrub.py b/src/datasmith/verify/scrub.py new file mode 100644 index 00000000..bc98bcfd --- /dev/null +++ b/src/datasmith/verify/scrub.py @@ -0,0 +1,142 @@ +"""Remove git history after HEAD from a task image, and check an image for history that exposes the fix. + +`scrub` writes a new tag `-scrub`; the source image is not changed. It commits only when the container's +changes are under .git dirs and HEAD, tree, `git describe` and status are unchanged. The origin URL stays because +lsv_init.py and the test scripts read it. Nested .git dirs (benchmark repos cloned at their latest commit) are +removed when their files hash the same afterwards. +""" + +from __future__ import annotations + +import json +import re +import uuid + +from datasmith.verify.images import sh + +SCRUB = r""" +set -eu +cd "$(git rev-parse --show-toplevel)" +state() { printf '%s %s %s %s' "$(git rev-parse HEAD)" "$(git rev-parse 'HEAD^{tree}')" \ + "$(git describe --tags --always 2>/dev/null || true)" "$(git status --porcelain --untracked-files=no | sha256sum | cut -c1-16)"; } +before=$(state); head=$(git rev-parse HEAD); abbrev=$(git rev-parse --short HEAD | wc -c) +git for-each-ref --format='%(refname) %(objectname)' | while read -r ref obj; do + c=$(git rev-parse -q --verify "$obj^{commit}" 2>/dev/null) && git merge-base --is-ancestor "$c" "$head" || git update-ref -d "$ref" +done +git stash clear 2>/dev/null || true +rm -rf .git/logs .git/refs/original .git/FETCH_HEAD .git/ORIG_HEAD .git/objects/info/alternates +git config --unset-all gc.pruneExpire 2>/dev/null || true +git config --unset-all gc.worktreePruneExpire 2>/dev/null || true +git config gc.auto 0 +git reflog expire --expire=now --expire-unreachable=now --all +git -c gc.pruneExpire=now gc --prune=now --quiet +git config core.abbrev $((abbrev-1)) +after=$(state) +left=$(git fsck --unreachable --no-reflogs --no-progress 2>/dev/null | grep -c '^unreachable commit' || true) +treehash() { (cd "$1" && find . -path ./.git -prune -o -type f -print0 | sort -z | xargs -0 -r sha256sum | sha256sum | cut -c1-16); } +nested=""; same=true +for g in $(find /workspace /tmp /opt -maxdepth 4 -name .git -type d -not -path "$(pwd)/.git" 2>/dev/null); do + d=$(dirname "$g"); h=$(treehash "$d"); rm -rf "$g"; [ "$h" = "$(treehash "$d")" ] || same=false; nested="$nested$g " +done +printf '{"repo":"%s","before":"%s","after":"%s","unreachable_left":%s,"nested_git_removed":"%s","nested_tree_same":%s}\n' \ + "$(pwd)" "$before" "$after" "$left" "${nested% }" "$same" +""" + +PROBE = r""" +cd "$(git rev-parse --show-toplevel 2>/dev/null)" || { echo '{"error":"no git repo"}'; exit 0; } +head=$(git rev-parse HEAD); head_ct=$(git log -1 --format=%ct HEAD) +remotes=$(git remote | tr '\n' ' ') +refs=0; ahead=0 +for r in $(git for-each-ref --format='%(objectname)'); do + refs=$((refs+1)) + c=$(git rev-parse -q --verify "$r^{commit}") || continue + [ "$c" != "$head" ] && git merge-base --is-ancestor "$head" "$c" && ahead=$((ahead+1)) +done +unreach=$(git fsck --unreachable --no-reflogs --no-progress 2>/dev/null | awk '/^unreachable commit/{print $3}') +n_unreach=$(printf '%s\n' "$unreach" | grep -c . || true) +n_newer=$(printf '%s\n' "$unreach" | grep . | git log --no-walk --stdin --format=%ct 2>/dev/null | awk -v h="$head_ct" '$1>h' | wc -l) +gt=missing; [ -n "$GT" ] && git cat-file -e "$GT^{commit}" 2>/dev/null && gt=present +pr_commits=0; [ -n "$PR" ] && pr_commits=$({ printf '%s\n' "$unreach"; git rev-list --all --not HEAD; } | grep . | sort -u | git log --no-walk --stdin --format=%s 2>/dev/null | grep -cE "#$PR([^0-9]|$)" || true) +extra=$(find /workspace /tmp /opt -maxdepth 4 -name .git -not -path "$(pwd)/.git" 2>/dev/null | while read -r g; do + printf '%s@%s;' "$g" "$(git --git-dir="$g" log -1 --format=%cI 2>/dev/null)"; done) +printf '{"head":"%s","head_ct":%s,"remotes":"%s","refs":%s,"refs_ahead":%s,"unreachable_commits":%s,"unreachable_newer":%s,"gt_commit":"%s","pr_commits":%s,"other_git_dirs":"%s"}\n' \ + "$head" "$head_ct" "$remotes" "$refs" "$ahead" "$n_unreach" "$n_newer" "$gt" "$pr_commits" "$extra" +""" + + +def pr_number(task_id: str) -> str: + m = re.search(r"__(\d+)$", task_id) + return m[1] if m else "" + + +def probe(task_id: str, image: str, gt_hash: str, timeout: int = 900) -> dict: + """status `clean` or `leak` (unreachable commits newer than HEAD, refs ahead of HEAD, the fix commit, PR commits).""" + cmd = [ + "docker", + "run", + "--rm", + "--network", + "none", + "-e", + f"GT={gt_hash}", + "-e", + f"PR={pr_number(task_id)}", + "--entrypoint", + "bash", + image, + "-c", + PROBE, + ] + try: + res = json.loads(sh(cmd, timeout=timeout).stdout.strip().splitlines()[-1]) + except Exception as e: + return {"status": "error", "error": repr(e)[:200]} + if "error" in res: + return {"status": "error", **res} + leaks = [ + k + for k, bad in [ + ("unreachable_newer", res["unreachable_newer"] > 0), + ("refs_ahead", res["refs_ahead"] > 0), + ("gt_commit", res["gt_commit"] == "present"), + ("pr_commit", res["pr_commits"] > 0), + ] + if bad + ] + return {"status": "leak" if leaks else "clean", **res, "leaks": " ".join(leaks)} + + +def scrub(image: str) -> tuple[str | None, dict]: + """(new tag or None, scrub report).""" + name = f"fc-scrub-{uuid.uuid4().hex[:10]}" + cfg = json.loads(sh(["docker", "image", "inspect", image]).stdout)[0]["Config"] + try: + p = sh( + ["docker", "run", "--name", name, "--network", "none", "--entrypoint", "bash", image, "-c", SCRUB], + timeout=3600, + ) + if p.returncode: + return None, {"error": f"scrub failed: {p.stderr[-400:]}"} + res = json.loads(p.stdout.strip().splitlines()[-1]) + if res["before"] != res["after"] or res["unreachable_left"] or not res["nested_tree_same"]: + return None, {"error": "state changed or unreachable commits left", **res} + git_dir = res["repo"].rstrip("/") + "/.git" + changed = [line.split(" ", 1)[1] for line in sh(["docker", "diff", name]).stdout.splitlines()] + removed = [git_dir, *res["nested_git_removed"].split()] + roots = {"/workspace", "/workspace/repo", "/tmp", "/opt", res["repo"]} # noqa: S108 + parents = roots | {g.rsplit("/", 1)[0] for g in removed} + stray = [c for c in changed if c not in parents and not any(c == g or c.startswith(g + "/") for g in removed)] + if stray: + return None, {"error": f"changes outside .git: {stray[:5]}", **res} + changes = [ + f"ENTRYPOINT {json.dumps(cfg.get('Entrypoint') or [])}", + f"CMD {json.dumps(cfg.get('Cmd') or [])}", + f"LABEL fc.scrubbed_from={image}", + ] + new = f"{image}-scrub" + c = sh(["docker", "commit", *[x for ch in changes for x in ("--change", ch)], name, new]) + if c.returncode: + return None, {"error": f"commit failed: {c.stderr[-300:]}", **res} + return new, res + finally: + sh(["docker", "rm", "-f", name]) diff --git a/tests/verify/test_images.py b/tests/verify/test_images.py new file mode 100644 index 00000000..2f920928 --- /dev/null +++ b/tests/verify/test_images.py @@ -0,0 +1,146 @@ +import json +from pathlib import Path + +import pytest + +from datasmith.verify import images, pin, registry + + +def task(root: Path, name: str, dockerfile: str = "FROM base\n") -> Path: + d = root / name + (d / "environment").mkdir(parents=True) + (d / "environment" / "Dockerfile").write_text(dockerfile) + (d / "task.toml").write_text('[task]\nname = "x"\n\n[environment]\ncpus = 2\n') + return d + + +def test_env_digest_changes_with_the_build_context(tmp_path): + d = task(tmp_path, "o__r__1") + first = images.env_digest(d / "environment") + (d / "environment" / "Dockerfile").write_text("FROM other\n") + assert images.env_digest(d / "environment") != first + assert images.image_tag(d).startswith("fc-task/o__r__1:") + + +def test_set_and_read_pin(): + toml = '[task]\nname = "x"\n\n[environment]\ncpus = 2\n' + pinned = pin.set_pin(toml, "fc-task/a:1") + assert pin.read_pin(pinned) == "fc-task/a:1" + assert pin.read_pin(pin.set_pin(pinned, "fc-task/a:2")) == "fc-task/a:2" + assert pin.read_pin(pin.set_pin('[task]\nname = "x"\n', "t:1")) == "t:1" + assert pin.read_pin('[task]\ndocker_image = "not-env"\n') is None + + +def test_plan_pins_only_a_verified_image_of_the_same_context(tmp_path): + d = task(tmp_path, "o__r__1") + row = {"tag": "fc-task/o__r__1:abc", "verified": True, "env_digest": images.env_digest(d / "environment")} + assert pin.plan(d, None) == ("unbuilt", None) + assert pin.plan(d, {**row, "verified": False}) == ("unverified", None) + assert pin.plan(d, {**row, "env_digest": "other"}) == ("env-mismatch", None) + assert pin.plan(d, row) == ("unpinned", row["tag"]) + pin.write_pin(d, row["tag"]) + assert pin.plan(d, row) == ("ok", None) + + +def test_rows_by_task_keeps_the_last_row(tmp_path): + m = tmp_path / "manifest.jsonl" + for tag in ("a:1", "a:2"): + images.append_jsonl(m, {"task_id": "a", "tag": tag}) + assert images.rows_by_task(m)["a"]["tag"] == "a:2" + + +def test_build_claims_each_task_once_and_records_rows(tmp_path, monkeypatch): + for n in (1, 2, 3): + task(tmp_path / "tasks", f"o__r__{n}") + built = [] + + def fake_build_one(task_dir, out, limits, preexisting): + built.append(task_dir.name) + if task_dir.name == "o__r__2": + return None, {"task_id": task_dir.name, "stage": "build", "error": "boom"} + tar = out / f"{task_dir.name}.tar.zst" + tar.write_text("x") + return { + "task_id": task_dir.name, + "tag": "t", + "verified": True, + "tar_path": str(tar), + "tar_bytes": 1, + "bake_seconds": 1, + }, None + + monkeypatch.setattr(images, "build_one", fake_build_one) + monkeypatch.setattr(images, "sh", lambda cmd, **kw: type("P", (), {"stdout": "", "returncode": 0})()) + monkeypatch.setattr(images, "root_free_gb", lambda: 1000.0) + out = tmp_path / "out" + ids = ["o__r__1", "o__r__2", "o__r__3", "o__r__1"] + images.build(tmp_path / "tasks", ids, out, workers=2, limits=images.Limits(out_min_free_gb=0)) + assert sorted(built) == ["o__r__1", "o__r__2", "o__r__3"] + assert {r["task_id"] for r in images.read_jsonl(out / "manifest.jsonl")} == {"o__r__1", "o__r__3"} + assert [r["task_id"] for r in images.read_jsonl(out / "failures.jsonl")] == ["o__r__2"] + images.build(tmp_path / "tasks", ids, out, workers=2, limits=images.Limits(out_min_free_gb=0)) + assert len(built) == 3 + + +def test_build_stops_at_a_stop_file(tmp_path, monkeypatch): + task(tmp_path / "tasks", "o__r__1") + monkeypatch.setattr(images, "build_one", lambda *a: pytest.fail("built after STOP")) + monkeypatch.setattr(images, "sh", lambda cmd, **kw: type("P", (), {"stdout": "", "returncode": 0})()) + out = tmp_path / "out" + out.mkdir() + (out / "STOP").touch() + images.build(tmp_path / "tasks", ["o__r__1"], out) + + +class FakeRegistry: + def __init__(self): + self.pushed = [] + + def push(self, task_id, tag): + self.pushed.append(tag) + return {"task_id": task_id, "tag": tag, "uri": f"reg/{tag}", "digest": "sha256:1"} + + +def test_publish_pins_scrubs_probes_and_pushes(tmp_path, monkeypatch): + d = task(tmp_path / "tasks", "o__r__1") + row = { + "task_id": "o__r__1", + "tag": "fc-task/o__r__1:abc", + "image_id": "sha256:x", + "verified": True, + "env_digest": images.env_digest(d / "environment"), + "tar_path": "/x.tar.zst", + "tar_bytes": 1, + } + manifest = tmp_path / "manifest.jsonl" + images.append_jsonl(manifest, row) + monkeypatch.setattr(registry.images, "load", lambda r: r["tag"]) + monkeypatch.setattr( + registry.scrub, "scrub", lambda image: (image + "-scrub", {"nested_git_removed": "", "nested_tree_same": True}) + ) + monkeypatch.setattr(registry.scrub, "probe", lambda t, image, gt: {"status": "clean"}) + monkeypatch.setattr( + registry, + "sh", + lambda cmd, **kw: type("P", (), {"stdout": json.dumps([{"Id": "sha256:s", "Size": 5}]), "returncode": 0})(), + ) + reg = FakeRegistry() + out = tmp_path / "pub" + registry.publish(tmp_path / "tasks", manifest, out, reg) + assert reg.pushed == ["fc-task/o__r__1:abc-scrub"] + assert pin.read_pin((d / "task.toml").read_text()) == "fc-task/o__r__1:abc-scrub" + published = images.read_jsonl(out / "published.jsonl") + assert published[0]["scrubbed_from"] == row["tag"] and published[0]["tar_path"] is None + registry.publish(tmp_path / "tasks", manifest, out, reg) + assert len(reg.pushed) == 1 + + +def test_publish_refuses_a_leaking_image(tmp_path, monkeypatch): + d = task(tmp_path / "tasks", "o__r__1") + row = {"task_id": "o__r__1", "tag": "t:1", "verified": True, "env_digest": images.env_digest(d / "environment")} + monkeypatch.setattr(registry.images, "load", lambda r: r["tag"]) + monkeypatch.setattr(registry.scrub, "scrub", lambda image: (image + "-scrub", {})) + monkeypatch.setattr(registry.scrub, "probe", lambda t, image, gt: {"status": "leak", "leaks": "gt_commit"}) + monkeypatch.setattr(registry, "sh", lambda cmd, **kw: None) + with pytest.raises(RuntimeError, match="still leaks"): + registry.publish_one(d, row, FakeRegistry(), tmp_path) From 37fb966e8ad5293fdc895543eb5cbeedf3f9f904 Mon Sep 17 00:00:00 2001 From: Arjun Sharma Date: Mon, 19 Oct 2026 04:43:47 +0000 Subject: [PATCH 04/32] verify: measure stage, measure gate service and NopTouch; template logs the gate lease id --- .../harbor_adapter/template/tests/setup.sh | 6 +- .../harbor_adapter/template/tests/test.sh | 5 +- src/datasmith/verify/agents.py | 22 + src/datasmith/verify/cli.py | 40 +- src/datasmith/verify/measure.py | 445 ++++++++++++++++++ src/datasmith/verify/services/__init__.py | 0 src/datasmith/verify/services/measure_gate.py | 178 +++++++ tests/verify/test_measure.py | 141 ++++++ tests/verify/test_measure_gate.py | 69 +++ 9 files changed, 897 insertions(+), 9 deletions(-) create mode 100644 src/datasmith/verify/agents.py create mode 100644 src/datasmith/verify/measure.py create mode 100644 src/datasmith/verify/services/__init__.py create mode 100644 src/datasmith/verify/services/measure_gate.py create mode 100644 tests/verify/test_measure.py create mode 100644 tests/verify/test_measure_gate.py diff --git a/src/datasmith/harbor_adapter/template/tests/setup.sh b/src/datasmith/harbor_adapter/template/tests/setup.sh index 633f178c..7d074890 100644 --- a/src/datasmith/harbor_adapter/template/tests/setup.sh +++ b/src/datasmith/harbor_adapter/template/tests/setup.sh @@ -152,13 +152,13 @@ SETUP_PHASE="lsv_init" echo "[$(ts)] [setup] Starting LSV init..." lsv_init_start=$(date +%s) set +e -# Serialize the measure through the host gate (rl/measure_gate.py); fail-open. SID uses the kernel uuid +# Serialize the measure through the host gate; it runs ungated when the gate is down. SID uses the kernel uuid # because $$ is the same in every container. --connect-timeout keeps a dropped-packet host from stalling -m. _MG_URL="${MEASURE_GATE_URL:-}" _MG_SID="mg-init-$(cat /proc/sys/kernel/random/uuid 2>/dev/null || echo "$(hostname 2>/dev/null || echo h)-$$-${RANDOM}")" -# Logged like test.sh's acquires (" 1|0") so the pipeline can drop ungated trials. +# Logged like test.sh's acquires (" 1|0"). _mg_resp="$(curl -fsS --connect-timeout "${MEASURE_GATE_CONNECT_TIMEOUT:-5}" -m "${MEASURE_GATE_ACQUIRE_WAIT:-7200}" -X POST "${_MG_URL}/acquire?sid=${_MG_SID}" 2>/dev/null)" -case "${_mg_resp}" in *'"acquired": true'*) echo "init 1" ;; *) echo "init 0" ;; esac >> /logs/artifacts/measure_gate.txt +case "${_mg_resp}" in *'"acquired": true'*) echo "init ${_MG_SID} 1" ;; *) echo "init ${_MG_SID} 0" ;; esac >> /logs/artifacts/measure_gate.txt python /tests/lsv_init.py{% if rounds is not none %} --rounds {{ rounds }}{% endif %} lsv_exit=$? curl -fsS --connect-timeout "${MEASURE_GATE_CONNECT_TIMEOUT:-5}" -m 5 -X POST "${_MG_URL}/release?sid=${_MG_SID}" >/dev/null 2>&1 || true diff --git a/src/datasmith/harbor_adapter/template/tests/test.sh b/src/datasmith/harbor_adapter/template/tests/test.sh index d0f267fe..4f3d49fe 100644 --- a/src/datasmith/harbor_adapter/template/tests/test.sh +++ b/src/datasmith/harbor_adapter/template/tests/test.sh @@ -38,14 +38,15 @@ REWARD_DIR="/logs/verifier" mkdir -p "${LOG_DIR}" "${LOG_DIR}/.snapshots" "${LOG_DIR}/lsv" "${REWARD_DIR}" -# Host measure gate (rl/measure_gate.py), fail-open. Each acquire is logged (" 1|0") so ungated trials can be dropped. +# Host measure gate (datasmith.verify.services.measure_gate); a trial runs ungated when the gate is down. +# Each acquire is logged as " 1|0": ungated trials are dropped, and a stopped job can release its leases. _MG_URL="${MEASURE_GATE_URL:-}" mg_acquire() { _MG_SID="mg-$1-$(cat /proc/sys/kernel/random/uuid 2>/dev/null || echo "$(hostname 2>/dev/null || echo h)-$$-${RANDOM}")" local resp ok=0 resp="$(curl -fsS --connect-timeout "${MEASURE_GATE_CONNECT_TIMEOUT:-5}" -m "${MEASURE_GATE_ACQUIRE_WAIT:-7200}" -X POST "${_MG_URL}/acquire?sid=${_MG_SID}" 2>/dev/null || true)" case "${resp}" in *'"acquired": true'*) ok=1 ;; esac - echo "$1 ${ok}" >> "${LOG_DIR}/measure_gate.txt" + echo "$1 ${_MG_SID} ${ok}" >> "${LOG_DIR}/measure_gate.txt" } mg_release() { curl -fsS --connect-timeout "${MEASURE_GATE_CONNECT_TIMEOUT:-5}" -m 5 -X POST "${_MG_URL}/release?sid=${_MG_SID}" >/dev/null 2>&1 || true diff --git a/src/datasmith/verify/agents.py b/src/datasmith/verify/agents.py new file mode 100644 index 00000000..b095b120 --- /dev/null +++ b/src/datasmith/verify/agents.py @@ -0,0 +1,22 @@ +"""Harbor agents for measuring: NopTouch appends a newline to each file the oracle patch changes, so LSV times the +oracle's benchmarks on unchanged code (an empty diff selects nothing).""" + +from __future__ import annotations + +import shlex + +from harbor.agents.nop import NopAgent + + +class NopTouch(NopAgent): + def __init__(self, *args, files: list[str] | None = None, **kwargs): + super().__init__(*args, **kwargs) + self._files = list(files or []) + + async def run(self, instruction, environment, context) -> None: + if not self._files: + return + quoted = " ".join(shlex.quote(f) for f in self._files) + await environment.exec( + command=f'cd /workspace/repo && for f in {quoted}; do [ -f "$f" ] && printf "\\n" >> "$f"; done; git diff --stat' + ) diff --git a/src/datasmith/verify/cli.py b/src/datasmith/verify/cli.py index 4a27fd28..5364e8f1 100644 --- a/src/datasmith/verify/cli.py +++ b/src/datasmith/verify/cli.py @@ -10,6 +10,8 @@ fc-verify build TASKS_ROOT OUT [--ids FILE ...] [--workers N] [--retry-failed] fc-verify publish TASKS_ROOT MANIFEST OUT [--region R] [--repo NAME] [--until-idle] fc-verify pins TASKS_ROOT MANIFEST pin status of every task folder +fc-verify measure SURVEY TASK ... [--manifest M] [--workers N] [--oracle-runs 3] [--nop-runs 3] [--gate URL] +fc-verify release-leases JOB_DIR [--gate URL] free the gate slots of a stopped job fc-verify failure LOG ... error class of failed image builds """ @@ -21,7 +23,7 @@ import sys from pathlib import Path -from datasmith.verify import images, pin, precheck, registry, report, review, status +from datasmith.verify import images, measure, pin, precheck, registry, report, review, status from datasmith.verify.noise import classify_all, load_rule from datasmith.verify.survey import DECISIONS, Survey, task_id @@ -74,6 +76,19 @@ def _precheck(survey: Survey, a: argparse.Namespace) -> None: print(json.dumps({"task": t, "end": end, **{k: result.get(k) for k in ("by_deps", "patch_failed", "error")}})) +def _measure(survey: Survey, a: argparse.Namespace) -> None: + s = measure.Settings( + n_oracle=a.oracle_runs, + n_nop=a.nop_runs, + gate_url=a.gate, + confirm_rounds=a.confirm_rounds, + max_heavy=a.max_heavy, + harbor=a.harbor, + ) + for task, phase in measure.measure_many(survey, _tasks(a.tasks), s, a.workers, a.manifest).items(): + print(f"{task}: {phase}") + + COMMANDS = { "status": _status, "classify": _classify, @@ -126,13 +141,18 @@ def _pins(a: argparse.Namespace) -> int: return 1 if bad else 0 +def _release(a: argparse.Namespace) -> int: + print("\n".join(measure.release_leases(a.job_dir, a.gate))) + return 0 + + def _failure(a: argparse.Namespace) -> int: for log in a.logs: print(json.dumps({"log": str(log), **report.failure_reason(log)})) return 0 -NO_SURVEY = {"build": _build, "publish": _publish, "pins": _pins, "failure": _failure} +NO_SURVEY = {"build": _build, "publish": _publish, "pins": _pins, "failure": _failure, "release-leases": _release} def main(argv: list[str] | None = None) -> int: @@ -140,10 +160,10 @@ def main(argv: list[str] | None = None) -> int: prog="fc-verify", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) sub = ap.add_subparsers(dest="cmd", required=True) - for name in ("status", "classify", "report", "review", "auto", "add", "precheck"): + for name in ("status", "classify", "report", "review", "auto", "add", "precheck", "measure"): p = sub.add_parser(name) p.add_argument("survey", type=Path) - if name in ("classify", "report", "add", "precheck"): + if name in ("classify", "report", "add", "precheck", "measure"): p.add_argument("tasks", nargs="*") sub.choices["review"].add_argument("task") sub.choices["review"].add_argument("decision", choices=DECISIONS) @@ -171,6 +191,18 @@ def main(argv: list[str] | None = None) -> int: pb.add_argument("--region", default="us-east-1") pb.add_argument("--repo", default="formulacode") pb.add_argument("--until-idle", action="store_true", help="keep publishing while a build runs") + m = sub.choices["measure"] + m.add_argument("--manifest", type=Path, help="image manifest; images are loaded before measuring") + m.add_argument("--workers", type=int, default=1) + m.add_argument("--oracle-runs", type=int, default=3) + m.add_argument("--nop-runs", type=int, default=3) + m.add_argument("--gate", default="http://127.0.0.1:8267") + m.add_argument("--confirm-rounds", type=int, default=0) + m.add_argument("--max-heavy", type=int, default=1) + m.add_argument("--harbor", default="harbor") + rl = sub.add_parser("release-leases") + rl.add_argument("job_dir", type=Path) + rl.add_argument("--gate", default="http://127.0.0.1:8267") pp = sub.add_parser("pins") pp.add_argument("tasks_root", type=Path) pp.add_argument("manifest", type=Path) diff --git a/src/datasmith/verify/measure.py b/src/datasmith/verify/measure.py new file mode 100644 index 00000000..0043250d --- /dev/null +++ b/src/datasmith/verify/measure.py @@ -0,0 +1,445 @@ +"""Measure stage: one harbor job per task with the oracle runs and the no-change runs, written to oracle/.json. + +Each run is a paired A-B-B-A timing inside one container (template tests/lsv_measure.py); the host measure gate limits +how many timings run at once. A job whose trials did not all finish, or that has infrastructure failures, is counted +in the state (`measure_jobs`) and runs again under a new job name, so harbor never resumes it. +""" + +from __future__ import annotations + +import json +import math +import os +import re +import statistics +import subprocess +import time +import urllib.request +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any + +import yaml + +from datasmith.verify import images +from datasmith.verify.noise import load_rule, repo_spreads, rule_for, thetas +from datasmith.verify.survey import Survey, repo_of + +NOP_AGENT = "datasmith.verify.agents:NopTouch" +# Container temp dir inside the mounted artifacts dir, so benchmark leftovers land on the jobs disk. +TRIAL_TMPDIR = "/logs/artifacts/.tmp" +INFRA_EXCEPTIONS = { + "AgentSetupTimeoutError", + "EnvironmentStartTimeoutError", + "SetupScriptError", + "HealthcheckError", + "AddTestsDirError", + "DownloadVerifierDirError", +} +SECRET_ENV = re.compile(r"KEY|TOKEN|SECRET|PASSWORD") + + +@dataclass +class Settings: + n_oracle: int = 3 + n_nop: int = 3 + gate_url: str = "http://127.0.0.1:8267" + gateway_url: str = "http://127.0.0.1:8268" + gate_acquire_wait: int = 7200 + gate_secret_file: str | None = None + trial_cpus: int = 2 + trial_memory_mb: int = 12288 + timeout_s: int = 3600 + lsv_rounds: int = 5 + confirm_rounds: int = 0 + snapshot_workers: int = 1 + verifier_timeout_multiplier: float = 1.0 + max_heavy: int = 1 + heavy_benchmarks: int = 200 + harbor: str = "harbor" + # Per task: {"mounts": {host: container}, "env": {...}} for tasks whose benchmarks read files at run time. + offline: dict[str, dict] = field(default_factory=dict) + + +def read_json(path: Path) -> Any: + try: + return json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return None + + +def utc() -> str: + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + + +def when(stamp: str | None) -> float | None: + try: + return datetime.fromisoformat(stamp.replace("Z", "+00:00")).timestamp() + except (AttributeError, ValueError): + return None + + +def seconds(span: dict | None) -> float | None: + start, end = when((span or {}).get("started_at")), when((span or {}).get("finished_at")) + return None if start is None or end is None else end - start + + +def outputs_dir(trial: Path) -> Path: + """/verifier/artifacts with a separate verifier container, else /artifacts.""" + sep = trial / "verifier" / "artifacts" + return sep if sep.is_dir() else trial / "artifacts" + + +def patch_files(solve_sh: Path) -> list[str]: + return sorted(set(re.findall(r"^\+\+\+ b/(\S+)", solve_sh.read_text(), re.M))) if solve_sh.exists() else [] + + +def confirmed_speedups(trial: Path) -> dict[str, float]: + """Benchmarks the confirmation pass timed again: median over both passes' paired log ratios.""" + lsv = outputs_dir(trial) / "lsv" + first = (read_json(lsv / "lsv_measure_results.json") or {}).get("benchmarks") or {} + out = {} + for name, b in ((read_json(lsv / "lsv_measure_confirm.json") or {}).get("benchmarks") or {}).items(): + logs = [x for x in ((first.get(name) or {}).get("paired") or {}).get("log_ratios") or [] if x is not None] + logs += [x for x in (b.get("paired") or {}).get("log_ratios") or [] if x is not None] + if logs: + out[name] = math.exp(statistics.median(logs)) + return out + + +def log_speedups(trial: Path) -> dict[str, float]: + """ln speedup per benchmark: reward.json's, with the confirmation pass's medians in place.""" + per = (read_json(trial / "verifier" / "reward.json") or {}).get("per_benchmark_speedups") + if not isinstance(per, dict): + return {} + out = {} + for name, value in {**per, **confirmed_speedups(trial)}.items(): + try: + g = float(value) + except (TypeError, ValueError): + continue + if math.isfinite(g) and g > 0: + out[name] = math.log(g) + return out + + +def test_outcomes(trial: Path) -> dict[str, Any]: + tr = read_json(outputs_dir(trial) / "test_results.json") or {} + reg = tr.get("regression") or {} + failing = { + t["nodeid"] for t in (tr.get("results") or {}).get("tests") or [] if t.get("outcome") in ("failed", "error") + } + return { + "ran": bool(reg.get("ran")), + "n_base_pass": reg.get("n_base_pass") or 0, + "n_tests": reg.get("n_agent_tests") or 0, + "n_regressed": reg.get("n_regressed") or 0, + "failing": sorted(failing), + } + + +def patch_failures(trial: Path) -> list[str]: + """Files with a hunk the oracle's `patch` could not apply.""" + log = trial / "agent" / "oracle.txt" + failed, current = [], None + for line in log.read_text(errors="replace").splitlines() if log.exists() else []: + if line.startswith("patching file "): + current = line[len("patching file ") :].strip() + elif line.startswith("Hunk #") and "FAILED" in line and current and current not in failed: + failed.append(current) + return failed + + +def ungated(trial: Path, timeouts: list[dict]) -> bool: + """True when a gate acquire of this trial failed, so a timing may have run beside others.""" + acquires = outputs_dir(trial) / "measure_gate.txt" + if acquires.is_file(): + return any(line.split()[-1] != "1" for line in acquires.read_text().splitlines() if line.strip()) + result = read_json(trial / "result.json") or {} + start, end = when(result.get("started_at")), when(result.get("finished_at")) + if start is None: + return bool(timeouts) + end = end or time.time() + return any(t["requested_at"] <= end and t["timed_out_at"] >= start for t in timeouts) + + +def trial_run(trial: Path, timeouts: list[dict]) -> dict[str, Any]: + """One measure trial as a record run. `infra` marks a trial without data (setup failure, ungated, no reward file).""" + rj = read_json(trial / "verifier" / "reward.json") + agent = (read_json(trial / "config.json") or {}).get("agent") or {} + role = ( + "nop" + if str(agent.get("import_path") or "").endswith(":NopTouch") + else ("oracle_baseline" if not agent.get("model_name") else "oracle") + ) + reuse = (read_json(outputs_dir(trial) / "lsv" / "lsv_init_results.json") or {}).get("baseline_reuse") or {} + result = read_json(trial / "result.json") or {} + exc = (result.get("exception_info") or {}).get("exception_type") + gated = not ungated(trial, timeouts) + return { + "trial": trial.name, + "trial_dir": str(trial), + "role": role, + "run": int((agent.get("model_name") or "run1")[3:]), + "exception": exc, + "cpu_model": (reuse.get("trial_fingerprint") or {}).get("cpu_model"), + "gate_acquired": gated, + "baseline_reused": reuse.get("reused"), + "lsv_mean_speedup": (rj or {}).get("lsv_mean_speedup"), + "tests_passed": (rj or {}).get("tests_passed"), + "snapshots_passed": (rj or {}).get("snapshots_passed"), + "snapshot_verify_ran": (rj or {}).get("snapshot_verify_ran"), + "project_shadowed": (rj or {}).get("project_shadowed"), + "log_speedups": log_speedups(trial), + "tests": test_outcomes(trial), + "patch_failed": patch_failures(trial), + "timing": { + "started_at": result.get("started_at"), + "finished_at": result.get("finished_at"), + "trial_s": seconds(result), + **{ + f"{k}_s": seconds(result.get(k)) + for k in ("environment_setup", "agent_setup", "agent_execution", "verifier") + }, + }, + "infra": exc in INFRA_EXCEPTIONS or not gated or rj is None, + } + + +def job_leases(job_dir: Path) -> list[str]: + """Gate lease ids its trials logged; lines without an id (older templates) are skipped.""" + sids = [] + for log in job_dir.glob("*/**/measure_gate.txt"): + sids += [parts[1] for line in log.read_text().splitlines() if len(parts := line.split()) == 3] + return sorted(set(sids)) + + +def release_leases(job_dir: Path, gate_url: str) -> list[str]: + """Release the leases of a stopped job; the gate ignores ids it does not hold.""" + released = [] + for sid in job_leases(job_dir): + req = urllib.request.Request(f"{gate_url}/release?sid={sid}", method="POST") # noqa: S310 + try: + with urllib.request.urlopen(req, timeout=10): # noqa: S310 + released.append(sid) + except OSError: + continue + return released + + +def gate_token(secret_file: str, label: str) -> str: + from datasmith.verify.services.measure_gate import token_for + + return token_for(Path(secret_file).read_bytes().strip(), label) + + +def confirm_spec(survey: Survey, task: str, rule: dict, rounds: int) -> dict: + """tests/confirm.json: theta per benchmark from the task's existing record, else theta_min for every benchmark.""" + r = rule_for(rule, repo_of(task)) + out = { + "rounds": rounds, + "theta_default": r.get("theta_min", r.get("theta")), + "theta": {}, + "oracle_up": [], + "theta_source": "floor", + } + record = read_json(survey.root / "oracle" / f"{task}.json") + if record and record.get("runs"): + spreads = repo_spreads(list(survey.records().values())) + out.update(theta=thetas(rule, record, spreads), theta_source="record") + return out + + +def job_config(survey: Survey, task: str, job: str, s: Settings, tasks_root: Path) -> dict[str, Any]: + """Harbor job: the oracle once per oracle run and NopTouch once per no-change run, all concurrent.""" + timeout = s.timeout_s + agents = [{"name": "oracle", "override_timeout_sec": timeout}] + agents += [ + {"name": "oracle", "model_name": f"run{i}", "override_timeout_sec": timeout} for i in range(2, s.n_oracle + 1) + ] + files = patch_files(survey.task_dir(task) / "solution" / "solve.sh") + agents += [ + {"import_path": NOP_AGENT, "model_name": f"run{i}", "kwargs": {"files": files}} for i in range(1, s.n_nop + 1) + ] + env = { + "MEASURE_GATE_URL": s.gate_url, + "TMPDIR": TRIAL_TMPDIR, + "MEASURE_GATE_ACQUIRE_WAIT": str(s.gate_acquire_wait), + "FC_AGENT_TAG": "agent", + "FORMULACODE_SNAPSHOT_WORKERS": str(s.snapshot_workers), + } + cfg: dict[str, Any] = { + "job_name": job, + "jobs_dir": str(survey.root / "jobs"), + "n_attempts": 1, + "n_concurrent_trials": s.n_oracle + s.n_nop, + "quiet": False, + "environment": { + "type": "docker", + "force_build": False, + "delete": True, + "override_cpus": s.trial_cpus, + "override_memory_mb": s.trial_memory_mb, + "env": env, + }, + "verifier": { + "env": { + "SUPABASE_URL": s.gateway_url, + "SUPABASE_ANON_KEY": "local", + "FORMULACODE_SNAPSHOT_FILTER": ".*", + "FORMULACODE_NO_UPLOAD": "", + "DATASMITH_LSV_ROUNDS": str(s.lsv_rounds), + } + }, + "agents": agents, + "datasets": [{"path": str(tasks_root), "task_names": [task]}], + "verifier_timeout_multiplier": s.verifier_timeout_multiplier, + } + if offline := s.offline.get(task): + cfg["environment"]["mounts_json"] = [ + {"type": "bind", "source": src, "target": dst, "read_only": True, "bind": {"create_host_path": False}} + for src, dst in offline.get("mounts", {}).items() + ] + env.update(offline.get("env", {})) + if s.gate_secret_file: + cfg["verifier"]["env"]["MEASURE_GATE_TOKEN"] = gate_token(s.gate_secret_file, job) + return cfg + + +def tasks_root_for(survey: Survey, task: str, job: str, confirm: dict | None) -> Path: + """The pinned tasks root, or a copy for this job with tests/confirm.json for the confirmation pass.""" + if not confirm: + return survey.root / "pinned" + import shutil + + dst = survey.root / "tasks-confirm" / job + shutil.copytree(survey.task_dir(task), dst / task, symlinks=True, dirs_exist_ok=True) + (dst / task / "tests" / "confirm.json").write_text(json.dumps(confirm)) + return dst + + +def run_harbor( + survey: Survey, job: str, config: Path, s: Settings, run: Callable[..., Any] = subprocess.run +) -> list[Path]: + env = {k: v for k, v in os.environ.items() if not SECRET_ENV.search(k)} + env.update(HARBOR_DOCKER_HOST_NET="1", COMPOSE_BAKE="false", DOCKER_BUILDKIT="0") + (survey.root / "logs").mkdir(parents=True, exist_ok=True) + with (survey.root / "logs" / f"{job}.log").open("a") as log: + run( + [s.harbor, "run", "--config", str(config)], + env=env, + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + return sorted(p.parent for p in (survey.root / "jobs" / job).glob("*/result.json")) + + +def gate_timeouts(s: Settings) -> list[dict]: + try: + with urllib.request.urlopen(s.gate_url + "/timeouts", timeout=10) as r: # noqa: S310 + return json.load(r)["timeouts"] + except OSError: + return [] + + +def is_heavy(survey: Survey, task: str, s: Settings) -> bool: + cov = read_json(survey.root / "coverage" / f"{task}.json") or {} + return (cov.get("selected") or 0) > s.heavy_benchmarks + + +def measure_task( + survey: Survey, + task: str, + s: Settings, + manifest: Path | None = None, + grade: Callable[[Path, dict], dict] | None = None, + run: Callable[..., Any] = subprocess.run, +) -> str: + """Measure one task and return its new phase: measured, no_benchmarks or error. + + `grade(trial, run_row)` may add fields to each run (for example a correctness verdict from the training grader). + """ + st = survey.state(task) + n = st.get("measure_jobs", 0) + job = f"{task}__oracle" + (f"-{n}" if n else "") + if manifest is not None and (row := images.rows_by_task(manifest).get(task)): + images.load(row) + rule = load_rule(survey.rule_path) if s.confirm_rounds > 0 else None + confirm = confirm_spec(survey, task, rule, s.confirm_rounds) if rule else None + config = survey.root / "configs" / f"{job}.yaml" + if not config.exists(): + config.parent.mkdir(parents=True, exist_ok=True) + config.write_text( + yaml.safe_dump( + job_config(survey, task, job, s, tasks_root_for(survey, task, job, confirm)), sort_keys=False + ) + ) + trials = run_harbor(survey, job, config, s, run) + timeouts = gate_timeouts(s) + runs = [trial_run(t, timeouts) for t in trials] + for r, t in zip(runs, trials, strict=True): + if grade: + r.update(grade(t, r)) + infra = [r["trial"] for r in runs if r.pop("infra")] + want = s.n_oracle + s.n_nop + if len(runs) < want or infra: + survey.save_state( + task, + { + **st, + "phase": "error", + "measure_jobs": n + 1, + "last_error": f"{job}: {len(runs)}/{want} trials, infra failures {infra}", + }, + ) + return "error" + checksum = (read_json(trials[0] / "result.json") or {}).get("task_checksum") + record = { + "task": task, + "job": job, + "tag": st.get("tag"), + "n_oracle": s.n_oracle, + "n_nop": s.n_nop, + "task_checksum": checksum, + "measured_at": utc(), + "cpu_models": sorted({r["cpu_model"] for r in runs if r.get("cpu_model")}), + "runs": sorted(runs, key=lambda r: (r["role"] == "nop", r["run"])), + } + out = survey.root / "oracle" / f"{task}.json" + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n") + phase = "measured" if any(r["log_speedups"] for r in runs) else "no_benchmarks" + survey.save_state(task, {**st, "phase": phase, "measure_jobs": n + 1, "cpu_model": ",".join(record["cpu_models"])}) + return phase + + +def measure_many( + survey: Survey, + tasks: list[str], + s: Settings, + workers: int = 1, + manifest: Path | None = None, + grade: Callable[[Path, dict], dict] | None = None, +) -> dict[str, str]: + """Measure tasks `workers` at a time, with at most `s.max_heavy` heavy tasks at once (0 = no limit).""" + import threading + + heavy = threading.Semaphore(s.max_heavy) if s.max_heavy else None + results: dict[str, str] = {} + + def one(task: str) -> None: + if (survey.root / "STOP").exists(): + return + if heavy and is_heavy(survey, task, s): + with heavy: + results[task] = measure_task(survey, task, s, manifest, grade) + else: + results[task] = measure_task(survey, task, s, manifest, grade) + + with ThreadPoolExecutor(max(1, workers)) as pool: + list(pool.map(one, tasks)) + return results diff --git a/src/datasmith/verify/services/__init__.py b/src/datasmith/verify/services/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/datasmith/verify/services/measure_gate.py b/src/datasmith/verify/services/measure_gate.py new file mode 100644 index 00000000..3e8f8b0e --- /dev/null +++ b/src/datasmith/verify/services/measure_gate.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Host-side semaphore limiting concurrent LSV measure steps (timing is memory-bandwidth sensitive). + +Trials POST /acquire?sid=X before measuring and /release?sid=X after; a trial runs ungated when the gate is down, +and a lease TTL frees the slots of dead holders. Defaults env-overridable via MEASURE_GATE_{PORT,SLOTS,TTL,ACQUIRE_WAIT}. +With --secret-file, acquire and release also need &token=