diff --git a/benchmark/README.md b/benchmark/README.md index 855bac62d2..56ef85c300 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -396,30 +396,61 @@ python -m benchmark.edgebench.checkpoint_judge --checkpoint-capture \ --tasks-dir /path/to/tasks --log-dir /path/to/private-run-logs ``` -The supported public checkpoint provider is the compatibility fork at -[`071a884012754127f28f5b84be10f533a42db527`](https://github.com/loopx-agent/EdgeBench/commit/071a884012754127f28f5b84be10f533a42db527). -It contains `sforge.harness.checkpoint` and the native -`create_app(..., checkpoints=True)` seam used by the controller below. The earlier -`0e20fb499ee248f012d6467c00b3dec00a29fe85` compatibility pin and upstream -`sforge==1.1.0` do not contain that API and cannot qualify joint continuation. -Enabling capture against an unsupported provider fails before adding routes. -This is pinned proposed source, not an upstream release. From this LoopX checkout, -prepare an isolated Python 3.12 environment and verify the imported API: +Select the checkpoint provider through the same +[shared EdgeBench source protocol](edgebench/README.md#select-and-freeze-the-native-provider) +as the workers and judge. A new qualification cohort resolves the canonical +integration branch and applies the reviewed native compatibility patch under +that protocol; its `.edgebench-source/receipt.json` records the effective provider +commit, upstream revision and patch digest. Existing cohorts remain frozen. +Patch representation changes receive a new digest and effective commit even +when the provider tree is unchanged; the original frozen bytes and provenance +remain independently verifiable for an older cohort. +An explicit source selection uses unmodified provider code and may not include +the fix; follow the linked protocol when qualifying or replaying a known source. + +The provider must expose `sforge.harness.checkpoint` and native +`create_app(..., checkpoints=True)`. API presence alone does not qualify +continuation: a released result with a non-finite score or nested metric must +remain capturable without changing native ranking, history or consumed rounds. +The controller transports native v1 and v2 manifests and verifies their bound +bytes; the provider owns non-finite encoding and restoration. New default cohorts +receive this fix through the bundled compatibility patch. Updating the LoopX +runner alone does not change an older cohort's frozen provider. +Run the HTTP continuity regression below against the selected source before +using it for a new cohort; an older provider can expose the API and still fail +this regression. From this LoopX checkout: ```sh CKPT_PROVIDER_ROOT=$(mktemp -d) -git clone https://github.com/loopx-agent/EdgeBench.git "$CKPT_PROVIDER_ROOT/provider" -git -C "$CKPT_PROVIDER_ROOT/provider" checkout --detach 071a884012754127f28f5b84be10f533a42db527 +python -m benchmark.edgebench.source --log-dir "$CKPT_PROVIDER_ROOT/cohort" uv venv --python 3.12 "$CKPT_PROVIDER_ROOT/venv" uv pip install --python "$CKPT_PROVIDER_ROOT/venv/bin/python" \ - -e '.[test]' -e "$CKPT_PROVIDER_ROOT/provider" 'httpx>=0.28,<1' -"$CKPT_PROVIDER_ROOT/venv/bin/python" -c 'import inspect,sforge,loopx,sforge.harness.checkpoint; from sforge.harness.judge_server import create_app; assert "checkpoints" in inspect.signature(create_app).parameters; print(sforge.__file__); print(loopx.__file__)' -"$CKPT_PROVIDER_ROOT/venv/bin/python" -m pytest -q benchmark/tests/test_checkpoint_judge.py + -e '.[test]' -e "$CKPT_PROVIDER_ROOT/cohort/.edgebench-source/checkout" 'httpx>=0.28,<1' +LOOPX_EDGEBENCH_SOURCE_RECEIPT="$CKPT_PROVIDER_ROOT/cohort/.edgebench-source/receipt.json" \ + "$CKPT_PROVIDER_ROOT/venv/bin/python" - <<'PY' +import inspect +import json +import sforge.harness.checkpoint +from benchmark.edgebench.source import active_source_receipt +from sforge.harness.judge_server import create_app +assert "checkpoints" in inspect.signature(create_app).parameters +print(json.dumps(active_source_receipt(), indent=2)) +PY +LOOPX_EDGEBENCH_SOURCE_RECEIPT="$CKPT_PROVIDER_ROOT/cohort/.edgebench-source/receipt.json" \ + "$CKPT_PROVIDER_ROOT/venv/bin/python" -m pytest -q \ + benchmark/tests/test_checkpoint_judge.py \ + benchmark/tests/test_checkpoint_controller.py::test_native_controller_transport_preserves_original_point_and_fresh_authority \ + benchmark/tests/test_checkpoint_controller.py::test_native_controller_transport_retains_nonfinite_released_results # Linux with an owned disposable Docker daemon; grades only synthetic fixtures. LOOPX_EDGEBENCH_DOCKER_SMOKE=1 "$CKPT_PROVIDER_ROOT/venv/bin/python" -m pytest -q \ benchmark/tests/test_edgebench_feedback_docker.py::test_checkpoint_opportunity_uses_native_admission_and_real_grade ``` +The HTTP regression uses synthetic grading and a synthetic backend, with the real +native HTTP service and LoopX capture/export/restore transport. It retains a +finite incumbent beside a partial non-finite result, preserves nested native +values and consumed opportunities, and accepts the next eligible submission +without mutating the source. It does not run a model, Docker or benchmark task. The Docker journey checks a real grade, remaining rounds/cooldown, incumbent bytes, and the cold-registration counterexample; it does not restore evaluator authority. Native session qualification is separate: the tested Host is `codex-cli 0.160.0`. diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index 01179b263e..b094e70260 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -13,13 +13,50 @@ 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/checkout` and atomically freezes the effective source 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. +Later entries reuse the frozen checkout without fetching, even if the branch, +repository default, or bundled compatibility patch 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 source; +never repin an active or historical cohort. + +For a **new cohort with no explicit source options**, `source-default.json` +selects the shared branch and the versioned +[`nonfinite-checkpoint-v1` compatibility patch](compatibility/nonfinite-checkpoint-v1.patch). +This LoopX-owned patch fixes the native checkpoint representation of nonfinite +released scores and metrics. Native scoring, report values and selection remain +unchanged. Preparation verifies the asset digest and each touched file's exact +upstream blob and mode, applies the patch before provider import/installation, +and creates a clean, reproducible Git commit with the fetched commit as its +parent. The snapshot uses an explicit LoopX compatibility identity and the +upstream commit time; it does not attribute the patch to an upstream author. +Unrelated upstream changes can advance normally. A changed patch preimage, +including a later upstream implementation of the fix, stops preparation without +publishing a source directory. Review and update the bundled patch, or explicitly +select the original source, before starting that new cohort. + +Patched sources use `edgebench_source_v2` receipts. `upstream_commit` is the +fetched remote revision; `compatibility_patch` records the patch ID and SHA-256; +`commit` is the **effective local derived revision** actually imported and bound +by native checkpoint provenance. This derived SHA is not an upstream revision +that a fresh remote fetch can retrieve. The frozen `compatibility.patch` copy, +Git parent/tree, complete upstream-to-effective diff, and deterministic commit +contents are cross-checked on reuse. Verification uses the frozen copy rather +than the current LoopX asset, so later asset changes do not invalidate an older +cohort. Preserve the complete `.edgebench-source` directory for offline reuse or +exact patched-source replay. Older `edgebench_source_v1` cohorts retain their +original unmodified source and receipt; preparation never upgrades them. + +The bundled patch uses one-line unified context and Git's +`diff.suppressBlankEmpty=true` representation. This keeps its full-index file +identities and changed content readable while avoiding unrelated example text +and trailing-space context markers in the containing repository. Validation +accepts that exact canonical form or the original three-line, space-prefixed +context form retained by earlier v2 cohorts; it does not ignore arbitrary +whitespace. Changing the representation changes its digest and derived commit, +even when the provider tree is identical. Existing cohorts keep their original +patch bytes, receipt and commit and remain verifiable offline. Prepare dependencies in an isolated environment before starting a study: @@ -51,8 +88,14 @@ 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. +**Any explicit repository, ref, or revision selects unmodified source for a new +cohort**, even when it names the default repository/branch. These selections use +the original v1 receipt contract and do not apply the bundled patch. Use an +explicit upstream SHA to replay an unmodified known revision; retain its native +checkpoint limitations. For an already frozen cohort, options only assert that +its existing selection matches; they cannot apply/remove a patch or change its +commit. 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 diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py index 5e0c26a2cf..5b58460d7f 100644 --- a/benchmark/edgebench/checkpoint_controller.py +++ b/benchmark/edgebench/checkpoint_controller.py @@ -90,7 +90,7 @@ def _bound_bank(folder: Path, reference: dict) -> dict: witnessed = _physical(folder / "manifest.sha256").read_text() packet = json.loads(raw) if (digest != reference["manifest_sha256"] or witnessed != digest or - packet.get("schema_version") != "sforge_judge_checkpoint_v1" or + packet.get("schema_version") not in {"sforge_judge_checkpoint_v1", "sforge_judge_checkpoint_v2"} or any(packet.get(key) != reference[key] for key in ( "task_id", "source_run_id", "opportunity", "incumbent"))): raise ValueError("Native bank bytes differ from the captured reference") diff --git a/benchmark/edgebench/compatibility/nonfinite-checkpoint-v1.patch b/benchmark/edgebench/compatibility/nonfinite-checkpoint-v1.patch new file mode 100644 index 0000000000..d65f4f1dda --- /dev/null +++ b/benchmark/edgebench/compatibility/nonfinite-checkpoint-v1.patch @@ -0,0 +1,476 @@ +diff --git a/docs/en/reference/judge-api.md b/docs/en/reference/judge-api.md +index 3b2b23fcf6b29770905be3cf24669cc83f0b144a..4209f8c6ac20c1de5c79495e56ad096649eb3499 100644 +--- a/docs/en/reference/judge-api.md ++++ b/docs/en/reference/judge-api.md +@@ -83,2 +83,19 @@ Content-Type: application/json + ++Capsule manifests remain strict JSON. Finite history and reports keep the ++`sforge_judge_checkpoint_v1` representation. A report can legitimately retain a ++nonfinite score or metric; capsules containing those values use ++`sforge_judge_checkpoint_v2`. That representation replaces only the nonfinite ++numbers in `history` and `results` with `null`, and records each original value ++in the top-level `nonfinite_values` list, for example ++`{"path":["history",0,"score"],"kind":"positive_infinity"}`. Kinds are ++`positive_infinity`, `negative_infinity`, and `nan`; path components are exact ++object keys or nonnegative array indices. This is storage encoding, not missing ++feedback: the reader restores the original numbers before checking history, ++results, or the native-selected incumbent. Literal strings and similarly shaped ++objects in reports are ordinary data. Unknown versions, duplicate or unavailable ++paths, non-null targets, paths outside released data, and unmarked nonfinite ++numbers are rejected. Earlier readers reject v2 rather than treating its nulls ++as scores. New readers retain finite v1 format support; the existing exact ++native-source/evaluator binding still applies to both versions. ++ + ```http +@@ -92,2 +109,8 @@ The response issues a fresh submission token and feedback relay for `fresh-branc + ++Restored `report.json` files retain the ordinary native report format used by ++evaluation, including Python JSON's `Infinity`, `-Infinity`, and `NaN` spellings ++when present. They do not contain the capsule's path encoding. Live reports, ++selection, and public API projections keep their existing semantics; neither ++capture nor restore converts a nonfinite score to zero or missing feedback. ++ + If the restore response is lost, call `POST /api/v1/checkpoint/restore-result` with the **same** JSON body (`checkpoint_id`, `manifest_sha256`, `run_id`, and current `admin_secret`). It returns the original response shape and the already published submission token. It does not allocate another registration, replay grading, or restart cooldown. Readback also remains valid after subsequent submissions and preserves their consumed opportunities. The existing origin marker must match and the registration must still be live; a missing, mismatched, partial or lost registration returns 409. Wrong admin authority returns 403, and malformed bodies return 422. A persisted origin marker alone cannot recreate credentials after a judge process restart. This recovery does not recover a runner's phase deadline: its controller must retain the original activation-time deadline and must not initialize a new duration on retry. A new `run_id` represents a different branch, not recovery of this response. +diff --git a/docs/zh/reference/judge-api.md b/docs/zh/reference/judge-api.md +index 19f9be961ddcf385b0639986c5434601318e2fad..6cb4d4bf36748ec212211f57b234ab4a8e2887e4 100644 +--- a/docs/zh/reference/judge-api.md ++++ b/docs/zh/reference/judge-api.md +@@ -77,2 +77,20 @@ Token 是一个 32 字符的十六进制字符串，内部维护任务 ID、运 + ++Capsule 的 manifest 始终使用严格 JSON。有限分数和指标继续使用 ++`sforge_judge_checkpoint_v1`。历史或报告中存在非有限数值时，使用 ++`sforge_judge_checkpoint_v2`：只将 `history`、`results` 中对应数值存为 `null`， ++并在顶层 `nonfinite_values` 列表记录原值，例如 ++`{"path":["history",0,"score"],"kind":"positive_infinity"}`。 ++kind 仅支持 `positive_infinity`、`negative_infinity`、`nan`；path 的每一项必须 ++是确切的对象键或非负数组下标。这只是存储表示，不是缺失反馈。读取时先还原 ++原始数值，再核对历史、结果和原生选优。报告中的同名字符串、相似形状的字典 ++仍是普通数据。未知版本、重复或不存在的路径、非 null 目标、越过历史/结果 ++范围的路径及未标记的非有限数字均拒绝。旧 reader 会拒绝 v2，不会将占位 null ++当成分数；新 reader 继续支持有限 v1 格式，但两种格式均保留原有精确源码/ ++评测器绑定要求。 ++ ++恢复的 `report.json` 沿用正常评测生成的原生报告格式，保留 Python JSON 的 ++`Infinity`、`-Infinity`、`NaN` 表示，不把 capsule 的路径编码带入普通报告。 ++运行中的报告、选优和公开 API 投影保持原有语义；捕获和恢复不会把非有限分数 ++改成零或缺失反馈。 ++ + 恢复不会评测或多消费一轮。后续提交仍由原生 `consume_round`、预算、冷却及选优规则决定。冷却在新注册发布时，从捕获时剩余的**时长**重新计时；离线等待和 capsule 物化既不会补发机会，也不会消耗这段时长；旧 capsule 不包含来源后来取得的提升。 +diff --git a/sforge/harness/checkpoint.py b/sforge/harness/checkpoint.py +index 31d62ed85f68bea5bd2565f72b3b630d260e8a7a..42e46cd23ea329b099a99d2a32abb3dd9e7020b8 100644 +--- a/sforge/harness/checkpoint.py ++++ b/sforge/harness/checkpoint.py +@@ -45,3 +45,5 @@ if TYPE_CHECKING: + SCHEMA = "sforge_judge_checkpoint_v1" ++NONFINITE_SCHEMA = "sforge_judge_checkpoint_v2" + RESTORE_MARKER = "checkpoint-origin.json" ++_NONFINITE_KINDS = {"positive_infinity": "inf", "negative_infinity": "-inf", "nan": "nan"} + +@@ -81,2 +83,116 @@ def _encode(value: object) -> bytes: + ++def _encode_manifest(packet: dict) -> bytes: ++ """Keep nonfinite result numbers lossless without non-JSON constants.""" ++ values = [] ++ ++ def visit(value, path): ++ if isinstance(value, float) and not math.isfinite(value): ++ kind = "nan" if math.isnan(value) else ( ++ "positive_infinity" if value > 0 else "negative_infinity" ++ ) ++ values.append({"path": path, "kind": kind}) ++ return None ++ if isinstance(value, dict): ++ # Match the existing JSON conversion of int/float/bool/None keys. ++ # Paths address that wire object, never the untouched live dict. ++ pairs = [] ++ for key, item in value.items(): ++ if not isinstance(key, str): ++ key = next(iter(json.loads(_encode({key: None})))) ++ pairs.append((key, item)) ++ if len({key for key, _ in pairs}) != len(pairs): ++ raise CheckpointUnavailable("Checkpoint result keys collide in JSON") ++ return {key: visit(item, [*path, key]) for key, item in sorted(pairs)} ++ if isinstance(value, (list, tuple)): ++ return [visit(item, [*path, index]) for index, item in enumerate(value)] ++ return value ++ ++ # Only released data has this representation. Provenance, resources and ++ # opportunity clocks retain their existing strict numeric validation. ++ data = {key: visit(packet[key], [key]) for key in ("history", "results")} ++ if not values: ++ return _encode(packet) # Existing finite v1 bytes are unchanged. ++ return _encode({**packet, **data, "schema_version": NONFINITE_SCHEMA, ++ "nonfinite_values": values}) ++ ++ ++def _decode_manifest(encoded: bytes) -> dict: ++ """Decode v2 data before native validation; retain finite v1 compatibility.""" ++ def finite_number(text): ++ value = float(text) ++ if not math.isfinite(value): ++ raise CheckpointUnavailable("Unmarked nonfinite checkpoint number") ++ return value ++ ++ def unique_object(pairs): ++ result = {} ++ for key, value in pairs: ++ if key in result: ++ raise CheckpointUnavailable("Duplicate checkpoint object key") ++ result[key] = value ++ return result ++ ++ packet = json.loads(encoded, parse_float=finite_number, ++ parse_constant=finite_number, object_pairs_hook=unique_object) ++ if not isinstance(packet, dict): ++ raise CheckpointUnavailable("Invalid native checkpoint object") ++ if packet.get("schema_version") == SCHEMA: ++ if "nonfinite_values" in packet: ++ raise CheckpointUnavailable("Nonfinite values require checkpoint v2") ++ return packet ++ if packet.get("schema_version") != NONFINITE_SCHEMA: ++ raise CheckpointUnavailable("Unsupported native checkpoint schema") ++ values = packet.pop("nonfinite_values", None) ++ if not isinstance(values, list) or not values: ++ raise CheckpointUnavailable("Checkpoint v2 requires nonfinite value paths") ++ ++ def lookup(container, key): ++ if isinstance(container, dict) and isinstance(key, str) and key in container: ++ return container[key] ++ if (isinstance(container, list) and type(key) is int ++ and 0 <= key < len(container)): ++ return container[key] ++ raise CheckpointUnavailable("Nonfinite checkpoint path is unavailable") ++ ++ seen = set() ++ for entry in values: ++ if not isinstance(entry, dict) or set(entry) != {"path", "kind"}: ++ raise CheckpointUnavailable("Invalid nonfinite checkpoint value") ++ path, kind = entry["path"], entry["kind"] ++ if (not isinstance(path, list) or not path ++ or not all(type(key) is str or (type(key) is int and key >= 0) for key in path) ++ or path[0] not in {"history", "results"} ++ or not isinstance(kind, str) or kind not in _NONFINITE_KINDS): ++ raise CheckpointUnavailable("Invalid nonfinite checkpoint path or kind") ++ identity = tuple(path) ++ if identity in seen: ++ raise CheckpointUnavailable("Duplicate nonfinite checkpoint path") ++ seen.add(identity) ++ container = packet ++ for key in path[:-1]: ++ container = lookup(container, key) ++ if lookup(container, path[-1]) is not None: ++ raise CheckpointUnavailable("Nonfinite checkpoint value must replace null") ++ container[path[-1]] = float(_NONFINITE_KINDS[kind]) ++ # v2 changes only the wire representation, not the native logical contract. ++ packet["schema_version"] = SCHEMA ++ return packet ++ ++ ++def _same_released_value(left, right) -> bool: ++ """NaNs in independent decoded history/report records denote the same value.""" ++ if isinstance(left, float) and isinstance(right, float): ++ if math.isnan(left) and math.isnan(right): ++ return True ++ if isinstance(left, dict) and isinstance(right, dict): ++ return left.keys() == right.keys() and all( ++ _same_released_value(value, right[key]) for key, value in left.items() ++ ) ++ if isinstance(left, list) and isinstance(right, list): ++ return len(left) == len(right) and all( ++ _same_released_value(a, b) for a, b in zip(left, right) ++ ) ++ return left == right ++ ++ + def _sha(raw: bytes) -> str: +@@ -173,3 +289,3 @@ class JudgeCheckpoints: + raise CheckpointUnavailable("Checkpoint manifest digest differs") +- packet = json.loads(encoded) ++ packet = _decode_manifest(encoded) + self._validate(packet) +@@ -253,4 +369,3 @@ class JudgeCheckpoints: + ) +- or entry +- != submission_history_entry( ++ or not _same_released_value(entry, submission_history_entry( + entry["submission_id"], +@@ -261,3 +376,3 @@ class JudgeCheckpoints: + status, +- ) ++ )) + ): +@@ -413,3 +528,3 @@ class JudgeCheckpoints: + self._validate(packet) +- encoded = _encode( ++ encoded = _encode_manifest( + packet +@@ -530,3 +645,8 @@ class JudgeCheckpoints: + folder.mkdir(parents=True) +- _write(folder / "report.json", _encode(report)) ++ # Native report.json uses json.dumps' Inf/NaN spelling ++ # (run_evaluation); preserve that existing reader format. ++ # Only the checkpoint capsule requires strict JSON. ++ _write(folder / "report.json", json.dumps( ++ report, indent=2, ensure_ascii=False ++ ).encode()) + incumbent = packet["incumbent"] +diff --git a/tests/test_judge_checkpoint.py b/tests/test_judge_checkpoint.py +index be90734f9ec7939a8b7a89fafb6532e2a6e1e677..f991fd8c80501e50ab59738f156ced950373ce80 100644 +--- a/tests/test_judge_checkpoint.py ++++ b/tests/test_judge_checkpoint.py +@@ -20,2 +20,3 @@ import copy + import json ++import math + import time +@@ -28,3 +29,5 @@ from fastapi.testclient import TestClient + from sforge.harness import judge_server as native +-from sforge.harness.checkpoint import _encode, _sha ++from sforge.harness.checkpoint import ( ++ CheckpointUnavailable, _decode_manifest, _encode, _encode_manifest, _sha, ++) + from sforge.harness.config import SForgeConfig +@@ -777 +780,244 @@ def test_restore_materialization_does_not_spend_remaining_cooldown( + assert submit(client, fresh, 12).status_code == 429 ++ ++ ++@pytest.mark.parametrize( ++ "score,direction,next_score", [(float("inf"), "minimize", 6), ++ (-float("inf"), "maximize", 8), ++ (float("nan"), "maximize", 8)], ++) ++def test_nonfinite_result_checkpoint_preserves_history_and_opportunities( ++ service, monkeypatch, score, direction, next_score, ++): ++ state, client = service ++ state.tasks["fixture"].judge.score_direction = direction ++ now = [10000.0] ++ monkeypatch.setattr(time, "time", lambda: now[0]) ++ original_grade = native.judge_submission ++ ++ def grade(**kwargs): ++ report = original_grade(**kwargs).to_dict() ++ report["metrics"] = { ++ "bounds": [float("inf"), -float("inf"), float("nan"), None], ++ "literal": {"kind": "nan", "path": ["score"], "value": None}, ++ "text": "Infinity", ++ } ++ return SimpleNamespace(to_dict=lambda: report) ++ ++ token = register(state, client, max_agent_submissions=3, submission_cooldown=30) ++ first = submit(client, token, 7) ++ now[0] += 30 ++ monkeypatch.setattr(native, "judge_submission", grade) ++ second = submit(client, token, score) ++ assert first.status_code == second.status_code == 200 ++ source = copy.deepcopy((state.tokens, state.run_history, state.submissions)) ++ source_result = client.get("/api/v1/result/" + second.json()["submission_id"], ++ params={"token": token}).json() ++ ++ point = capture(state, client, token) ++ assert point["incumbent"]["round"] == "agent-1" ++ assert point["opportunity"]["next_agent"] == 3 ++ assert point["opportunity"]["remaining_cooldown_seconds"] == 30 ++ assert (state.tokens, state.run_history, state.submissions) == source ++ manifest_path = state.config.log_dir / "checkpoints" / point["checkpoint_id"] / "manifest.json" ++ ++ def reject_constant(value): ++ raise AssertionError("Checkpoint manifest contains a non-JSON constant: " + value) ++ ++ manifest = json.loads(manifest_path.read_bytes(), parse_constant=reject_constant) ++ assert manifest["incumbent"]["round"] == "agent-1" ++ now[0] += 600 # Offline time cannot consume a retained cooldown. ++ response = restore(state, client, point) ++ assert response.status_code == 200, response.text ++ fresh = response.json()["token"] ++ retained = state.run_history["branch/fixture"] ++ assert len(retained) == 2 and retained[1]["round"] == "agent-2" ++ restored_score = retained[1]["score"] ++ assert math.isnan(restored_score) if math.isnan(score) else restored_score == score ++ restored_report = state.submissions[retained[1]["submission_id"]]["report"] ++ bounds = restored_report["metrics"]["bounds"] ++ assert bounds[:2] == [float("inf"), -float("inf")] ++ assert math.isnan(bounds[2]) and bounds[3] is None ++ assert restored_report["metrics"]["literal"] == { ++ "kind": "nan", "path": ["score"], "value": None, ++ } ++ # Ordinary native report files keep their established json.loads format; ++ # the checkpoint's private strict-JSON representation does not leak here. ++ report_file = state.config.log_dir / "runs/branch/fixture/submissions/agent-2/report.json" ++ persisted = json.loads(report_file.read_bytes()) ++ assert math.isnan(persisted["score"]) if math.isnan(score) else persisted["score"] == score ++ assert persisted["metrics"]["bounds"][:2] == bounds[:2] ++ assert math.isnan(persisted["metrics"]["bounds"][2]) ++ restored_result = client.get("/api/v1/result/" + retained[1]["submission_id"], ++ params={"token": fresh}).json() ++ assert restored_result["report"] == source_result["report"] ++ assert history(client, fresh)["best_score"] == 7 ++ recaptured = capture(state, client, fresh) ++ assert recaptured["incumbent"] == point["incumbent"] ++ assert recaptured["opportunity"]["next_agent"] == 3 ++ assert submit(client, fresh, next_score).status_code == 429 ++ now[0] += 29 ++ assert submit(client, fresh, next_score).status_code == 429 ++ now[0] += 1 ++ assert submit(client, fresh, next_score).json()["round_id"] == "agent-3" ++ assert history(client, fresh)["best_score"] == next_score ++ assert submit(client, fresh, next_score).status_code == 429 ++ ++ ++def test_manifest_codec_preserves_literal_objects_and_finite_v1_bytes(): ++ literal = {"schema_version": "sforge_judge_checkpoint_v2", "nonfinite_values": [ ++ {"kind": "nan", "path": ["score"], "value": None}, ++ ]} ++ packet = { ++ "schema_version": "sforge_judge_checkpoint_v1", ++ "history": [{"score": 7.0}], ++ "results": {"synthetic": {"report": {"metrics": { ++ "literal": literal, "0": "Infinity", "finite": [None, True, 0, 1.5], ++ }}}}, ++ } ++ finite_bytes = _encode(packet) ++ assert _encode_manifest(packet) == finite_bytes ++ assert _decode_manifest(finite_bytes) == packet ++ packet["results"]["synthetic"]["report"]["metrics"]["numbers"] = [ ++ float("inf"), -float("inf"), float("nan"), ++ ] ++ before = copy.deepcopy(packet) ++ encoded = _encode_manifest(packet) ++ wire = json.loads(encoded, parse_constant=lambda value: pytest.fail(value)) ++ assert wire["schema_version"] == "sforge_judge_checkpoint_v2" ++ assert packet == before # Encoding cannot alter live grading/selection facts. ++ decoded = _decode_manifest(encoded) ++ metrics = decoded["results"]["synthetic"]["report"]["metrics"] ++ assert metrics["literal"] == literal and metrics["0"] == "Infinity" ++ assert metrics["finite"] == [None, True, 0, 1.5] ++ assert metrics["numbers"][:2] == [float("inf"), -float("inf")] ++ assert math.isnan(metrics["numbers"][2]) ++ assert _encode_manifest(decoded) == encoded ++ ++ ++@pytest.mark.parametrize("key", [0, 1.25, False, None]) ++def test_finite_v1_preserves_existing_json_object_key_encoding(key): ++ packet = {"schema_version": "sforge_judge_checkpoint_v1", "history": [], ++ "results": {"synthetic": {"report": {"metrics": {key: "literal"}}}}} ++ assert _encode_manifest(packet) == _encode(packet) ++ ++ ++def test_nonfinite_paths_follow_json_object_keys_without_mutating_source(): ++ metrics = {1: float("inf"), 2: float("nan"), 3: "Infinity"} ++ packet = {"schema_version": "sforge_judge_checkpoint_v1", "history": [], ++ "results": {"synthetic": {"report": {"metrics": metrics}}}} ++ encoded = _encode_manifest(packet) ++ restored = _decode_manifest(encoded)["results"]["synthetic"]["report"]["metrics"] ++ assert restored["1"] == float("inf") and math.isnan(restored["2"]) ++ assert restored["3"] == "Infinity" and set(metrics) == {1, 2, 3} ++ packet["results"]["synthetic"]["report"]["metrics"] = {1: float("inf"), "1": None} ++ with pytest.raises(CheckpointUnavailable, match="collide"): ++ _encode_manifest(packet) ++ ++ ++@pytest.mark.parametrize("fault", [ ++ "unknown_schema", "v1_sidecar", "missing_sidecar", "empty_sidecar", ++ "extra_record_key", "unknown_kind", "duplicate_path", "missing_path", ++ "non_null_target", "authority_path", "boolean_index", "negative_index", ++ "string_index", "empty_path", ++]) ++def test_manifest_codec_rejects_ambiguous_nonfinite_paths(fault): ++ wire = json.loads(_encode_manifest({ ++ "schema_version": "sforge_judge_checkpoint_v1", "captured_at_epoch": None, ++ "history": [{"score": float("inf"), "passed": 1}], "results": {}, ++ })) ++ entry = wire["nonfinite_values"][0] ++ if fault == "unknown_schema": ++ wire["schema_version"] = "sforge_judge_checkpoint_v99" ++ elif fault == "v1_sidecar": ++ wire["schema_version"] = "sforge_judge_checkpoint_v1" ++ elif fault == "missing_sidecar": ++ del wire["nonfinite_values"] ++ elif fault == "empty_sidecar": ++ wire["nonfinite_values"] = [] ++ elif fault == "extra_record_key": ++ entry["value"] = "Infinity" ++ elif fault == "unknown_kind": ++ entry["kind"] = "infinite" ++ elif fault == "duplicate_path": ++ wire["nonfinite_values"].append(copy.deepcopy(entry)) ++ elif fault == "missing_path": ++ entry["path"][-1] = "absent" ++ elif fault == "non_null_target": ++ entry["path"][-1] = "passed" ++ elif fault == "authority_path": ++ entry["path"] = ["captured_at_epoch"] ++ elif fault == "boolean_index": ++ entry["path"][1] = False ++ elif fault == "negative_index": ++ entry["path"][1] = -1 ++ elif fault == "string_index": ++ entry["path"][1] = "0" ++ else: ++ entry["path"] = [] ++ with pytest.raises(CheckpointUnavailable): ++ _decode_manifest(_encode(wire)) ++ ++ ++@pytest.mark.parametrize("raw", [ ++ b'{"schema_version":"sforge_judge_checkpoint_v1","history":[NaN]}', ++ b'{"schema_version":"sforge_judge_checkpoint_v1","history":[Infinity]}', ++ b'{"schema_version":"sforge_judge_checkpoint_v1","history":[-Infinity]}', ++ b'{"schema_version":"sforge_judge_checkpoint_v1","history":[1e999]}', ++ b'{"schema_version":"sforge_judge_checkpoint_v1","history":[],"history":[]}', ++]) ++def test_manifest_codec_rejects_unmarked_numbers_and_duplicate_keys(raw): ++ with pytest.raises(CheckpointUnavailable): ++ _decode_manifest(raw) ++ ++ ++@pytest.mark.parametrize("score", [float("inf"), -float("inf"), float("nan")]) ++def test_all_nonfinite_history_and_capture_retry_retain_native_selection(service, score): ++ state, client = service ++ token = register(state, client, max_agent_submissions=2) ++ submitted = submit(client, token, score) ++ assert submitted.status_code == 200 ++ body = {"token": token, "admin_secret": state.admin_secret, "capture_id": "4" * 32} ++ first = client.post("/api/v1/checkpoint/capture", json=body) ++ assert first.status_code == 200, first.text ++ point = first.json() ++ assert point["incumbent"]["round"] == "agent-1" ++ # A new source submission must not change a retry's original history. ++ assert submit(client, token, 8).status_code == 200 ++ assert client.post("/api/v1/checkpoint/capture", json=body).json() == point ++ restored = restore(state, client, point) ++ assert restored.status_code == 200, restored.text ++ fresh = restored.json()["token"] ++ assert state.resolve_token(fresh)["next_agent"] == 2 ++ retained = state.run_history["branch/fixture"] ++ assert len(retained) == 1 and history(client, fresh)["best_round"] == "agent-1" ++ assert math.isnan(retained[0]["score"]) if math.isnan(score) else retained[0]["score"] == score ++ recaptured = capture(state, client, fresh) ++ assert recaptured["incumbent"] == point["incumbent"] ++ ++ ++@pytest.mark.parametrize("fault", ["history_value", "result_value", "duplicate_path", "unknown_schema"]) ++def test_nonfinite_manifest_tamper_cannot_publish_restore(service, monkeypatch, fault): ++ state, client = service ++ token = register(state, client) ++ submit(client, token, float("nan")) ++ point = capture(state, client, token) ++ path = state.config.log_dir / "checkpoints" / point["checkpoint_id"] / "manifest.json" ++ wire = json.loads(path.read_bytes()) ++ if fault in {"history_value", "result_value"}: ++ root = "history" if fault == "history_value" else "results" ++ entry = next(row for row in wire["nonfinite_values"] if row["path"][0] == root) ++ entry["kind"] = "positive_infinity" ++ elif fault == "duplicate_path": ++ wire["nonfinite_values"].append(copy.deepcopy(wire["nonfinite_values"][0])) ++ else: ++ wire["schema_version"] = "sforge_judge_checkpoint_v99" ++ path.write_bytes(_encode(wire)) ++ point["manifest_sha256"] = _sha(path.read_bytes()) ++ path.with_name("manifest.sha256").write_text(point["manifest_sha256"]) ++ prepare = Mock(side_effect=AssertionError("Invalid capsule cannot prepare authority")) ++ monkeypatch.setattr(state, "_prepare_session", prepare) ++ response = restore(state, client, point) ++ assert response.status_code == 409, response.text ++ prepare.assert_not_called() ++ assert len(state.tokens) == 1 and len(state.submissions) == 1 ++ assert not (state.config.log_dir / "runs/branch").exists() diff --git a/benchmark/edgebench/source-default.json b/benchmark/edgebench/source-default.json index 45510bed9d..ac35af2709 100644 --- a/benchmark/edgebench/source-default.json +++ b/benchmark/edgebench/source-default.json @@ -1,4 +1,9 @@ { "repository": "https://github.com/huangruiteng/EdgeBench.git", - "ref": "refs/heads/codex/edgebench-integration" + "ref": "refs/heads/codex/edgebench-integration", + "compatibility_patch": { + "id": "nonfinite-checkpoint-v1", + "relative_path": "compatibility/nonfinite-checkpoint-v1.patch", + "sha256": "44e5c9b3dacf65aaabfc036e26323aac27effbf933b9ff4477fd9a5e36831ffb" + } } diff --git a/benchmark/edgebench/source.py b/benchmark/edgebench/source.py index 256297e063..177080fe36 100644 --- a/benchmark/edgebench/source.py +++ b/benchmark/edgebench/source.py @@ -7,6 +7,7 @@ import argparse import fcntl +import hashlib import importlib.util import json import os @@ -19,24 +20,154 @@ DEFAULT_FILE = Path(__file__).with_name("source-default.json") STATE_DIR = ".edgebench-source" RECEIPT_ENV = "LOOPX_EDGEBENCH_SOURCE_RECEIPT" +FROZEN_PATCH = "compatibility.patch" +_DIFF_ARGS = ("--no-ext-diff", "--no-textconv", "--binary", "--full-index", + "--no-color", "--no-renames", "--src-prefix=a/", "--dst-prefix=b/", + "--inter-hunk-context=0", "-O/dev/null", + "--diff-algorithm=myers", "--indent-heuristic") +# Keep the original v2 cohort representation verifiable without its current asset. +# New assets use Git's compact context format; neither form relaxes byte equality. +_DIFF_FORMATS = (("true", "--unified=1"), ("false", "--unified=3")) -def _git(checkout: Path, *args: str) -> str: +def _git_bytes(checkout: Path, *args: str, input: bytes | None = None) -> bytes: try: result = subprocess.run(["git", "-C", str(checkout), *args], - capture_output=True, text=True, timeout=180) + capture_output=True, input=input, timeout=180, + env={**os.environ, "GIT_NO_REPLACE_OBJECTS": "1"}) 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() + return result.stdout + + +def _git(checkout: Path, *args: str) -> str: + return _git_bytes(checkout, *args).decode().strip() + + +def _patch_identity(value: object) -> bool: + return (isinstance(value, dict) + and set(value) == {"id", "sha256"} + and isinstance(value["id"], str) + and re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,79}", value["id"]) is not None + and isinstance(value["sha256"], str) + and re.fullmatch(r"[0-9a-f]{64}", value["sha256"]) is not None) + + +def _safe_path(value: object) -> bool: + return (isinstance(value, str) + and re.fullmatch(r"[A-Za-z0-9_.-]+(?:/[A-Za-z0-9_.-]+)*", value) is not None + and all(part not in {".", ".."} and part.lower() != ".git" + for part in value.split("/"))) + + +def _bundled_patch(value: object) -> tuple[dict, bytes]: + if (not isinstance(value, dict) or set(value) != {"id", "sha256", "relative_path"} + or not _safe_path(value.get("relative_path"))): + raise ValueError("Invalid default EdgeBench compatibility patch") + identity = {key: value[key] for key in ("id", "sha256")} + root = DEFAULT_FILE.parent.resolve() + path = root / value["relative_path"] + if not _patch_identity(identity) or path.resolve() != path or not path.is_file(): + raise ValueError("Invalid default EdgeBench compatibility patch") + patch = path.read_bytes() + if hashlib.sha256(patch).hexdigest() != identity["sha256"]: + raise ValueError("Default EdgeBench compatibility patch digest differs") + return identity, patch + + +def _check_preimages(checkout: Path, upstream: str, patch: bytes) -> None: + """Accept a full-index patch only against its exact regular-file preimages.""" + blocks = re.split(rb"(?m)^diff --git ", patch) + if blocks[0] or len(blocks) == 1 or not patch.endswith(b"\n"): + raise ValueError("Invalid EdgeBench compatibility patch format") + paths = set() + for block in blocks[1:]: + header = block.split(b"\n@@", 1)[0].split(b"\nGIT binary patch", 1)[0] + lines = header.splitlines() + names = re.fullmatch(rb"a/(\S+) b/(\S+)", lines[0]) + indexes = re.findall(rb"(?m)^index ([0-9a-f]{40})\.\.([0-9a-f]{40})(?: (100644|100755))?$", header) + if names is None or names[1] != names[2] or len(indexes) != 1: + raise ValueError("Invalid EdgeBench compatibility patch file header") + try: + path = names[1].decode("ascii") + except UnicodeDecodeError: + raise ValueError("Invalid EdgeBench compatibility patch path") from None + if not _safe_path(path) or path in paths: + raise ValueError("Invalid EdgeBench compatibility patch path") + paths.add(path) + before, after, mode = (part.decode() for part in indexes[0]) + entry = _git(checkout, "ls-tree", upstream, "--", path) + if before == "0" * 40: + matches = (not entry and after != "0" * 40 + and re.search(rb"(?m)^new file mode (100644|100755)$", header)) + else: + old_mode = mode or next((line.removeprefix(b"deleted file mode ").decode() + for line in lines if line.startswith(b"deleted file mode ")), "") + matches = (old_mode in {"100644", "100755"} + and entry == f"{old_mode} blob {before}\t{path}") + if not matches: + raise ValueError("EdgeBench compatibility patch preimage differs; use a new reviewed patch " + "or explicitly select an unmodified source") + + +def _commit_payload(checkout: Path, upstream: str, tree: str, identity: dict) -> bytes: + epoch = _git(checkout, "show", "-s", "--format=%ct", upstream) + if not re.fullmatch(r"[0-9]+", epoch): + raise ValueError("Invalid EdgeBench upstream commit time") + author = f"LoopX Compatibility {epoch} +0000" + return (f"tree {tree}\nparent {upstream}\nauthor {author}\ncommitter {author}\n\n" + f"LoopX compatibility: {identity['id']}\n\n" + f"Upstream-Commit: {upstream}\nPatch-SHA256: {identity['sha256']}\n").encode() + + +def _validate_patched(state: Path, receipt: dict) -> None: + identity = receipt.get("compatibility_patch") + if (set(receipt) != {"schema_version", "repository", "ref", "commit", + "upstream_commit", "compatibility_patch"} + or not isinstance(receipt.get("upstream_commit"), str) + or not re.fullmatch(r"[0-9a-f]{40}", receipt["upstream_commit"]) + or not _patch_identity(identity)): + raise ValueError("Invalid patched EdgeBench source receipt") + path = state / FROZEN_PATCH + if path.is_symlink() or not path.is_file(): + raise ValueError("Frozen EdgeBench compatibility patch is missing or changed") + patch = path.read_bytes() + if hashlib.sha256(patch).hexdigest() != identity["sha256"]: + raise ValueError("Frozen EdgeBench compatibility patch digest differs") + checkout, upstream, commit = state / "checkout", receipt["upstream_commit"], receipt["commit"] + _check_preimages(checkout, upstream, patch) + if not any(_git_bytes(checkout, "-c", f"diff.suppressBlankEmpty={blank}", "diff", + *_DIFF_ARGS, context, upstream, commit, "--") == patch + for blank, context in _DIFF_FORMATS): + raise ValueError("Frozen EdgeBench source differs from its compatibility patch") + tree = _git(checkout, "rev-parse", f"{commit}^{{tree}}") + payload = _commit_payload(checkout, upstream, tree, identity) + expected = hashlib.sha1(b"commit " + str(len(payload)).encode() + b"\0" + payload).hexdigest() + if commit != expected or _git_bytes(checkout, "cat-file", "commit", commit) != payload: + raise ValueError("Frozen EdgeBench compatibility commit differs from its provenance") + + +def _apply_patch(state: Path, upstream: str, identity: dict, patch: bytes) -> str: + checkout = state / "checkout" + _check_preimages(checkout, upstream, patch) + _git_bytes(checkout, "apply", "--check", "--index", "--whitespace=error", "-", input=patch) + _git_bytes(checkout, "apply", "--index", "--whitespace=error", "-", input=patch) + tree = _git(checkout, "write-tree") + payload = _commit_payload(checkout, upstream, tree, identity) + # A deterministic dependency snapshot, with no hooks, signing or ambient author identity. + commit = _git_bytes(checkout, "hash-object", "-t", "commit", "-w", "--stdin", input=payload).decode().strip() + _git(checkout, "checkout", "--quiet", "--detach", commit) + (state / FROZEN_PATCH).write_bytes(patch) + return commit 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 receipt.get("schema_version") not in {"edgebench_source_v1", "edgebench_source_v2"} 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] @@ -49,6 +180,10 @@ def _validate(state: Path) -> dict: 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") + if receipt["schema_version"] == "edgebench_source_v2": + _validate_patched(state, receipt) + elif (state / FROZEN_PATCH).exists() or {"upstream_commit", "compatibility_patch"} & receipt.keys(): + raise ValueError("Patched EdgeBench source requires its original v2 receipt") return receipt @@ -63,6 +198,7 @@ def prepare_source(log_dir: Path, *, repository: str | None = None, 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") + default_selection = repository is None and ref is None and revision is None log_dir = log_dir.resolve() log_dir.mkdir(parents=True, exist_ok=True) state = log_dir / STATE_DIR @@ -77,6 +213,8 @@ def prepare_source(log_dir: Path, *, repository: str | None = None, 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()) + compatibility = (_bundled_patch(defaults["compatibility_patch"]) + if default_selection and "compatibility_patch" in defaults else None) 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: @@ -92,6 +230,11 @@ def prepare_source(log_dir: Path, *, repository: str | None = None, _git(checkout, "checkout", "--quiet", "--detach", commit) receipt = dict(schema_version="edgebench_source_v1", repository=repository, ref=ref, commit=commit) + if compatibility is not None: + identity, patch = compatibility + receipt.update(schema_version="edgebench_source_v2", upstream_commit=commit, + compatibility_patch=identity, + commit=_apply_patch(staged, commit, identity, patch)) (staged / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n") _validate(staged) staged.rename(state) diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py index c19fa8b46b..fe6851b638 100644 --- a/benchmark/tests/test_checkpoint_controller.py +++ b/benchmark/tests/test_checkpoint_controller.py @@ -2,8 +2,10 @@ from __future__ import annotations import copy +import hashlib import io import json +import math import os import subprocess import socket @@ -18,6 +20,7 @@ import pytest from benchmark.edgebench import checkpoint_controller as transport +from benchmark.edgebench.source import active_source_receipt @contextmanager @@ -51,6 +54,7 @@ def wait(predicate): @pytest.fixture def controllers(tmp_path, monkeypatch): + active_source_receipt() # A supplied frozen-source receipt must match the real imported provider. pytest.importorskip("sforge.harness.checkpoint", reason="Requires native checkpoint API source pin") from sforge.harness import judge_server as native from sforge.harness.config import SForgeConfig @@ -127,6 +131,19 @@ def test_activation_accepts_only_bound_native_fields_and_valid_optional_relay(): transport._activation_result(altered, body) +def test_native_bank_rejects_an_unknown_schema_even_with_matching_bytes(tmp_path): + packet = {"schema_version": "sforge_judge_checkpoint_v3", "task_id": "fixture", + "source_run_id": "source", "opportunity": {}, "incumbent": None} + raw = json.dumps(packet).encode() + digest = hashlib.sha256(raw).hexdigest() + (tmp_path / "manifest.json").write_bytes(raw) + (tmp_path / "manifest.sha256").write_text(digest) + reference = {key: packet[key] for key in ("task_id", "source_run_id", "opportunity", "incumbent")} + reference["manifest_sha256"] = digest + with pytest.raises(ValueError, match="Native bank bytes differ"): + transport._bound_bank(tmp_path, reference) + + def test_native_controller_transport_preserves_original_point_and_fresh_authority(controllers, tmp_path, capsys): import requests source, destination, first, second, files, grades = controllers @@ -175,6 +192,80 @@ def test_native_controller_transport_preserves_original_point_and_fresh_authorit assert len(grades) == 3 +def test_native_controller_transport_retains_nonfinite_released_results( + controllers, tmp_path, monkeypatch): + from sforge.harness import judge_server as native + + source, destination, first, second, files, grades = controllers + clock = SimpleNamespace(epoch=time.time()) + # Native admission and checkpoint capture share this wall clock. HTTP and + # transport deadlines keep using the real monotonic clock. + monkeypatch.setattr(native.time, "time", lambda: clock.epoch) + for app in (source, destination): + app.state.judge.tasks["fixture"].judge.score_direction = "minimize" + original_grade = native.judge_submission + + def grade(**kwargs): + report = original_grade(**kwargs).to_dict() + if kwargs["archive"] == b"inf": + report.update(valid=False, pass_rate=.5, passed=1, failed=1, total_tests=2, + summary="Synthetic partial result", + metrics={"lower_bound": -math.inf, "residual": math.nan}, + details=[{"name": "synthetic-timeout", "score": math.inf}]) + return SimpleNamespace(to_dict=lambda: report) + + monkeypatch.setattr(native, "judge_submission", grade) + token = register(source, first, max_agent_submissions=3, submission_cooldown=2) + assert submit(first, token, 7).status_code == 200 + clock.epoch += 2.05 + rejected = submit(first, token, math.inf) + assert rejected.status_code == 200 + rejected_id = rejected.json()["submission_id"] + # A released partial result still consumed its native round. It neither + # replaces the finite incumbent nor prevents preserving the whole point. + before = json.dumps((source.state.judge.tokens, source.state.judge.submissions, + source.state.judge.run_history), sort_keys=True) + exported = tmp_path / "nonfinite-point" + point = transport.capture(files[0], token=token, capture_id="e" * 32, + destination=exported) + assert point["native_reference"]["incumbent"]["round"] == "agent-1" + assert (exported / "components/native_judge/incumbent.tar.gz").read_bytes() == b"7" + + def reject_non_json_constant(value): + pytest.fail(f"Native checkpoint contains a non-JSON numeric constant: {value}") + + manifest = exported / "components/native_judge/manifest.json" + json.loads(manifest.read_bytes(), parse_constant=reject_non_json_constant) + restored = transport.restore(files[1], root=exported, expected_sha256=point["sha256"], + run_id="nonfinite-branch") + fresh = restored["token"] + assert fresh != token and len(grades) == 2 # Restoring does not regrade. + history = destination.state.judge.get_run_history("nonfinite-branch", "fixture") + assert history["best_round"] == "agent-1" and history["best_score"] == 7 + assert history["agent_submissions"] == 2 and history["auto_submissions"] == 0 + released = next(entry for entry in history["entries"] if entry["round"] == "agent-2") + assert released["submission_id"] != rejected_id + report = destination.state.judge.submissions[released["submission_id"]]["report"] + assert report["score"] == math.inf and report["valid"] is False + assert report["failed"] == 1 and report["pass_rate"] == .5 + assert report["metrics"]["lower_bound"] == -math.inf + assert math.isnan(report["metrics"]["residual"]) + assert report["details"] == [{"name": "synthetic-timeout", "score": math.inf}] + opportunity = destination.state.judge.tokens[fresh] + assert opportunity["next_agent"] == 3 and opportunity["max_agent_submissions"] == 3 + assert opportunity["submission_cooldown"] == 2 + assert submit(second, fresh, 1).status_code == 429 + clock.epoch += 2.05 + assert submit(second, fresh, 1).status_code == 200 + clock.epoch += 2.05 + assert submit(second, fresh, 0).status_code == 429 + history = destination.state.judge.get_run_history("nonfinite-branch", "fixture") + assert history["agent_submissions"] == 3 and history["best_score"] == 1 + assert history["best_round"] == "agent-3" and len(grades) == 3 + assert json.dumps((source.state.judge.tokens, source.state.judge.submissions, + source.state.judge.run_history), sort_keys=True) == before + + def host_phase_command(tmp_path, monkeypatch): """Real command builder; only installation/inference are outside this oracle.""" pytest.importorskip("harbor") diff --git a/benchmark/tests/test_edgebench_source.py b/benchmark/tests/test_edgebench_source.py index 3b610c88ea..02c8cd835f 100644 --- a/benchmark/tests/test_edgebench_source.py +++ b/benchmark/tests/test_edgebench_source.py @@ -2,6 +2,7 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor +import hashlib import json import os import subprocess @@ -178,3 +179,269 @@ def test_prepare_help_does_not_need_provider_or_create_source(tmp_path): assert "--sforge-revision" in result.stdout assert "null" not in result.stdout assert not (tmp_path / "logs").exists() + + +def install_test_patch(repository, other_changes=None, *, compact=True): + """A real full-index Git patch against the fixture provider, without a remote service.""" + target = repository / "sforge/__init__.py" + target.write_text("MARKER = 'patched-source'\n") + for path, contents in (other_changes or {}).items(): + (repository / path).write_text(contents) + patch = subprocess.check_output([ + "git", "-C", str(repository), "-c", + f"diff.suppressBlankEmpty={'true' if compact else 'false'}", + "diff", "--binary", "--full-index", f"--unified={1 if compact else 3}", + "--no-color", "--no-ext-diff", "--no-textconv", "HEAD", "--", + ]) + git(repository, "checkout", "--", "sforge/__init__.py", *(other_changes or {})) + asset = source.DEFAULT_FILE.parent / "compatibility/fixture-v1.patch" + asset.parent.mkdir(exist_ok=True) + asset.write_bytes(patch) + defaults = json.loads(source.DEFAULT_FILE.read_text()) + defaults["compatibility_patch"] = { + "id": "fixture-v1", "relative_path": "compatibility/fixture-v1.patch", + "sha256": hashlib.sha256(patch).hexdigest(), + } + source.DEFAULT_FILE.write_text(json.dumps(defaults)) + return asset + + +def test_bundled_patch_passes_required_public_checks(tmp_path): + from loopx.contract import scan_public_boundary + + defaults = json.loads(source.DEFAULT_FILE.read_text()) + _, patch = source._bundled_patch(defaults["compatibility_patch"]) + asset = tmp_path / "compatibility.patch" + asset.write_bytes(patch) + git(tmp_path, "init", "-q") + git(tmp_path, "add", "--", asset.name) + assert git(tmp_path, "diff", "--cached", "--check") == "" + assert scan_public_boundary([asset], registry={})["ok"] + + +def test_default_patch_creates_reproducible_clean_source(repository, tmp_path, monkeypatch): + asset = install_test_patch(repository) + upstream = git(repository, "rev-parse", "HEAD") + first, receipt = source.prepare_source(tmp_path / "one") + # The dependency commit must not borrow the operator's identity or clock. + monkeypatch.setenv("GIT_AUTHOR_NAME", "Unrelated Operator") + monkeypatch.setenv("GIT_COMMITTER_EMAIL", "operator@example.invalid") + monkeypatch.setenv("GIT_COMMITTER_DATE", "2000-01-01T00:00:00Z") + second, again = source.prepare_source(tmp_path / "two") + assert receipt == again + assert receipt["schema_version"] == "edgebench_source_v2" + assert receipt["upstream_commit"] == upstream != receipt["commit"] + assert receipt["compatibility_patch"] == { + "id": "fixture-v1", "sha256": hashlib.sha256(asset.read_bytes()).hexdigest(), + } + for checkout in (first, second): + assert git(checkout, "status", "--porcelain", "--untracked-files=normal") == "" + assert git(checkout, "rev-parse", "HEAD") == receipt["commit"] + assert git(checkout, "rev-parse", "HEAD^") == upstream + assert git(checkout, "show", "-s", "--format=%an <%ae>") == ( + "LoopX Compatibility ") + assert git(checkout, "show", "-s", "--format=%ct") == git(repository, "show", "-s", "--format=%ct") + assert (checkout / "sforge/__init__.py").read_text() == "MARKER = 'patched-source'\n" + assert (checkout.parent / "compatibility.patch").read_bytes() == asset.read_bytes() + + +def test_patch_accepts_only_untouched_preimages_as_default_branch_advances(repository, tmp_path): + install_test_patch(repository) + original = source.prepare_source(tmp_path / "old")[1] + (repository / "unrelated.txt").write_text("unrelated branch change\n") + git(repository, "add", "unrelated.txt") + git(repository, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "unrelated change") + checkout, advanced = source.prepare_source(tmp_path / "new") + assert advanced["upstream_commit"] == git(repository, "rev-parse", "HEAD") + assert advanced["commit"] != original["commit"] + assert advanced["compatibility_patch"] == original["compatibility_patch"] + assert (checkout / "unrelated.txt").read_text() == "unrelated branch change\n" + assert source.prepare_source(tmp_path / "old")[1] == original + + # A context patch could still apply; changing any touched blob must fail before apply. + with (repository / "sforge/__init__.py").open("a") as target: + target.write("OTHER = True\n") + git(repository, "add", "sforge/__init__.py") + git(repository, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "changed preimage") + with pytest.raises(ValueError, match="preimage differs"): + source.prepare_source(tmp_path / "incompatible") + assert not (tmp_path / "incompatible" / source.STATE_DIR).exists() + assert not list((tmp_path / "incompatible").glob(".edgebench-source-*")) + + +@pytest.mark.parametrize("compact", [False, True]) +def test_canonical_patch_verification_ignores_operator_diff_presentation(repository, tmp_path, monkeypatch, compact): + lines = [f"# line {index}\n" for index in range(40)] + lines[1] = "\n" + (repository / "sforge/long.py").write_text("".join(lines)) + git(repository, "add", "sforge/long.py") + git(repository, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "separate patch hunks") + lines[2], lines[30] = "# first change\n", "# distant change\n" + install_test_patch(repository, {"sforge/long.py": "".join(lines)}, compact=compact) + expected = source.prepare_source(tmp_path / "one")[1] + order = tmp_path / "operator-diff-order" + order.write_text("sforge/long.py\nsforge/__init__.py\n") + settings = {"diff.interHunkContext": "40", "diff.suppressBlankEmpty": "true", + "diff.orderFile": str(order)} + monkeypatch.setenv("GIT_CONFIG_COUNT", str(len(settings))) + for index, (key, value) in enumerate(settings.items()): + monkeypatch.setenv(f"GIT_CONFIG_KEY_{index}", key) + monkeypatch.setenv(f"GIT_CONFIG_VALUE_{index}", value) + assert source.prepare_source(tmp_path / "one")[1] == expected + assert source.prepare_source(tmp_path / "two")[1] == expected + + +def test_represented_patch_preserves_tree_and_existing_offline_cohort(repository, tmp_path, monkeypatch): + original = "# first\n\n# before\n# old\n# after\n\n# last\n" + target = repository / "sforge/representation.py" + target.write_text(original) + git(repository, "add", "sforge/representation.py") + git(repository, "-c", "core.hooksPath=/dev/null", "commit", "-qm", "context fixture") + changes = {"sforge/representation.py": original.replace("# old\n", "# new\n")} + asset = install_test_patch(repository, changes, compact=False) + legacy_patch = asset.read_bytes() + legacy_checkout, legacy = source.prepare_source(tmp_path / "legacy") + legacy_receipt = (legacy_checkout.parent / "receipt.json").read_bytes() + + install_test_patch(repository, changes) + assert asset.read_bytes() != legacy_patch + assert b"\n \n" in legacy_patch and b"\n \n" not in asset.read_bytes() + checkout, current = source.prepare_source(tmp_path / "current") + assert current["upstream_commit"] == legacy["upstream_commit"] + assert current["commit"] != legacy["commit"] + assert current["compatibility_patch"]["sha256"] != legacy["compatibility_patch"]["sha256"] + assert git(checkout, "rev-parse", "HEAD^{tree}") == git(legacy_checkout, "rev-parse", "HEAD^{tree}") + + original_git = source._git_bytes + + def offline(checkout, *args, **kwargs): + assert args[0] not in {"fetch", "clone", "ls-remote"} + return original_git(checkout, *args, **kwargs) + + monkeypatch.setattr(source, "_git_bytes", offline) + asset.unlink() + source.DEFAULT_FILE.unlink() + for log, expected in (("legacy", legacy), ("current", current)): + assert source.prepare_source(tmp_path / log)[1] == expected + assert (legacy_checkout.parent / "receipt.json").read_bytes() == legacy_receipt + assert (legacy_checkout.parent / "compatibility.patch").read_bytes() == legacy_patch + + +@pytest.mark.parametrize("change", ["digest", "format", "noncanonical", "path", "symlink"]) +def test_invalid_default_patch_leaves_no_published_source(repository, tmp_path, change): + asset = install_test_patch(repository) + defaults = json.loads(source.DEFAULT_FILE.read_text()) + if change == "digest": + asset.write_bytes(asset.read_bytes() + b"\n") + elif change in {"format", "noncanonical"}: + raw = b"not a Git patch\n" if change == "format" else asset.read_bytes().replace( + b"@@ -1 +1 @@", b"@@ -1,1 +1,1 @@") + asset.write_bytes(raw) + defaults["compatibility_patch"]["sha256"] = hashlib.sha256(raw).hexdigest() + elif change == "path": + defaults["compatibility_patch"]["relative_path"] = "../outside.patch" + else: + alias = asset.with_name("alias.patch") + alias.symlink_to(asset) + defaults["compatibility_patch"]["relative_path"] = "compatibility/alias.patch" + source.DEFAULT_FILE.write_text(json.dumps(defaults)) + with pytest.raises(ValueError): + source.prepare_source(tmp_path / "logs") + assert not (tmp_path / "logs" / source.STATE_DIR).exists() + assert not list((tmp_path / "logs").glob(".edgebench-source-*")) + + +@pytest.mark.parametrize("change", ["patch", "digest", "id", "extra_commit", "extra_change", "schema", "extra_field"]) +def test_patched_receipt_and_exact_commit_are_verified(repository, tmp_path, change): + install_test_patch(repository) + checkout, receipt = source.prepare_source(tmp_path / "logs") + if change == "patch": + path = checkout.parent / "compatibility.patch" + path.write_bytes(path.read_bytes() + b"\n") + elif change == "digest": + receipt["compatibility_patch"]["sha256"] = "0" * 64 + elif change == "id": + receipt["compatibility_patch"]["id"] = "unrecorded-patch" + elif change in {"extra_commit", "extra_change"}: + if change == "extra_change": + (checkout / "unexpected.txt").write_text("outside the frozen patch\n") + git(checkout, "add", "unexpected.txt") + git(checkout, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", + "-c", "core.hooksPath=/dev/null", "commit", "--allow-empty", "-qm", "extra commit") + receipt["commit"] = git(checkout, "rev-parse", "HEAD") + elif change == "schema": + receipt["schema_version"] = "edgebench_source_v1" + else: + receipt["extra"] = "unversioned metadata" + (checkout.parent / "receipt.json").write_text(json.dumps(receipt)) + with pytest.raises(ValueError): + source.prepare_source(tmp_path / "logs") + + +def test_legacy_receipt_is_not_upgraded_when_default_gains_patch(repository, tmp_path): + checkout, legacy = source.prepare_source(tmp_path / "legacy") + original = (checkout.parent / "receipt.json").read_bytes() + install_test_patch(repository) + assert source.prepare_source(tmp_path / "legacy")[1] == legacy + assert legacy["schema_version"] == "edgebench_source_v1" + assert (checkout.parent / "receipt.json").read_bytes() == original + assert not (checkout.parent / "compatibility.patch").exists() + assert source.prepare_source(tmp_path / "new")[1]["schema_version"] == "edgebench_source_v2" + + +def test_patched_cohort_is_offline_and_independent_of_current_bundle(repository, tmp_path, monkeypatch): + asset = install_test_patch(repository) + checkout, expected = source.prepare_source(tmp_path / "logs") + original = source._git_bytes + + def offline(checkout, *args, **kwargs): + assert args[0] not in {"fetch", "clone", "ls-remote"} + return original(checkout, *args, **kwargs) + + monkeypatch.setattr(source, "_git_bytes", offline) + asset.unlink() + source.DEFAULT_FILE.unlink() + assert source.prepare_source(tmp_path / "logs")[1] == expected + assert source.prepare_source(tmp_path / "logs", revision=expected["commit"])[1] == expected + with pytest.raises(ValueError, match="conflicts"): + source.prepare_source(tmp_path / "logs", revision=expected["upstream_commit"]) + assert git(checkout, "status", "--porcelain") == "" + + +@pytest.mark.parametrize("choice", ["ref", "revision", "repository"]) +def test_explicit_selection_does_not_apply_default_patch(repository, tmp_path, choice): + asset = install_test_patch(repository) + # An explicit choice must not even depend on the bundled patch being usable. + asset.unlink() + upstream = git(repository, "rev-parse", "HEAD") + selection = {"ref": "refs/heads/integration"} if choice != "revision" else {"revision": upstream} + if choice == "repository": + alternate = tmp_path / "alternate" + subprocess.run(["git", "clone", "-q", str(repository), str(alternate)], check=True) + (alternate / "sforge/__init__.py").write_text("MARKER = 'custom-source'\n") + git(alternate, "add", "sforge/__init__.py") + git(alternate, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", + "-c", "core.hooksPath=/dev/null", "commit", "-qm", "custom source") + selection["repository"] = str(alternate) + upstream = git(alternate, "rev-parse", "HEAD") + checkout, receipt = source.prepare_source(tmp_path / "logs", **selection) + assert receipt["schema_version"] == "edgebench_source_v1" + assert receipt["commit"] == upstream + assert not (checkout.parent / "compatibility.patch").exists() + expected = "custom-source" if choice == "repository" else "shared-source" + assert (checkout / "sforge/__init__.py").read_text() == f"MARKER = '{expected}'\n" + + +def test_real_bootstrap_imports_prepared_patched_source(repository, tmp_path): + install_test_patch(repository) + checkout, receipt = source.prepare_source(tmp_path / "logs") + env = dict(os.environ) + env.pop(source.RECEIPT_ENV, None) + result = subprocess.run([ + sys.executable, "-m", "benchmark.edgebench.serve", "--log-dir", str(tmp_path / "logs"), + "--port", "8088", + ], env=env, capture_output=True, text=True, check=True) + observed = json.loads(result.stdout) + assert observed["marker"] == "patched-source" + assert observed["source"] == receipt + assert observed["source"]["commit"] == git(checkout, "rev-parse", "HEAD")