From 77de50a1e4bd54f1941eca35dc732ec45148e987 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 22 Oct 2026 16:07:21 +0800 Subject: [PATCH] fix(edgebench): freeze shared integration source for CLI cohorts Signed-off-by: huangruiteng --- benchmark/edgebench/README.md | 66 ++++++++- benchmark/edgebench/online_judge.py | 5 + benchmark/edgebench/run.py | 12 +- benchmark/edgebench/serve.py | 18 +++ benchmark/edgebench/source-default.json | 4 + benchmark/edgebench/source.py | 145 ++++++++++++++++++ benchmark/tests/test_edgebench_source.py | 180 +++++++++++++++++++++++ 7 files changed, 425 insertions(+), 5 deletions(-) create mode 100644 benchmark/edgebench/serve.py create mode 100644 benchmark/edgebench/source-default.json create mode 100644 benchmark/edgebench/source.py create mode 100644 benchmark/tests/test_edgebench_source.py diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index 1164c53a58..a450a7f93a 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -3,10 +3,68 @@ This research adapter calls SForge's native `run_agent`. Task images, prompts, submission cooldowns, judge isolation and scoring remain owned by the pinned EdgeBench checkout. Best-only adds host capture/admission scheduling; native and -blind retain native automatic evaluation. Install SForge and Harbor in an isolated -runner environment. Run the controller on Linux with Docker and native SForge +blind retain native automatic evaluation. The CLI defaults to the shared +[EdgeBench integration branch](https://github.com/huangruiteng/EdgeBench/tree/codex/edgebench-integration), +maintained jointly by `huangruiteng` and `loopx-agent`. The canonical repository +and branch live in [`source-default.json`](source-default.json). +Run the controller on Linux with Docker and native SForge iptables permissions; a macOS Docker socket alone does not provide that boundary. +## Select and freeze the native provider + +The first CLI entry for a log directory fetches the selected branch into +`.edgebench-source/checkout` and atomically records its exact commit in +`.edgebench-source/receipt.json`. Concurrent worker/judge startup shares that lock. +Later entries reuse the frozen checkout without fetching, even if the branch or +repository default changes. Dirty/moved source, conflicting explicit selection, +and existing runs without a source receipt fail before model/container entry. +Use a new log directory to adopt a newer integration revision; never repin an +active or historical cohort. + +Prepare dependencies in an isolated environment before starting a study: + +```sh +python -m benchmark.edgebench.source --log-dir /data/private-runs +python -m pip install /data/private-runs/.edgebench-source/checkout harbor +``` + +The bootstrap selects this checkout before importing SForge, so an unrelated +installed package or ambient `PYTHONPATH` cannot silently supply the provider. +Worker `runtime-receipt.json` includes the verified `sforge_source`. This records +code selection, not enabled task/scoring options or benchmark qualification. +No dependency installation is performed automatically. + +Start the native judge and workers with the **same log directory**: + +```sh +SFORGE_TASKS_DIR=/data/tasks python -m benchmark.edgebench.serve \ + --log-dir /data/private-runs --port 8080 +python -m benchmark.edgebench.run \ + --task TASK_ID --tasks-dir /data/tasks --log-dir /data/private-runs \ + --run-id UNIQUE_ATTEMPT --worker native-goal --model MODEL --effort high \ + --feedback native --submission-cooldown 1800 --eval-interval 300 \ + --judge-url http://HOST:8080 +``` + +For an alternative source, pass `--sforge-repository URL` with `--sforge-ref REF` +or a full `--sforge-revision SHA` on the first entry (including preparation). +For example, `--sforge-repository https://github.com/ByteDance-Seed/EdgeBench.git +--sforge-ref refs/heads/main` selects upstream instead. These source options also +apply to the online judge CLI; `SFORGE_LOG_DIR` is equivalent to `--log-dir` there. +Use an explicit SHA to replay a known revision. A server in another log directory +is a separate cohort: the launcher cannot attest a remote server's loaded code. + +These defaults apply to the LoopX EdgeBench CLI. Direct `sforge serve` and imported +Python APIs remain caller-managed compatibility entrypoints; use them only with +an explicitly pinned provider environment. Keep offline backfill pinned to the +original scorer, or record regrading under a different source separately. +Maintainers update the shared branch through reviewed integration work and keep +its PR manifest current. Changing the default repository/branch requires updating +the single source file and this protocol; adding a second contributor-specific +default would split the source of truth again. + +## Worker profiles + The five worker profiles are `official`, `single`, `native-goal`, `heartbeat-resume`, and `heartbeat-explore`. `official` uses SForge's Codex Stop hook and outer recovery. `single` executes once with neither. `native-goal` uses @@ -65,7 +123,7 @@ run's slot after verified worker removal. An evaluation still in flight keeps that slot until its native result is terminal; stopping a solver alone does not free evaluator capacity. Native histories, archives and original registration identities remain readable, and released registrations reject new captures. -Native/blind use ordinary `sforge serve` instead. Existing v1 cohorts remain +Native/blind use `python -m benchmark.edgebench.serve` instead. Existing v1 cohorts remain frozen: this does not upgrade or reclaim slots in an already-running server. An operator may explicitly pass `--allow-resource-overcommit` for a monitored @@ -196,7 +254,7 @@ resource policy in the study manifest before launching a new comparison. Use explicit `--feedback best-only` to retain strict-improvement publication, or `--feedback blind` as an evaluator-feedback-free control. Harbor is unchanged. -For native archive submission, start ordinary `sforge serve` and use the runner +For native archive submission, start `python -m benchmark.edgebench.serve` and use the runner without `--feedback` to select the defaults. Explicit `--feedback native --submission-cooldown 120 --eval-interval 300` restores the previous native timing for a new attempt. Do not edit running receipts. Game diff --git a/benchmark/edgebench/online_judge.py b/benchmark/edgebench/online_judge.py index f93b7ad23c..930b73d34e 100644 --- a/benchmark/edgebench/online_judge.py +++ b/benchmark/edgebench/online_judge.py @@ -17,6 +17,11 @@ from enum import StrEnum from pathlib import Path +from benchmark.edgebench.source import bootstrap + +if __name__ == "__main__": + bootstrap(keep_log_arg=False) + from fastapi import File, Form, HTTPException, Query, UploadFile from sforge.harness.config import load_config from sforge.harness.judge_server import ( diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 0ed8137756..0052db2ce8 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -1,4 +1,8 @@ -"""Launch one native EdgeBench trial; run the judge with native `sforge serve`.""" +"""Launch a native trial; start its judge with benchmark.edgebench.serve. + +CLI source options: --sforge-repository, --sforge-ref, --sforge-revision. +Defaults and the experiment-local source lock are owned by edgebench.source. +""" from __future__ import annotations @@ -14,6 +18,11 @@ from pathlib import Path from urllib.parse import urlsplit +from benchmark.edgebench.source import active_source_receipt, bootstrap + +if __name__ == "__main__": + bootstrap() # Select SForge before any provider or adapter imports. + from sforge.harness.benchmark import load_benchmark from sforge.harness.config import SForgeConfig from sforge.harness.constants import get_admin_secret @@ -352,6 +361,7 @@ def main(argv=None): if agent.task_entry == "loopx-planned" else {}), "model": args.model, "effort": args.effort, "timeout_seconds": args.timeout, "loopx_commit": pins[0], "runner_commit": pins[1], + "sforge_source": active_source_receipt(), **turn_envelope_receipt(explicit=args.turn_envelope), **({"native_goal_continue": True} if args.native_goal_continue else {}), "task_sha256": hashlib.sha256(task_file.read_bytes()).hexdigest(), diff --git a/benchmark/edgebench/serve.py b/benchmark/edgebench/serve.py new file mode 100644 index 0000000000..1ce0d188b8 --- /dev/null +++ b/benchmark/edgebench/serve.py @@ -0,0 +1,18 @@ +"""Native SForge judge using the same frozen source as EdgeBench workers.""" +from __future__ import annotations + +import sys + +from benchmark.edgebench.source import bootstrap + + +def main(): + bootstrap(keep_log_arg=False) + from sforge.cli import main as native_main + + sys.argv.insert(1, "serve") + native_main() + + +if __name__ == "__main__": + main() diff --git a/benchmark/edgebench/source-default.json b/benchmark/edgebench/source-default.json new file mode 100644 index 0000000000..45510bed9d --- /dev/null +++ b/benchmark/edgebench/source-default.json @@ -0,0 +1,4 @@ +{ + "repository": "https://github.com/huangruiteng/EdgeBench.git", + "ref": "refs/heads/codex/edgebench-integration" +} diff --git a/benchmark/edgebench/source.py b/benchmark/edgebench/source.py new file mode 100644 index 0000000000..256297e063 --- /dev/null +++ b/benchmark/edgebench/source.py @@ -0,0 +1,145 @@ +"""Experiment-local SForge selection, before importing the native provider. + +This research adapter owns a dependency choice, not scoring or LoopX policy. +One atomic source directory freezes the worker/judge cohort's revision. +""" +from __future__ import annotations + +import argparse +import fcntl +import importlib.util +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile + +DEFAULT_FILE = Path(__file__).with_name("source-default.json") +STATE_DIR = ".edgebench-source" +RECEIPT_ENV = "LOOPX_EDGEBENCH_SOURCE_RECEIPT" + + +def _git(checkout: Path, *args: str) -> str: + try: + result = subprocess.run(["git", "-C", str(checkout), *args], + capture_output=True, text=True, timeout=180) + except subprocess.TimeoutExpired: + raise RuntimeError(f"EdgeBench git {args[0]} timed out") from None + if result.returncode: + # A transport error can contain a credential-bearing remote URL. + raise RuntimeError(f"EdgeBench git {args[0]} failed (exit {result.returncode})") + return result.stdout.strip() + + +def _validate(state: Path) -> dict: + receipt = json.loads((state / "receipt.json").read_text()) + if (not isinstance(receipt, dict) + or receipt.get("schema_version") != "edgebench_source_v1" + or not isinstance(receipt.get("commit"), str) + or not re.fullmatch(r"[0-9a-f]{40}", receipt["commit"]) + or not all(isinstance(receipt.get(k), str) and receipt[k] + for k in ("repository", "ref"))): + raise ValueError("Invalid EdgeBench source receipt; preserve it and use a new log directory") + checkout = state / "checkout" + if (_git(checkout, "rev-parse", "HEAD") != receipt["commit"] + or _git(checkout, "remote", "get-url", "origin") != receipt["repository"] + or _git(checkout, "status", "--porcelain", "--untracked-files=normal")): + raise ValueError("Frozen EdgeBench source changed; restore it or use a new log directory") + if not (checkout / "sforge/__init__.py").is_file(): + raise ValueError("Selected source does not contain the SForge package") + return receipt + + +def prepare_source(log_dir: Path, *, repository: str | None = None, + ref: str | None = None, revision: str | None = None) -> tuple[Path, dict]: + """Resolve once; an existing cohort can be reused offline, never refreshed.""" + if revision is not None and not re.fullmatch(r"[0-9a-f]{40}", revision): + raise ValueError("--sforge-revision requires a full lowercase commit SHA") + if ref is not None and revision is not None: + raise ValueError("Select either --sforge-ref or --sforge-revision") + if ref is not None and (not ref or ref.startswith("-")): + raise ValueError("--sforge-ref must be a nonempty Git reference") + if repository is not None and ref is None and revision is None: + raise ValueError("An alternate repository requires --sforge-ref or --sforge-revision") + log_dir = log_dir.resolve() + log_dir.mkdir(parents=True, exist_ok=True) + state = log_dir / STATE_DIR + with (log_dir / f"{STATE_DIR}.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + if state.exists(): + receipt = _validate(state) + for key, requested in (("repository", repository), ("ref", ref), ("commit", revision)): + if requested is not None and receipt[key] != requested: + raise ValueError("EdgeBench selection conflicts with this cohort; use a new log directory") + return state / "checkout", receipt + if (log_dir / "runs").exists() and any((log_dir / "runs").iterdir()): + raise ValueError("Existing runs have no source lock; preserve them and use a new log directory") + defaults = json.loads(DEFAULT_FILE.read_text()) + repository = repository or defaults["repository"] + ref = ref or (revision if revision else defaults["ref"]) + with tempfile.TemporaryDirectory(prefix=".edgebench-source-", dir=log_dir) as temporary: + staged = Path(temporary) / "source" + checkout = staged / "checkout" + checkout.mkdir(parents=True) + _git(checkout, "init", "--quiet") + _git(checkout, "remote", "add", "origin", repository) + _git(checkout, "fetch", "--quiet", "--depth=1", "--", "origin", revision or ref) + commit = _git(checkout, "rev-parse", "FETCH_HEAD^{commit}") + if revision is not None and commit != revision: + raise ValueError("Fetched EdgeBench commit differs from the requested revision") + _git(checkout, "checkout", "--quiet", "--detach", commit) + receipt = dict(schema_version="edgebench_source_v1", repository=repository, + ref=ref, commit=commit) + (staged / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n") + _validate(staged) + staged.rename(state) + return state / "checkout", receipt + + +def bootstrap(*, keep_log_arg: bool = True) -> None: + """CLI-only bootstrap; imported Python APIs retain their explicit environment.""" + parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False, + description="Select and freeze the shared EdgeBench source for a log directory.") + parser.add_argument("--log-dir", type=Path, default=os.environ.get("SFORGE_LOG_DIR")) + parser.add_argument("--sforge-repository") + parser.add_argument("--sforge-ref") + parser.add_argument("--sforge-revision") + if "--help" in sys.argv[1:] or "-h" in sys.argv[1:]: + parser.print_help() + return + args, remaining = parser.parse_known_args() + if args.log_dir is None: + parser.error("provide --log-dir or SFORGE_LOG_DIR to freeze the EdgeBench source") + try: + checkout, _ = prepare_source(args.log_dir, repository=args.sforge_repository, + ref=args.sforge_ref, revision=args.sforge_revision) + except (OSError, ValueError, RuntimeError) as error: + parser.error(str(error)) + sys.path.insert(0, str(checkout)) + os.environ["PYTHONPATH"] = os.pathsep.join(filter(None, (str(checkout), os.environ.get("PYTHONPATH")))) + os.environ["SFORGE_LOG_DIR"] = str(args.log_dir.resolve()) + os.environ[RECEIPT_ENV] = str(checkout.parent / "receipt.json") + sys.argv[1:] = remaining + (["--log-dir", str(args.log_dir)] if keep_log_arg else []) + active_source_receipt() + + +def active_source_receipt() -> dict | None: + """Prove the module Python will load belongs to the frozen source.""" + path = os.environ.get(RECEIPT_ENV) + if not path: + return None # Legacy imported API: the embedding caller owns its source. + state = Path(path).resolve().parent + receipt = _validate(state) + spec = importlib.util.find_spec("sforge") + if spec is None or not spec.origin or Path(spec.origin).resolve() != state / "checkout/sforge/__init__.py": + raise ValueError("Imported SForge differs from the experiment's source receipt") + return receipt + + +if __name__ == "__main__": + # Prepare dependencies without launching a worker, judge or model call. + bootstrap(keep_log_arg=False) + if "--help" not in sys.argv[1:] and "-h" not in sys.argv[1:]: + print(json.dumps(active_source_receipt(), indent=2)) diff --git a/benchmark/tests/test_edgebench_source.py b/benchmark/tests/test_edgebench_source.py new file mode 100644 index 0000000000..3b610c88ea --- /dev/null +++ b/benchmark/tests/test_edgebench_source.py @@ -0,0 +1,180 @@ +"""Source selection is cohort-wide, explicit, and independent of installed SForge.""" +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +import json +import os +import subprocess +import sys + +import pytest + +from benchmark.edgebench import source + + +def git(path, *args): + return subprocess.check_output(["git", "-C", str(path), *args], text=True).strip() + + +@pytest.fixture +def repository(tmp_path, monkeypatch): + repo = tmp_path / "remote" + repo.mkdir() + git(repo, "init", "-q", "-b", "integration") + git(repo, "config", "user.name", "Fixture") + git(repo, "config", "user.email", "fixture@example.invalid") + # Synthetic provider: the serve CLI identifies the imported source, without Docker. + (repo / "sforge").mkdir() + (repo / "sforge/__init__.py").write_text("MARKER = 'shared-source'\n") + (repo / "sforge/cli.py").write_text( + "def main():\n" + " import json, sys, sforge\n" + " from benchmark.edgebench.source import active_source_receipt\n" + " print(json.dumps({'marker':sforge.MARKER,'source':active_source_receipt(),'args':sys.argv[1:]}))\n") + (repo / ".gitignore").write_text("__pycache__/\n") + git(repo, "add", ".") + git(repo, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "synthetic source") + defaults = tmp_path / "defaults.json" + defaults.write_text(json.dumps({"repository": str(repo), "ref": "refs/heads/integration"})) + monkeypatch.setattr(source, "DEFAULT_FILE", defaults) + return repo + + +def test_default_freezes_cohort_while_new_cohort_sees_branch_advance(repository, tmp_path): + first = git(repository, "rev-parse", "HEAD") + checkout, receipt = source.prepare_source(tmp_path / "old") + assert receipt["commit"] == first + assert (checkout / "sforge/__init__.py").read_text() == "MARKER = 'shared-source'\n" + (repository / "new.txt").write_text("second revision") + git(repository, "add", "new.txt") + git(repository, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "advance") + second = git(repository, "rev-parse", "HEAD") + assert source.prepare_source(tmp_path / "old")[1] == receipt + assert source.prepare_source(tmp_path / "new")[1]["commit"] == second != first + + +def test_frozen_cohort_reuses_source_without_network(repository, tmp_path, monkeypatch): + expected = source.prepare_source(tmp_path / "logs")[1] + original = source._git + + def offline(checkout, *args): + assert args[0] not in {"fetch", "clone", "ls-remote"} + return original(checkout, *args) + + monkeypatch.setattr(source, "_git", offline) + assert source.prepare_source(tmp_path / "logs")[1] == expected + + +def test_parallel_worker_and_judge_share_one_source(repository, tmp_path): + with ThreadPoolExecutor(max_workers=4) as pool: + results = list(pool.map(lambda _: source.prepare_source(tmp_path / "logs")[1], range(4))) + assert all(result == results[0] for result in results) + + +def test_explicit_pin_overrides_default_but_cannot_change_existing_cohort(repository, tmp_path): + revision = git(repository, "rev-parse", "HEAD") + log = tmp_path / "logs" + _, receipt = source.prepare_source(log, repository=str(repository), revision=revision) + assert receipt["ref"] == receipt["commit"] == revision + with pytest.raises(ValueError, match="conflicts"): + source.prepare_source(log, revision="f" * 40) + with pytest.raises(ValueError, match="conflicts"): + source.prepare_source(log, repository="https://example.invalid/other.git", ref=revision) + assert source.prepare_source(log)[1] == receipt + + +@pytest.mark.parametrize("change", ["tracked", "untracked", "head", "receipt", "repository", "receipt_type"]) +def test_source_drift_rejected(repository, tmp_path, change): + checkout, _ = source.prepare_source(tmp_path / "logs") + if change == "tracked": + (checkout / "sforge/__init__.py").write_text("MARKER='changed'\n") + elif change == "untracked": + (checkout / "sforge/injected.py").write_text("injected=True\n") + elif change == "head": + git(checkout, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", + "-c", "core.hooksPath=/dev/null", "commit", "--allow-empty", "-qm", "changed head") + else: + path = checkout.parent / "receipt.json" + receipt = json.loads(path.read_text()) + if change == "receipt_type": + receipt = [] + elif change == "repository": + receipt["repository"] = "https://example.invalid/different.git" + else: + receipt["commit"] = "not-a-sha" + path.write_text(json.dumps(receipt)) + with pytest.raises(ValueError): + source.prepare_source(tmp_path / "logs") + + +def test_legacy_runs_cannot_be_silently_relabelled(repository, tmp_path): + (tmp_path / "logs/runs/existing").mkdir(parents=True) + with pytest.raises(ValueError, match="Existing runs"): + source.prepare_source(tmp_path / "logs") + assert not (tmp_path / "logs" / source.STATE_DIR).exists() + + +def test_failed_fetch_leaves_no_source_receipt_and_can_retry(repository, tmp_path): + with pytest.raises(RuntimeError, match="fetch failed"): + source.prepare_source(tmp_path / "logs", ref="refs/heads/missing") + assert not (tmp_path / "logs" / source.STATE_DIR).exists() + assert source.prepare_source(tmp_path / "logs")[1]["commit"] == git(repository, "rev-parse", "HEAD") + + +def test_real_serve_entry_loads_selected_provider_over_ambient_pythonpath(repository, tmp_path): + shadow = tmp_path / "ambient/sforge" + shadow.mkdir(parents=True) + (shadow / "__init__.py").write_text("raise AssertionError('wrong installed provider')\n") + env = {**os.environ, "PYTHONPATH": str(shadow.parent)} + env.pop(source.RECEIPT_ENV, None) + cmd = [sys.executable, "-m", "benchmark.edgebench.serve", "--log-dir", str(tmp_path / "logs"), + "--sforge-repository", str(repository), "--sforge-ref", "refs/heads/integration", "--port", "8088"] + result = subprocess.run(cmd, env=env, capture_output=True, text=True, check=True) + observed = json.loads(result.stdout) + assert observed["marker"] == "shared-source" + assert observed["source"]["commit"] == git(repository, "rev-parse", "HEAD") + assert observed["args"] == ["serve", "--port", "8088"] + + +def test_imported_wrong_provider_rejected(repository, tmp_path, monkeypatch): + checkout, _ = source.prepare_source(tmp_path / "logs") + monkeypatch.setenv(source.RECEIPT_ENV, str(checkout.parent / "receipt.json")) + from types import SimpleNamespace + monkeypatch.setattr(source.importlib.util, "find_spec", lambda _: SimpleNamespace(origin="unrelated.py")) + with pytest.raises(ValueError, match="Imported SForge differs"): + source.active_source_receipt() + + +def test_annotated_tag_resolves_to_commit(repository, tmp_path): + git(repository, "tag", "-a", "release", "-m", "synthetic release") + _, receipt = source.prepare_source(tmp_path / "logs", ref="refs/tags/release") + assert receipt["commit"] == git(repository, "rev-parse", "HEAD") + + +@pytest.mark.parametrize("selection", [ + {"ref": "--upload-pack=unexpected"}, {"ref": ""}, {"revision": "short"}, + {"ref": "main", "revision": "f" * 40}, {"repository": "https://example.invalid/repo"}, +]) +def test_invalid_selection_rejected_before_creating_cohort(tmp_path, selection): + with pytest.raises(ValueError): + source.prepare_source(tmp_path / "logs", **selection) + assert not (tmp_path / "logs").exists() + + +def test_transport_timeout_omits_command_and_credentials(tmp_path, monkeypatch): + def timeout(*args, **kwargs): + raise subprocess.TimeoutExpired(["git", "https://user:secret@example.invalid/repo"], 180) + + monkeypatch.setattr(source.subprocess, "run", timeout) + with pytest.raises(RuntimeError, match="^EdgeBench git fetch timed out$"): + source._git(tmp_path, "fetch", "origin", "main") + + +def test_prepare_help_does_not_need_provider_or_create_source(tmp_path): + result = subprocess.run( + [sys.executable, "-m", "benchmark.edgebench.source", "--help", "--log-dir", str(tmp_path / "logs")], + capture_output=True, text=True, check=True, + ) + assert "--sforge-revision" in result.stdout + assert "null" not in result.stdout + assert not (tmp_path / "logs").exists()