From dce4f04b1ee895d7408147972f7fc976b39843eb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 11:03:12 +0800 Subject: [PATCH 1/3] fix(benchmark): reconcile managed terminal evidence before slot release Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> (cherry picked from commit cd2f27317f57caeadd19db6ced3fc8bc0cd19d45) Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/edgebench/README.md | 39 +++ benchmark/edgebench/managed_run.py | 280 ++++++++++++++++++ benchmark/edgebench/run.py | 44 ++- .../capabilities/benchmark_toolkit/README.md | 52 ++++ .../benchmark_toolkit/concurrency_envelope.py | 87 +++--- .../terminal_followthrough.py | 138 +++++++++ loopx/cli_commands/benchmark_concurrency.py | 45 ++- loopx/cli_commands/benchmark_dispatch.py | 2 + .../benchmark_terminal_followthrough.ts | 115 +++++++ .../control_plane/effect_runtime_handlers.ts | 1 + loopx/domain_state.py | 34 ++- tests/benchmark/test_edgebench_managed_run.py | 273 +++++++++++++++++ .../test_benchmark_terminal_followthrough.py | 231 +++++++++++++++ .../benchmark_terminal_followthrough.test.ts | 171 +++++++++++ 14 files changed, 1451 insertions(+), 61 deletions(-) create mode 100644 benchmark/edgebench/managed_run.py create mode 100644 loopx/capabilities/benchmark_toolkit/terminal_followthrough.py create mode 100644 loopx/control_plane/capabilities/benchmark_terminal_followthrough.ts create mode 100644 tests/benchmark/test_edgebench_managed_run.py create mode 100644 tests/capabilities/test_benchmark_terminal_followthrough.py create mode 100644 tests/control_plane_ts/benchmark_terminal_followthrough.test.ts diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index 004e4e0a91..19e8a1f858 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -855,6 +855,45 @@ Omit the flag on the next run to disable it. No active experiment is changed by installing this code. Inspect wake receipts and the saved decisions before claiming adoption; score countability still requires the usual integrity review. +## Close a managed native attempt + +For a native-feedback run with an existing experiment-board row and a fresh +concurrency admission, pass `--managed-run-json PRIVATE_BINDING.json` to the +EdgeBench runner. Keep the binding owner-only and outside the solver workspace. +It contains `goal_id`, the absolute controller `project` path, the original +`admission_receipt` from `loopx benchmark concurrency-admit --execute --format json`, +and the board `identity` (`benchmark_id`, `study_id`, `case_id`, `run_id`, `arm_id`, +`arm_role`, `attempt`). Match `case_id` to `--task` and `run_id` to `--run-id`. +Use the receipt that created the slot and a unique run ID across the Goal. + +The runner retains `managed-run.json` beside the runtime receipt. Before the +solver starts, the receipt captures the binding and source pins. At return, it +binds the original solver execution record when termination is confirmed. The +runner publishes `managed-terminal.json` after validating those receipts and +the native grading drain for the same run. The controller holds the registered +Goal and project access. + +If grading drains later, the native grading owner must first retain its +complete, drained `grading_status.json` and the matching history in the same +attempt. To validate that evidence or retry a proof write, run the following +once on the host that owns the attempt. Omit `--execute` to preview. + +```sh +python -m benchmark.edgebench.managed_run --trial "$ATTEMPT_DIR" --execute +``` + +The replay reads retained evidence. Pending, unknown, or mismatched native +evidence leaves the reservation active. Use the proof from the exact attempt on +the controller and retain the command's JSON receipt with its binding. + +```sh +loopx benchmark terminal-reconcile --goal-id "$GOAL_ID" --project "$PROJECT" \ + --request-json "$ATTEMPT_DIR/managed-terminal.json" --execute --format json +``` + +The [toolkit closeout contract](../../loopx/capabilities/benchmark_toolkit/README.md#concurrency-envelope) +defines readback, release and replay. Omit the flag on new ordinary trials. + ## Sharing terminal results The [Portfolio ten-arm retrospective report](studies/portfolio-ten-arm-20261006/README.md) diff --git a/benchmark/edgebench/managed_run.py b/benchmark/edgebench/managed_run.py new file mode 100644 index 0000000000..34ac48f68b --- /dev/null +++ b/benchmark/edgebench/managed_run.py @@ -0,0 +1,280 @@ +"""Host evidence for opt-in EdgeBench terminal reconciliation. + +The runner keeps the binding and proof in its attempt directory. A controller +can pass the proof to ``loopx benchmark terminal-reconcile`` on the registered +project. The runner needs no project or Goal access. +""" + +from __future__ import annotations + +import argparse +from datetime import UTC, datetime +import hashlib +import json +import os +from pathlib import Path +import stat + +from loopx.capabilities.benchmark_toolkit.experiment_identity import ( + ARM_ROLES, + EXPERIMENT_TOKEN_PATTERN, +) +from loopx.capabilities.benchmark_toolkit.runtime_continuity import ( + BenchmarkEventWindowState, + build_benchmark_runtime_continuity, +) +from loopx.control_plane.runtime.document_io import atomic_write_state_text +from loopx.file_lock import write_json_atomic +from loopx.public_safe_text import LETTER_LED_COMPACT_TOKEN_PATTERN + + +BINDING_FILENAME = "managed-run.json" +TERMINAL_FILENAME = "managed-terminal.json" +RUNTIME_FILENAME = "runtime-receipt.json" +_MAX_JSON_BYTES = 1024 * 1024 +_MAX_HISTORY_BYTES = 16 * 1024 * 1024 +_IDENTITY_FIELDS = {"benchmark_id", "study_id", "case_id", "run_id", "arm_id", "arm_role", "attempt"} +_BOARD_STATUS = {"terminal": "completed", "runner_failed": "runner_invalid", "cancelled": "cancelled"} +_NATIVE_TERMINAL_STATUS = {"normal_exit", "nonzero_exit", "timed_out", "cancelled", "signaled"} +_NATIVE_HISTORY_FILES = {"run_history.json", "game_history.json"} + + +def _read_record(path: Path, *, private: bool = False, limit: int = _MAX_JSON_BYTES) -> tuple[dict, str]: + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as source: + info = os.fstat(source.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_size > limit: + raise ValueError("Managed run evidence must be a bounded regular file") + if private and (info.st_uid != os.getuid() or info.st_mode & 0o077): + raise ValueError("Managed run record must be owner-only") + raw = source.read(limit + 1) + if len(raw) > limit: + raise ValueError("Managed run evidence exceeds its size limit") + try: + value = json.loads(raw) + except (ValueError, RecursionError): + raise ValueError("Managed run evidence must be valid JSON") from None + if not isinstance(value, dict): + raise ValueError("Managed run evidence must be a JSON object") + return value, hashlib.sha256(raw).hexdigest() + + +def _timestamp(value) -> datetime: + try: + parsed = datetime.fromisoformat(value) if isinstance(value, str) else None + except ValueError: + parsed = None + if parsed is None or parsed.tzinfo is None: + raise ValueError("Managed run evidence requires a timestamp with a timezone") + return parsed + + +def _validate_binding(binding: dict, *, run_id: str, task_id: str) -> None: + if set(binding) != {"goal_id", "project", "admission_receipt", "identity"}: + raise ValueError("Managed run binding requires Goal, project, admission and identity") + goal_id, project = binding["goal_id"], binding["project"] + if not isinstance(goal_id, str) or not LETTER_LED_COMPACT_TOKEN_PATTERN.fullmatch(goal_id): + raise ValueError("Invalid managed run Goal identity") + if not isinstance(project, str) or "\0" in project or not Path(project).is_absolute(): + raise ValueError("Managed run project must be an absolute controller path") + identity = binding["identity"] + if not isinstance(identity, dict) or set(identity) != _IDENTITY_FIELDS: + raise ValueError("Managed run requires the complete board identity") + for field in _IDENTITY_FIELDS - {"attempt"}: + if not isinstance(identity[field], str) or not EXPERIMENT_TOKEN_PATTERN.fullmatch(identity[field]): + raise ValueError(f"Invalid managed run {field}") + attempt = identity["attempt"] + if type(attempt) is not int or not 0 <= attempt < 2 ** 53 or identity["arm_role"] not in ARM_ROLES: + raise ValueError("Invalid managed run attempt or arm role") + if (identity["run_id"], identity["case_id"]) != (run_id, task_id): + raise ValueError("Managed run identity differs from the selected run and task") + receipt = binding["admission_receipt"] + if (not isinstance(receipt, dict) or receipt.get("ok") is not True + or receipt.get("admitted") is not True or receipt.get("dry_run") is not False + or receipt.get("idempotent") is not False or receipt.get("write_performed") is not True + or receipt.get("goal_id") != goal_id): + raise ValueError("Managed run requires the original new admission for the selected Goal") + run = receipt.get("run") + if not isinstance(run, dict) or any(run.get(key) != identity[key] for key in ("run_id", "case_id", "arm_role")): + raise ValueError("Managed run admission differs from the selected identity") + _timestamp(run.get("admitted_at")) + + +def read_managed_run(path: Path, *, run_id: str, task_id: str) -> dict: + """Bind the explicit launch to its executed admission before starting it.""" + binding, _ = _read_record(path, private=True) + _validate_binding(binding, run_id=run_id, task_id=task_id) + return binding + + +def _runtime_digest(receipt: dict) -> str: + source = receipt.get("sforge_source") + pins = [receipt.get("loopx_commit"), receipt.get("runner_commit"), + source.get("commit") if isinstance(source, dict) else None, receipt.get("task_sha256")] + if any(not isinstance(pin, str) or not pin for pin in pins): + raise ValueError("Managed terminal evidence requires the pinned launch sources and task") + return hashlib.sha256(json.dumps(pins, separators=(",", ":")).encode("utf-8")).hexdigest() + + +def _write_runtime_receipt(trial: Path, receipt: dict, retained: dict) -> None: + write_json_atomic(trial / RUNTIME_FILENAME, {**receipt, "managed_run": retained}, + mode=0o600, fsync_directory=True) + + +def persist_managed_run(trial: Path, binding: dict, *, receipt: dict, game_mode: bool) -> dict: + """Retain private launch provenance before starting the native solver.""" + path = trial / BINDING_FILENAME + runtime_digest = _runtime_digest(receipt) + write_json_atomic(path, binding, mode=0o600, fsync_directory=True) + saved, digest = _read_record(path, private=True) + if saved != binding: + raise ValueError("Managed run binding readback mismatch") + retained = { + "binding_sha256": digest, + "launch_runtime_sha256": runtime_digest, + "history_file": "game_history.json" if game_mode else "run_history.json", + } + _write_runtime_receipt(trial, receipt, retained) + return retained + + +def _solver_ended(execution: dict, *, admitted_at: float, observed_at: float) -> bool: + segments = execution.get("segments") + if type(execution.get("schema_version")) is not int or execution["schema_version"] != 1: + return False + if not isinstance(segments, list) or not segments: + return False + previous_end = admitted_at + for index, segment in enumerate(segments, 1): + if (not isinstance(segment, dict) or type(segment.get("index")) is not int + or segment["index"] != index or segment.get("terminal_observed") is not True + or segment.get("status") not in _NATIVE_TERMINAL_STATUS): + return False + started, finished = segment.get("started_at"), segment.get("finished_at") + if (type(started) not in (int, float) or type(finished) not in (int, float) + or not previous_end <= started <= finished <= observed_at): + return False + previous_end = finished + return execution.get("status") == segments[-1]["status"] + + +def retain_managed_solver(trial: Path, binding: dict, launch: dict, *, receipt: dict, result) -> bool: + """Bind the exact native solver snapshot to the returned runtime receipt.""" + retained = dict(launch) + if receipt["status"] in _BOARD_STATUS and getattr(result, "solver_execution", None) is not None: + execution, digest = _read_record(trial / "solver_execution.json") + admitted = _timestamp(binding["admission_receipt"]["run"]["admitted_at"]) + if (execution == result.solver_execution and _solver_ended( + execution, admitted_at=admitted.timestamp(), observed_at=datetime.now(UTC).timestamp())): + retained["solver_execution_sha256"] = digest + _write_runtime_receipt(trial, receipt, retained) + return "solver_execution_sha256" in retained + + +def _terminal_history_entry(entry, *, task_id: str) -> bool: + if (not isinstance(entry, dict) or entry.get("task_id", task_id) != task_id + or entry.get("type", "submission") not in ("submission", "game")): + return False + status = entry.get("status") + # SForge treats its legacy scored entries as terminal when status is absent. + return status in ("completed", "error") or ( + status is None and any(field in entry for field in ("score", "pass_rate", "max_score")) + ) + + +def _grading_drained(trial: Path, *, history_file: str, run_id: str, task_id: str) -> bool: + try: + grading, _ = _read_record(trial / "grading_status.json") + except FileNotFoundError: + return False + if (type(grading.get("schema_version")) is not int or grading["schema_version"] != 1 + or grading.get("status") != "complete" or grading.get("reason") != "drained" + or type(grading.get("pending_submissions")) is not int or grading["pending_submissions"] != 0 + or grading.get("history_observed") is not True + or grading.get("history_file") != history_file): + return False + history, digest = _read_record(trial / history_file, limit=_MAX_HISTORY_BYTES) + entries = history.get("entries") + return (grading.get("history_sha256") == digest and history.get("run_id") == run_id + and isinstance(entries, list) + and all(_terminal_history_entry(entry, task_id=task_id) for entry in entries)) + + +def publish_managed_terminal(trial: Path, *, execute: bool = False) -> bool: + """Check retained native evidence and optionally publish this attempt's proof.""" + if (trial.parent.parent.name != "runs" or not trial.is_dir() + or any(path.is_symlink() for path in (trial, trial.parent, trial.parent.parent))): + raise ValueError("Managed run requires its exact physical attempt directory") + run_id, task_id = trial.parent.name, trial.name + binding, binding_digest = _read_record(trial / BINDING_FILENAME, private=True) + _validate_binding(binding, run_id=run_id, task_id=task_id) + receipt, _ = _read_record(trial / RUNTIME_FILENAME, private=True) + if (receipt.get("run_id"), receipt.get("task"), receipt.get("feedback")) != (run_id, task_id, "native"): + raise ValueError("Managed runtime receipt differs from the selected attempt") + status = _BOARD_STATUS.get(receipt.get("status")) + retained = receipt.get("managed_run") + if status is None or not isinstance(retained, dict) or "solver_execution_sha256" not in retained: + return False + history_file = retained.get("history_file") + if history_file not in _NATIVE_HISTORY_FILES: + raise ValueError("Retained managed grading history path differs") + execution, solver_digest = _read_record(trial / "solver_execution.json") + admitted = _timestamp(binding["admission_receipt"]["run"]["admitted_at"]) + if (retained.get("solver_execution_sha256") != solver_digest + or not _solver_ended(execution, admitted_at=admitted.timestamp(), + observed_at=datetime.now(UTC).timestamp())): + raise ValueError("Retained solver termination evidence differs") + if not _grading_drained(trial, history_file=history_file, run_id=run_id, task_id=task_id): + return False + continuity = build_benchmark_runtime_continuity( + launch_runtime_digest=retained.get("launch_runtime_sha256"), + closeout_runtime_digest=_runtime_digest(receipt), + launch_generation_digest=retained.get("binding_sha256"), + closeout_generation_digest=binding_digest, + event_window_state=BenchmarkEventWindowState.QUALIFIED, + ) + if continuity["closeout_write_allowed"] is not True: + raise ValueError("Retained managed run continuity differs") + + observed = datetime.now(UTC) + identity = binding["identity"] + proof = { + "admission_receipt": binding["admission_receipt"], + "identity": identity, + "terminal": { + **identity, "status": status, "observed_at": observed.isoformat().replace("+00:00", "Z"), + "solver_terminal": True, "grading_drained": True, + }, + } + path = trial / TERMINAL_FILENAME + if execute: + try: + atomic_write_state_text(path, json.dumps(proof, ensure_ascii=False, indent=2) + "\n", create_only=True) + except FileExistsError: + pass # A concurrent replay can only adopt the same published proof. + if execute or path.exists() or path.is_symlink(): + existing, _ = _read_record(path, private=True) + terminal = existing.get("terminal") + finished = execution["segments"][-1]["finished_at"] + if not isinstance(terminal, dict) or _timestamp(terminal.get("observed_at")).timestamp() < finished: + raise ValueError("Existing managed terminal proof has an invalid timestamp") + proof["terminal"]["observed_at"] = terminal["observed_at"] + if existing != proof: + raise ValueError("Existing managed terminal proof differs from the retained attempt") + return True + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--trial", required=True, type=Path, help="Exact host-owned runs/RUN_ID/TASK_ID directory") + parser.add_argument("--execute", action="store_true", help="Publish the proof after validating retained evidence") + args = parser.parse_args(argv) + try: + ready = publish_managed_terminal(args.trial, execute=args.execute) + except (OSError, TypeError, ValueError) as error: + parser.error(f"Retained managed terminal evidence rejected ({type(error).__name__})") + print(json.dumps({"ready": ready, "dry_run": not args.execute})) + return 0 if ready else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 859de3bfa0..eb686ab247 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -120,6 +120,8 @@ def main(argv=None): parser.add_argument("--worker", choices=PROFILES, required=True) parser.add_argument("--feedback", choices=FEEDBACK_MODES, default="native", help="New-run default: native public feedback; blind/best-only are explicit controls") + parser.add_argument("--managed-run-json", type=Path, + help="Opt-in private binding for native terminal proof, consumed by the controller") parser.add_argument("--turn-envelope", action="store_true", help="Opt-in short heartbeat context with same-invocation full captures") parser.add_argument("--context-delivery", choices=("off", "on"), @@ -133,7 +135,8 @@ def main(argv=None): parser.add_argument("--native-goal-continue", action="store_true", help="Opt-in: when a native Goal declares completion with budget left, start " "continuation Goals until the budget edge (native-goal worker only)") - parser.add_argument("--model", required=True) + # Preserve the existing unique abbreviation after adding --managed-run-json. + parser.add_argument("--model", "--m", required=True) parser.add_argument("--effort", choices=REASONING_EFFORTS, required=True) parser.add_argument("--provider-rate-limit-retry-seconds", type=int, default=0, help="Host API gateway HTTP 429 retry window for new trials; " @@ -194,6 +197,19 @@ def main(argv=None): if args.native_child_limit is not None and ( args.native_child_limit < 1 or args.feedback != "native"): parser.error("--native-child-limit requires a positive limit and native feedback") + managed_binding = None + if args.managed_run_json is not None: + if args.feedback != "native": + parser.error("--managed-run-json requires native feedback") + from benchmark.edgebench.managed_run import ( + persist_managed_run, publish_managed_terminal, read_managed_run, retain_managed_solver, + ) + try: + managed_binding = read_managed_run(args.managed_run_json, run_id=args.run_id, task_id=args.task) + except OSError: + parser.error("Could not read the private managed run binding") + except ValueError as error: + parser.error(str(error)) api_base_url = os.environ.get("OPENAI_BASE_URL") or None api_key = os.environ.get("OPENAI_API_KEY") if api_base_url else None if api_base_url and not api_key: @@ -274,6 +290,9 @@ def main(argv=None): # One directory is one attempt: never reuse native registration or overwrite # source/profile evidence after an ambiguous launch. trial = args.log_dir / "runs" / args.run_id / args.task + if (managed_binding is not None + and trial.resolve() != args.log_dir.resolve() / "runs" / args.run_id / args.task): + parser.error("Managed run requires its exact physical attempt directory") if args.restore_checkpoint is None: trial.mkdir(parents=True, exist_ok=False) else: @@ -299,6 +318,8 @@ def main(argv=None): task = make_task_spec(task_file, load_benchmark(args.tasks_dir)) if args.native_child_limit is not None and task.game_mode: parser.error("--native-child-limit requires a non-game task with workspace instructions") + if managed_binding is not None and task.task_id != args.task: + parser.error("Managed run task specification differs from the selected task") research_sidecar = args.feedback == "native" and not task.game_mode if research_sidecar: from benchmark.runtime.research_artifacts import validate_research_scope, with_research_capture @@ -507,7 +528,11 @@ def main(argv=None): **({"preserve_failed_worker": True} if args.preserve_failed_worker else {}), } receipt_path = trial / "runtime-receipt.json" - receipt_path.write_text(json.dumps(receipt, indent=2)) + managed_launch = None + if managed_binding is not None: + managed_launch = persist_managed_run(trial, managed_binding, receipt=receipt, game_mode=task.game_mode) + else: + receipt_path.write_text(json.dumps(receipt, indent=2)) try: result, was_interrupted, elapsed = _observe_run(lambda: run_agent( task_spec=task, agent=agent, config=config, backend=backend, @@ -555,7 +580,20 @@ def main(argv=None): task=task.task_id, run_id=args.run_id, model=args.model, effort=args.effort, feedback=args.feedback, native_goal_outcome=native.get("native_goal_outcome")) - receipt_path.write_text(json.dumps(receipt, indent=2)) + if managed_launch is None: + receipt_path.write_text(json.dumps(receipt, indent=2)) + else: + try: + solver_retained = retain_managed_solver(trial, managed_binding, managed_launch, receipt=receipt, result=result) + except (OSError, TypeError, ValueError) as error: + receipt_path.write_text(json.dumps(receipt, indent=2)) + logger.warning("Managed solver evidence unavailable (%s)", type(error).__name__) + else: + try: + if not solver_retained or not publish_managed_terminal(trial, execute=True): + logger.warning("Managed terminal proof withheld: native solver termination or grading drain unconfirmed") + except (OSError, TypeError, ValueError) as error: + logger.warning("Managed terminal proof unavailable (%s)", type(error).__name__) print(json.dumps(receipt)) return 0 if status == "terminal" else 130 if status == "cancelled" else 1 diff --git a/loopx/capabilities/benchmark_toolkit/README.md b/loopx/capabilities/benchmark_toolkit/README.md index c9795efbb8..4b998c809f 100644 --- a/loopx/capabilities/benchmark_toolkit/README.md +++ b/loopx/capabilities/benchmark_toolkit/README.md @@ -929,6 +929,58 @@ transition, and a bounded periodic cadence, pass exact-job receipt and runner-ow `runtime-observation`. Apply its typed terminal or runner-invalid transition before releasing that reservation, then backfill the reported gap. +Managed providers can deliver a completed closeout through one host operation: + +```bash +loopx benchmark terminal-reconcile \ + --goal-id \ + --request-json managed-terminal.json \ + --execute --format json +``` + +Without `--execute` this previews the transition. The request contains the +original executed, newly-created `admission_receipt` (not an idempotent +admission lookup), the existing board `identity` +(`benchmark_id`, `study_id`, `case_id`, `run_id`, `arm_id`, `arm_role`, `attempt`), +and the provider's `terminal` proof: the same identity, terminal `status`, +`observed_at`, and explicit `solver_terminal=true` and `grading_drained=true`. +This is trusted controller input, not solver input. The provider must qualify +runtime continuity against its retained launch and native event evidence before +publishing it. A running solver, pending +grading, an unknown observation, missing board/ledger, mismatched admission, or +a conflicting terminal status is rejected without releasing a reservation. +The managed path requires a unique run ID across the Goal's board; it rejects +the same reservation appearing in multiple studies rather than guessing its owner. +Every launch or replacement must use a fresh Goal-wide run ID. Deliberately +reusing both an old run ID and its admission timestamp is outside this managed +contract; the legacy reservation format cannot distinguish that duplicate tuple. +Forward the original whole provider proof. Do not assemble its admission and +terminal pieces from separate runs. The shared CLI validates its bound state +transition; it does not independently reread provider files or attest the SDK. + +The TypeScript policy checks the exact admission generation (`admitted_at` as +well as run/case/role). The existing local ledger transports lock the concurrency +envelope and board in that order, recheck the policy's input snapshot, write and +read back the terminal row, then release and read back that original slot. +An intervening write requires retrying the original proof; it does not authorize +a replacement admission. A process interruption after the board write can be +replayed. Same-terminal replay preserves later scores, insight and observation +time, and an already released run is idempotent. This is process-crash recovery, +not a new power-loss durability guarantee. New state transitions use the actual +host reconciliation time, strictly after the current board observation and no +earlier than provider termination; replay keeps the persisted timestamp. A stale +running shard can therefore be merged again without an equal-time ambiguity. + +This command never stops workers, deletes artifacts, restarts an experiment, +changes duration allowances or upgrades score qualification. Keep accepted +grading alive until it drains. The provider/controller owns bounded retry after +an unavailable writer; no scheduler or polling loop is installed. A terminal +board row alone is not a release receipt. Read `board_readback_verified` and +`slot_readback_verified`; new admission still needs its own resource checks. +Standalone board upserts and explicit `concurrency-release` retain their +existing behavior. The managed command additionally requires a registered Goal +and cannot use an unregistered project override. + For hosts where parallel jobs can exhaust temporary storage, memory, process capacity, file descriptors, persistent storage, or provider capacity, enable `--require-resource-headroom-receipt`. Each new admission must then include a diff --git a/loopx/capabilities/benchmark_toolkit/concurrency_envelope.py b/loopx/capabilities/benchmark_toolkit/concurrency_envelope.py index 406fc34e32..a34de17346 100644 --- a/loopx/capabilities/benchmark_toolkit/concurrency_envelope.py +++ b/loopx/capabilities/benchmark_toolkit/concurrency_envelope.py @@ -640,6 +640,45 @@ def admit_benchmark_case( return result +def _release_preview( + current: Mapping[str, Any] | None, *, compact_run_id: str, timestamp: str, +) -> tuple[dict[str, Any], dict[str, Any] | None]: + if current is None: + return ( + { + "ok": True, + "released": False, + "idempotent": True, + "write_performed": False, + "reason_codes": ["concurrency_envelope_not_configured"], + "run_id": compact_run_id, + }, + None, + ) + normalized = normalize_benchmark_concurrency_envelope(current) + remaining = [ + item + for item in normalized["active_runs"] + if item["run_id"] != compact_run_id + ] + released = len(remaining) != len(normalized["active_runs"]) + envelope = normalize_benchmark_concurrency_envelope( + {**normalized, "updated_at": timestamp, "active_runs": remaining} + ) + return ( + { + "ok": True, + "released": released, + "idempotent": not released, + "write_performed": False, + "reason_codes": [] if released else ["run_not_active"], + "run_id": compact_run_id, + "status": build_benchmark_concurrency_status(envelope), + }, + envelope if released else None, + ) + + def release_benchmark_case( path: str | Path, *, @@ -652,52 +691,20 @@ def release_benchmark_case( compact_run_id = _token(run_id, field="run_id") timestamp = _timestamp(released_at or _now_iso(), field="released_at") - def build( - current: Mapping[str, Any] | None, - ) -> tuple[dict[str, Any], dict[str, Any] | None]: - if current is None: - return ( - { - "ok": True, - "released": False, - "idempotent": True, - "write_performed": False, - "reason_codes": ["concurrency_envelope_not_configured"], - "run_id": compact_run_id, - }, - None, - ) - normalized = normalize_benchmark_concurrency_envelope(current) - remaining = [ - item - for item in normalized["active_runs"] - if item["run_id"] != compact_run_id - ] - released = len(remaining) != len(normalized["active_runs"]) - envelope = normalize_benchmark_concurrency_envelope( - {**normalized, "updated_at": timestamp, "active_runs": remaining} - ) - return ( - { - "ok": True, - "released": released, - "idempotent": not released, - "write_performed": False, - "reason_codes": [] if released else ["run_not_active"], - "run_id": compact_run_id, - "status": build_benchmark_concurrency_status(envelope), - }, - envelope if released else None, - ) - if not execute: - result, _ = build(read_benchmark_concurrency_envelope(target)) + result, _ = _release_preview( + read_benchmark_concurrency_envelope(target), + compact_run_id=compact_run_id, timestamp=timestamp, + ) result["dry_run"] = True return result with exclusive_file_lock( target, agent_id=agent_id, operation="benchmark_concurrency_release" ): - result, envelope = build(read_benchmark_concurrency_envelope(target)) + result, envelope = _release_preview( + read_benchmark_concurrency_envelope(target), + compact_run_id=compact_run_id, timestamp=timestamp, + ) result["dry_run"] = False if envelope is not None: atomic_write_json(target, envelope) diff --git a/loopx/capabilities/benchmark_toolkit/terminal_followthrough.py b/loopx/capabilities/benchmark_toolkit/terminal_followthrough.py new file mode 100644 index 0000000000..8326867c40 --- /dev/null +++ b/loopx/capabilities/benchmark_toolkit/terminal_followthrough.py @@ -0,0 +1,138 @@ +"""Local ledger IO for the typed, managed benchmark terminal transition. + +The provider owns terminal/grading evidence. This adapter never discovers a +process, stops a solver, deletes a container, or invents scoring qualification. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from contextlib import ExitStack +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from ...control_plane.effect_runtime import ( + EffectRuntimeConflict, EffectRuntimeRejected, EffectRuntimeRemoteError, + EffectRuntimeStartupError, effect_runtime_result, +) +from ...domain_state import _replace_domain_state_jsonl +from ...file_lock import exclusive_file_lock +from ...registry import atomic_write_json +from .concurrency_envelope import ( + _release_preview, + default_benchmark_concurrency_envelope_path, + read_benchmark_concurrency_envelope, +) +from .experiment_board import ( + benchmark_experiment_board_row_key, + default_benchmark_experiment_board_path, + normalize_benchmark_experiment_board_row, + read_benchmark_experiment_board_rows, +) +from .runtime_observation import build_benchmark_runtime_observation + + +def reconcile_benchmark_terminal( + *, project: str | Path, goal_id: str, + admission_receipt: Mapping[str, Any], identity: Mapping[str, Any], + terminal: Mapping[str, Any], execute: bool = False, + agent_id: str | None = None, +) -> dict[str, Any]: + """Write/read terminal, then release/read the original admission generation. + + The typed decision is prepared before locking; its exact inputs are checked + under both existing Python locks, concurrency first, through readback. + A crash between the two atomic replacements leaves a terminal board row and + the original reservation; replay completes release without rewriting scores. + The standalone board and concurrency commands retain their existing policy. + """ + if not all(isinstance(item, Mapping) for item in (admission_receipt, identity, terminal)): + raise TypeError("benchmark terminal inputs must be objects") + envelope_path = default_benchmark_concurrency_envelope_path( + project=project, goal_id=goal_id, + ) + board_path = default_benchmark_experiment_board_path( + project=project, goal_id=goal_id, + ) + envelope = read_benchmark_concurrency_envelope(envelope_path) + if envelope is None: + raise ValueError("benchmark terminal admission ledger unavailable") + active_runs = envelope["active_runs"] + rows = read_benchmark_experiment_board_rows(board_path) + admitted = any(row["run_id"] == identity.get("run_id") for row in active_runs) + # This is classification, not authority: the typed owner below verifies + # both identities and the exact admission against the locked ledgers. + observation = build_benchmark_runtime_observation( + admission_active=admitted, job_receipt_state="resolved", + runner_owner_state="unknown", + terminal_result_present=(terminal.get("solver_terminal") is True + and terminal.get("grading_drained") is True), + ) + try: + decision = effect_runtime_result("benchmark.terminal-followthrough.prepare", { + "goal_id": goal_id, "admission_receipt": dict(admission_receipt), + "identity": dict(identity), "terminal": dict(terminal), + "reconciled_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + # All rows with this run id remain visible for cross-study + # ambiguity checks; unrelated history does not cross the RPC cap. + "rows": [row for row in rows if row["run_id"] == identity.get("run_id")], + "active_runs": [slot for slot in active_runs if slot["run_id"] == identity.get("run_id")], + "runtime_observation": observation, + }) + except (EffectRuntimeRejected, EffectRuntimeConflict) as exc: + raise ValueError("benchmark terminal identity or evidence rejected") from exc + except (EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: + raise ValueError("benchmark terminal runtime unavailable; retry original evidence") from exc + row = normalize_benchmark_experiment_board_row(decision["row"]) + key = benchmark_experiment_board_row_key(row) + before = next(item for item in rows if benchmark_experiment_board_row_key(item) == key) + result = { + "ok": True, "goal_id": goal_id, "row": row, + "dry_run": not execute, "write_performed": False, + "released": False, "already_released": decision["already_released"], + "board_readback_verified": False, "slot_readback_verified": False, + "runtime_observation": observation, "path_recorded": False, + } + if not execute: + return result + with ExitStack() as stack: + for path in (envelope_path, board_path): + stack.enter_context(exclusive_file_lock( + path, agent_id=agent_id, operation="benchmark_terminal_reconcile", + )) + # The policy subprocess never holds ledger locks. Reject any state + # change since its exact input snapshot, then commit under both locks. + if (read_benchmark_concurrency_envelope(envelope_path) != envelope + or read_benchmark_experiment_board_rows(board_path) != rows): + raise ValueError("benchmark terminal inputs changed; retry original evidence") + if row != before: + # Preserve other rows and the established domain-state key format. + next_rows = [row if benchmark_experiment_board_row_key(item) == key else item + for item in rows] + _replace_domain_state_jsonl(board_path, [ + {**item, "domain_state_key": benchmark_experiment_board_row_key(item)} + for item in next_rows + ]) + result["write_performed"] = True + persisted = [item for item in read_benchmark_experiment_board_rows(board_path) + if benchmark_experiment_board_row_key(item) == key] + if persisted != [row]: + raise ValueError("benchmark terminal board readback mismatch; reservation retained") + result["board_readback_verified"] = True + if not decision["already_released"]: + # The CAS input was rechecked under this same envelope lock. + # Reuse the existing release transformation, not a second ledger rule. + release, replacement = _release_preview( + envelope, compact_run_id=row["run_id"], + timestamp=datetime.now(UTC).isoformat().replace("+00:00", "Z"), + ) + if replacement is None or release["released"] is not True: + raise ValueError("benchmark terminal admission release unresolved") + atomic_write_json(envelope_path, replacement) + result.update(released=True, write_performed=True) + readback = read_benchmark_concurrency_envelope(envelope_path) + if readback is None or any(item["run_id"] == row["run_id"] for item in readback["active_runs"]): + raise ValueError("benchmark terminal slot readback unresolved") + result["slot_readback_verified"] = True + return result diff --git a/loopx/cli_commands/benchmark_concurrency.py b/loopx/cli_commands/benchmark_concurrency.py index f624ca93bf..a19a296fa4 100644 --- a/loopx/cli_commands/benchmark_concurrency.py +++ b/loopx/cli_commands/benchmark_concurrency.py @@ -32,9 +32,24 @@ "concurrency-admit", "concurrency-release", "concurrency-tune", + "terminal-reconcile", } +def _render_terminal_markdown(payload: dict[str, object]) -> str: + row = payload.get("row", {}) + return "\n".join([ + "# Benchmark Terminal Reconciliation", "", + f"- Run: `{row.get('run_id')}`", + f"- Status: `{row.get('status')}`", + f"- Preview: `{payload.get('dry_run')}`", + f"- Board readback verified: `{payload.get('board_readback_verified')}`", + f"- Slot readback verified: `{payload.get('slot_readback_verified')}`", + f"- Released: `{payload.get('released')}`", + f"- Already released: `{payload.get('already_released')}`", "", + ]) + + def _add_location_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("--goal-id", required=True) parser.add_argument( @@ -123,6 +138,18 @@ def register_benchmark_concurrency_commands( release_parser.add_argument("--run-id", required=True) _add_execute_argument(release_parser) + terminal_parser = benchmark_subparsers.add_parser( + "terminal-reconcile", + help="Write a bound provider terminal result, then release its original admission.", + ) + add_subcommand_format(terminal_parser) + _add_location_arguments(terminal_parser) + terminal_parser.add_argument( + "--request-json", required=True, + help="Host-owned terminal proof with admission_receipt, identity and terminal; never solver input.", + ) + _add_execute_argument(terminal_parser) + tune_parser = benchmark_subparsers.add_parser( "concurrency-tune", help="Adapt desired occupancy within the operator-owned hard ceiling.", @@ -237,13 +264,27 @@ def handle_benchmark_concurrency_command( execute=args.execute, agent_id=args.agent_id, ) + elif args.benchmark_command == "terminal-reconcile": + from ..capabilities.benchmark_toolkit.terminal_followthrough import reconcile_benchmark_terminal + + request = _read_json_argument(args.request_json) + if not isinstance(request, dict) or set(request) != {"admission_receipt", "identity", "terminal"}: + raise ValueError("invalid benchmark terminal request") + payload = reconcile_benchmark_terminal( + project=project, goal_id=args.goal_id, **request, + execute=args.execute, agent_id=args.agent_id, + ) else: payload = _tune_payload(args, path=path) payload["goal_id"] = args.goal_id payload["path_recorded"] = False except (OSError, TypeError, ValueError): args.benchmark_concurrency_parser.error( - "invalid benchmark concurrency envelope input" + "terminal reconciliation did not complete; check input and runtime, then retry the original proof" + if args.benchmark_command == "terminal-reconcile" + else "invalid benchmark concurrency envelope input" ) - print_payload(payload, output_format(args), render_benchmark_concurrency_markdown) + renderer = (_render_terminal_markdown if args.benchmark_command == "terminal-reconcile" + else render_benchmark_concurrency_markdown) + print_payload(payload, output_format(args), renderer) return 0 if payload.get("ok") is not False else 2 diff --git a/loopx/cli_commands/benchmark_dispatch.py b/loopx/cli_commands/benchmark_dispatch.py index 52520a8779..69b68dfe7b 100644 --- a/loopx/cli_commands/benchmark_dispatch.py +++ b/loopx/cli_commands/benchmark_dispatch.py @@ -94,6 +94,8 @@ def _resolve_benchmark_project( project_override=requested, ) return project + if args.benchmark_command == "terminal-reconcile": + raise ValueError("managed terminal reconciliation requires a registered Goal") return requested diff --git a/loopx/control_plane/capabilities/benchmark_terminal_followthrough.ts b/loopx/control_plane/capabilities/benchmark_terminal_followthrough.ts new file mode 100644 index 0000000000..bb77f8eede --- /dev/null +++ b/loopx/control_plane/capabilities/benchmark_terminal_followthrough.ts @@ -0,0 +1,115 @@ +import {EffectRuntimeConflictError, EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {requireJsonObject, requireNonEmptyString, requireStringLiteral, type JsonObject} from "../runtime_decode.ts"; +import {parseTodoTimestampMicros} from "../runtime_timestamp.ts"; + +const RUN_IDENTITY_FIELDS = ["benchmark_id", "study_id", "case_id", "run_id", "arm_id", "arm_role", "attempt"]; + +interface ActiveRun { + run_id: string; + case_id: string; + arm_role: string; + admitted_at: string; +} + +interface TerminalFollowthroughPlan { + row: JsonObject; + active_run: ActiveRun | null; + already_released: boolean; +} + +function objects(value: unknown, label: string): JsonObject[] { + if (!Array.isArray(value)) throw new EffectRuntimeRequestError(`${label} must be an array`); + return value.map(item => requireJsonObject(item, label)); +} + +function timestamp(value: unknown, label: string): {text: string; micros: bigint} { + const text = requireNonEmptyString(value, label); + const micros = parseTodoTimestampMicros(text, true); + if (micros === null) throw new EffectRuntimeRequestError(`${label} must be an ISO timestamp with a timezone`); + return {text, micros}; +} + +/** The adapter binds provider facts and holds the board and admission locks through readback and release. */ +export function prepareBenchmarkTerminalFollowthrough(value: unknown): TerminalFollowthroughPlan { + const request = requireJsonObject(value, "benchmark terminal followthrough"); + const goalId = requireNonEmptyString(request.goal_id, "goal_id"); + const identity = requireJsonObject(request.identity, "identity"); + for (const field of RUN_IDENTITY_FIELDS) { + if (field !== "attempt") requireNonEmptyString(identity[field], `identity.${field}`); + } + if (typeof identity.attempt !== "number" || !Number.isSafeInteger(identity.attempt) || identity.attempt < 0) { + throw new EffectRuntimeRequestError("identity.attempt must be a non-negative safe integer"); + } + + const admission = requireJsonObject(request.admission_receipt, "admission_receipt"); + if (admission.ok !== true || admission.admitted !== true || admission.dry_run !== false || + admission.write_performed !== true || admission.idempotent !== false || admission.goal_id !== goalId) { + throw new EffectRuntimeRequestError("terminal followthrough requires the original executed admission for this goal"); + } + const run = requireJsonObject(admission.run, "admission_receipt.run"); + const admittedAt = timestamp(run.admitted_at, "admission_receipt.run.admitted_at"); + const expectedRun: ActiveRun = { + run_id: requireNonEmptyString(run.run_id, "admission_receipt.run.run_id"), + case_id: requireNonEmptyString(run.case_id, "admission_receipt.run.case_id"), + arm_role: requireNonEmptyString(run.arm_role, "admission_receipt.run.arm_role"), + admitted_at: admittedAt.text, + }; + if (expectedRun.run_id !== identity.run_id || expectedRun.case_id !== identity.case_id || + expectedRun.arm_role !== identity.arm_role) { + throw new EffectRuntimeRequestError("admission receipt does not match the run identity"); + } + + const terminal = requireJsonObject(request.terminal, "terminal"); + if (!RUN_IDENTITY_FIELDS.every(field => terminal[field] === identity[field])) { + throw new EffectRuntimeRequestError("terminal evidence does not match the run identity"); + } + const status = requireStringLiteral(terminal.status, ["completed", "runner_invalid", "cancelled"] as const, + "terminal.status"); + const observedAt = timestamp(terminal.observed_at, "terminal.observed_at"); + if (observedAt.micros < admittedAt.micros) { + throw new EffectRuntimeRequestError("terminal observation predates admission"); + } + const reconciledAt = timestamp(request.reconciled_at, "reconciled_at"); + if (observedAt.micros > reconciledAt.micros) { + throw new EffectRuntimeRequestError("terminal observation follows host reconciliation time"); + } + if (terminal.solver_terminal !== true || terminal.grading_drained !== true) { + throw new EffectRuntimeRequestError("terminal followthrough requires a terminal solver and drained grading"); + } + // A goal's concurrency slot has no study or benchmark namespace. + const rows = objects(request.rows, "rows").filter(row => row.run_id === expectedRun.run_id); + if (rows.length !== 1 || rows[0].schema_version !== "benchmark_experiment_board_row_v0" || + !RUN_IDENTITY_FIELDS.every(field => rows[0][field] === identity[field])) { + throw new EffectRuntimeConflictError("terminal followthrough requires one existing exact board row"); + } + const [row] = rows; + const sameTerminal = row.status === status; + if (!sameTerminal && row.status !== "planned" && row.status !== "running") { + throw new EffectRuntimeConflictError("existing board status conflicts with the terminal observation"); + } + const slots = objects(request.active_runs, "active_runs").filter(slot => slot.run_id === expectedRun.run_id); + if (slots.length > 1 || (slots.length === 1 && + !Object.entries(expectedRun).every(([field, expected]) => slots[0][field] === expected))) { + throw new EffectRuntimeConflictError("active slot does not match the executed admission"); + } + const active = slots.length === 1; + if (!active && !sameTerminal) { + throw new EffectRuntimeConflictError("terminal board transition requires the admitted slot"); + } + const observation = requireJsonObject(request.runtime_observation, "runtime_observation"); + if (observation.schema_version !== "benchmark_runtime_observation_v0" || + observation.admission_active !== active || observation.job_receipt_state !== "resolved" || + observation.terminal_result_present !== true || observation.reconciliation_required !== active || + observation.classification !== (active ? "terminal_pending_reconcile" : "not_admitted") || + observation.recommended_transition !== (active ? "write_terminal_then_release" : "none")) { + throw new EffectRuntimeRequestError("terminal followthrough requires resolved terminal runtime evidence"); + } + if (!active) return {row, active_run: null, already_released: true}; + // A replay may arrive after score and insight updates on the terminal row. + if (sameTerminal) return {row, active_run: expectedRun, already_released: false}; + // Reconciliation observes a distinct board transition even if provider evidence arrived late. + if (reconciledAt.micros <= timestamp(row.observed_at, "row.observed_at").micros) { + throw new EffectRuntimeConflictError("host reconciliation time must follow the existing board observation"); + } + return {row: {...row, status, observed_at: reconciledAt.text}, active_run: expectedRun, already_released: false}; +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index af454f6393..719917925f 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -286,6 +286,7 @@ export function createEffectRuntimeHandlers( ["benchmark.trajectory.prepare", lazyHandler(() => import("./capabilities/trajectory_assessment.ts"), ({prepareTrajectoryAssessment}) => prepareTrajectoryAssessment)], ["benchmark.trajectory.qualify", lazyHandler(() => import("./capabilities/trajectory_assessment.ts"), ({qualifyTrajectoryAssessment}) => qualifyTrajectoryAssessment)], ["benchmark.trajectory.continuation", lazyHandler(() => import("./capabilities/trajectory_assessment_continuation.ts"), ({compareTrajectoryAssessments}) => compareTrajectoryAssessments)], + ["benchmark.terminal-followthrough.prepare", lazyHandler(() => import("./capabilities/benchmark_terminal_followthrough.ts"), ({prepareBenchmarkTerminalFollowthrough}) => prepareBenchmarkTerminalFollowthrough)], ["agent.preferences", lazyHandler(() => import("./capabilities/agent_preferences.ts"), ({agentPreferences}) => agentPreferences)], ["todo.archive.capture_dependencies", lazyHandler(() => Promise.all([import("./todos/archive_capture.ts"), import("./coordination/source_transfer.ts")]), ([{captureArchivedTodoDependencies}, {withCoordinationSourceTransfer}]) => withCoordinationSourceTransfer("todo.archive.capture_dependencies", captureArchivedTodoDependencies))], ["agent.supervisor.plan_append", lazyHandler(() => import("./agents/supervisor_event_append.ts"), ({planSupervisorEventAppend}) => planSupervisorEventAppend)], diff --git a/loopx/domain_state.py b/loopx/domain_state.py index 670bf1126f..f6727f8715 100644 --- a/loopx/domain_state.py +++ b/loopx/domain_state.py @@ -45,6 +45,23 @@ def default_domain_state_file_path( ) +def _replace_domain_state_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: + """Replace a ledger while its existing domain-state lock is held.""" + with tempfile.NamedTemporaryFile( + "w", encoding="utf-8", dir=path.parent, + prefix=f"{path.name}.", suffix=".tmp", delete=False, + ) as tmp_file: + tmp_name = tmp_file.name + tmp_file.write("".join( + json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n" + for row in rows + )) + try: + os.replace(tmp_name, path) + finally: + Path(tmp_name).unlink(missing_ok=True) + + def upsert_domain_state_jsonl( ledger_path: str | Path, payload: dict[str, Any], @@ -102,22 +119,7 @@ def upsert_domain_state_jsonl( rows.append(candidate) if not unchanged: - with tempfile.NamedTemporaryFile( - "w", - encoding="utf-8", - dir=path.parent, - prefix=f"{path.name}.", - suffix=".tmp", - delete=False, - ) as tmp_file: - tmp_name = tmp_file.name - tmp_file.write( - "".join( - json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n" - for row in rows - ) - ) - os.replace(tmp_name, path) + _replace_domain_state_jsonl(path, rows) return { "status": "unchanged" if unchanged else "updated" if updated else "inserted", "write_performed": not unchanged, diff --git a/tests/benchmark/test_edgebench_managed_run.py b/tests/benchmark/test_edgebench_managed_run.py new file mode 100644 index 0000000000..cdcf888d3e --- /dev/null +++ b/tests/benchmark/test_edgebench_managed_run.py @@ -0,0 +1,273 @@ +"""Exercise the EdgeBench entry and its controller handoff without model calls.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +import logging +import os +from pathlib import Path +import stat +import subprocess +import sys +import time +from types import SimpleNamespace + +import pytest + +from benchmark.edgebench import managed_run +from benchmark.edgebench.managed_run import BINDING_FILENAME, TERMINAL_FILENAME +from loopx.capabilities.benchmark_toolkit import ( + BENCHMARK_EXPERIMENT_BOARD_ROW_SCHEMA_VERSION, + build_benchmark_concurrency_config, + configure_benchmark_concurrency_envelope, + default_benchmark_concurrency_envelope_path, + default_benchmark_experiment_board_path, + read_benchmark_concurrency_envelope, + read_benchmark_experiment_board_rows, + upsert_benchmark_experiment_board_row, +) +from tests.source_cli import run_source_cli + + +GOAL = "fixture-goal" +IDENTITY = { + "benchmark_id": "edgebench-fixture", "study_id": "study", "case_id": "task", + "run_id": "attempt", "arm_id": "official", "arm_role": "baseline", "attempt": 2, +} + + +def _binding_file(tmp_path: Path, admission: dict, *, identity: dict | None = None) -> Path: + path = tmp_path / "private-binding.json" + path.write_text(json.dumps({ + "goal_id": GOAL, "project": str(tmp_path / "controller-project"), + "admission_receipt": admission, "identity": IDENTITY if identity is None else identity, + }), encoding="utf-8") + path.chmod(0o600) + return path + + +def _admission() -> dict: + return { + "ok": True, "admitted": True, "dry_run": False, "goal_id": GOAL, + "idempotent": False, "write_performed": True, + "run": {key: IDENTITY[key] for key in ("run_id", "case_id", "arm_role")} + | {"admitted_at": datetime.now(UTC).isoformat()}, + } + + +@pytest.fixture +def entry(tmp_path, monkeypatch): + native = pytest.importorskip("sforge.harness.run_agent") + pytest.importorskip("harbor") + if not all(hasattr(native, name) for name in ("_write_grading", "_write_solver_execution", "_wait_for_grading")): + pytest.skip("Selected EdgeBench SDK predates native terminal evidence") + from benchmark.edgebench import run + + for name, value in {"LOOPX_SRC_DIR": str(tmp_path), "LOOPX_EXPECTED_COMMIT": "fixture", + "CODEX_AUTH_JSON_PATH": str(tmp_path / "unused-auth.json")}.items(): + monkeypatch.setenv(name, value) + for name in ("OPENAI_BASE_URL", "OPENAI_API_KEY"): + monkeypatch.delenv(name, raising=False) + (tmp_path / "task.json").write_text("{}", encoding="utf-8") + monkeypatch.setattr(run, "source_pins", lambda *args: ("fixture", "fixture")) + monkeypatch.setattr(run, "active_source_receipt", lambda: {"commit": "7" * 40}) + monkeypatch.setattr(run, "load_benchmark", lambda *args: None) + monkeypatch.setattr(run, "make_task_spec", lambda *args: SimpleNamespace( + task_id="task", cwd="/task", work_image_key="work", judge_image_key="judge", + internet=False, game_mode=False, submit_paths=["solver.py"], + work=SimpleNamespace(agent_query="Synthetic task"), + judge=SimpleNamespace(selection="score_first", score_direction="maximize"))) + backend = SimpleNamespace(image_exists=lambda image: True, native_goal_receipt=None, + execution_receipt={"exit_code": 0, "execution_started": True}) + monkeypatch.setattr(run, "RecordingDockerBackend", lambda **kwargs: backend) + trial = tmp_path / "runs" / "attempt" / "task" + argv = ["--task", "task", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path), + "--run-id", "attempt", "--worker", "official", "--m", "fixture", + "--effort", "high", "--judge-url", "http://localhost:9999"] + + def invoke(binding: Path | None = None, *, evidence: str = "drained") -> int: + def finish(**kwargs): + assert kwargs["run_id"] == "attempt" and kwargs["task_spec"].task_id == "task" + assert "managed_run" not in kwargs and "admission_receipt" not in kwargs + if evidence == "exception": + raise RuntimeError("native provider failed") + (trial / "started_at").write_text("2026-01-01T00:00:00Z\n", encoding="utf-8") + now = time.time() + segment = { + "index": 1, "status": "normal_exit", "finished_at": time.time(), + "started_at": now - 100 if evidence == "predates_admission" else now, + "terminal_observed": evidence != "unconfirmed_solver", "native_exit_code": 0, + } + execution = {"schema_version": 1, "status": "normal_exit", "segments": [segment]} + native._write_solver_execution(trial, execution) + history = {"run_id": "attempt", "entries": []} + if evidence == "pending": + history["entries"].append({"type": "submission", "status": "running"}) + grading = native._GradingResult(history, "pending", "deadline", 1) + elif evidence == "unknown_grading": + grading = native._GradingResult(None, "unknown", "history_unavailable") + else: + grading = native._GradingResult(history, "complete", "drained", 0) + if evidence != "missing_grading": + native._write_grading(trial, "run_history.json", grading, "attempt") + return native.RunResult(runtime_seconds=2, best_score=4, total_rounds=1, + grading_status=grading.status, grading_reason=grading.reason, + solver_execution=execution) + + monkeypatch.setattr(run, "run_agent", finish) + return run.main(argv + (["--managed-run-json", str(binding)] if binding is not None else [])) + + def drain(run_id: str = "attempt", *, contradictory_status: str | None = None) -> None: + history = {"run_id": run_id, "entries": [{"type": "submission", "status": contradictory_status or "completed"}]} + if contradictory_status is None: + grading = native._wait_for_grading( + "http://unused.invalid", "fixture-session", "fixture-secret", eval_timeout=1, + shutdown_event=None, logger=logging.getLogger("fixture"), max_wait_seconds=1, + history_reader=lambda *args, **kwargs: (history, 0), + ) + else: + grading = native._GradingResult(history, "complete", "drained", 0) + assert grading.status == "complete" and grading.reason == "drained" + native._write_grading(trial, "run_history.json", grading, "attempt") + + return trial, invoke, drain + + +def _replay(trial: Path, *, execute: bool = False) -> subprocess.CompletedProcess[str]: + command = [sys.executable, "-m", "benchmark.edgebench.managed_run", "--trial", str(trial)] + return subprocess.run(command + (["--execute"] if execute else []), + capture_output=True, text=True, timeout=30) + + +@pytest.mark.parametrize("stage", ["drained", "pending", "write_failure"]) +def test_managed_terminal_flows_from_runner_to_controller(tmp_path, entry, capsys, monkeypatch, stage): + trial, invoke, drain = entry + project = tmp_path / "controller-project" + registry = project / ".loopx" / "registry.json" + registry.parent.mkdir(parents=True) + registry.write_text(json.dumps({ + "common_runtime_root": str(tmp_path / "runtime"), + "goals": [{"id": GOAL, "objective": "Run a benchmark fixture", "repo": str(project)}], + }), encoding="utf-8") + board = default_benchmark_experiment_board_path(project=project, goal_id=GOAL) + envelope = default_benchmark_concurrency_envelope_path(project=project, goal_id=GOAL) + config = build_benchmark_concurrency_config(max_active_cases=1, max_baseline_cases=1, max_test_cases=0) + configure_benchmark_concurrency_envelope(envelope, config, execute=True) + env = {**os.environ, "LOOPX_PYTHON": sys.executable} + + def controller(command: str, *args: str) -> dict: + result = run_source_cli([ + "--registry", str(registry), "benchmark", command, "--goal-id", GOAL, + "--project", str(project), "--format", "json", "--execute", *args, + ], env=env, timeout=30) + return json.loads(result.stdout) + + admission = controller("concurrency-admit", "--run-id", "attempt", "--case-id", "task", "--arm-role", "baseline") + running = upsert_benchmark_experiment_board_row(board, { + **IDENTITY, "schema_version": BENCHMARK_EXPERIMENT_BOARD_ROW_SCHEMA_VERSION, + "status": "running", "observed_at": admission["run"]["admitted_at"], "model_id": "fixture", + "protocol_id": "native", "comparison_protocol_id": "native", "claim_scope": "inventory_only", + "primary_metric": "score", "treatment_fidelity": "not_applicable", + "countability": {"integrity_qualified": False, "official_result_present": False, "score_countable": False}, + "insight": {"status": "pending"}, + })["row"] + + with monkeypatch.context() as patch: + if stage == "write_failure": + def fail_write(*args, **kwargs): + raise OSError("fixture proof storage failure") + + patch.setattr(managed_run, "atomic_write_state_text", fail_write) + assert invoke(_binding_file(tmp_path, admission), evidence=stage) == 0 + runtime = json.loads(capsys.readouterr().out) + assert runtime["status"] == "terminal" and runtime["best_score"] == 4 + assert GOAL not in json.dumps(runtime) and str(project) not in json.dumps(runtime) + proof = trial / TERMINAL_FILENAME + if stage != "drained": + assert not proof.exists() + if stage == "pending": + pending = _replay(trial, execute=True) + assert pending.returncode == 2 and json.loads(pending.stdout)["ready"] is False + drain(run_id="different-run") + wrong_run = _replay(trial, execute=True) + assert wrong_run.returncode == 2 and not proof.exists() + drain(contradictory_status="running") + contradictory = _replay(trial, execute=True) + assert contradictory.returncode == 2 and not proof.exists() + drain() + runtime_file = trial / "runtime-receipt.json" + original = runtime_file.read_bytes() + changed = json.loads(original) | {"runner_commit": "another-runner"} + runtime_file.write_text(json.dumps(changed), encoding="utf-8") + assert _replay(trial, execute=True).returncode == 2 and not proof.exists() + runtime_file.write_bytes(original) + preview = _replay(trial) + assert preview.returncode == 0 and json.loads(preview.stdout) == {"ready": True, "dry_run": True} + assert not proof.exists() + assert _replay(trial, execute=True).returncode == 0 + for path in (proof, trial / BINDING_FILENAME): + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + original_proof = proof.read_bytes() + assert _replay(trial, execute=True).returncode == 0 and proof.read_bytes() == original_proof + # The runner publishes proof only; the reserved slot survives until host readback. + assert read_benchmark_experiment_board_rows(board) == [running] + assert read_benchmark_concurrency_envelope(envelope)["active_runs"] == [admission["run"]] + + effect = controller("terminal-reconcile", "--request-json", str(proof)) + assert effect["board_readback_verified"] and effect["slot_readback_verified"] and effect["released"] + completed, = read_benchmark_experiment_board_rows(board) + assert completed["status"] == "completed" and completed["attempt"] == 2 + assert completed["metrics"] == running["metrics"] and completed["countability"] == running["countability"] + assert read_benchmark_concurrency_envelope(envelope)["active_runs"] == [] + replay = controller("terminal-reconcile", "--request-json", str(proof)) + assert replay["already_released"] and not replay["write_performed"] + + +@pytest.mark.parametrize("evidence", ["unknown_grading", "missing_grading", "unconfirmed_solver", "predates_admission"]) +def test_unconfirmed_managed_proof_preserves_provider_result(tmp_path, entry, evidence): + trial, invoke, _ = entry + assert invoke(_binding_file(tmp_path, _admission()), evidence=evidence) == 0 + assert (trial / BINDING_FILENAME).is_file() and not (trial / TERMINAL_FILENAME).exists() + runtime = json.loads((trial / "runtime-receipt.json").read_bytes()) + assert runtime["status"] == "terminal" and runtime["best_score"] == 4 + + +@pytest.mark.parametrize("change", ["goal", "case", "run", "preview", "idempotent", "attempt", "mode"]) +def test_wrong_managed_binding_rejected_before_launch(tmp_path, entry, change): + trial, invoke, _ = entry + admission, identity = _admission(), dict(IDENTITY) + if change == "goal": + admission["goal_id"] = "another-goal" + elif change == "case": + identity["case_id"] = "another-task" + elif change == "run": + identity["run_id"] = "another-run" + elif change == "preview": + admission["dry_run"] = True + elif change == "idempotent": + admission.update(idempotent=True, write_performed=False) + elif change == "attempt": + identity["attempt"] = True + binding = _binding_file(tmp_path, admission, identity=identity) + if change == "mode": + binding.chmod(0o644) + with pytest.raises(SystemExit) as stopped: + invoke(binding) + assert stopped.value.code == 2 and not trial.exists() + + +def test_unmanaged_runner_retains_ordinary_result(entry): + trial, invoke, _ = entry + assert invoke() == 0 + assert (trial / "final_result.json").is_file() + assert not (trial / BINDING_FILENAME).exists() and not (trial / TERMINAL_FILENAME).exists() + + +def test_managed_runner_exception_retains_original_failure(tmp_path, entry): + trial, invoke, _ = entry + with pytest.raises(RuntimeError, match="native provider failed"): + invoke(_binding_file(tmp_path, _admission()), evidence="exception") + runtime = json.loads((trial / "runtime-receipt.json").read_bytes()) + assert runtime["status"] == "runner_failed" + assert (trial / BINDING_FILENAME).is_file() and not (trial / TERMINAL_FILENAME).exists() diff --git a/tests/capabilities/test_benchmark_terminal_followthrough.py b/tests/capabilities/test_benchmark_terminal_followthrough.py new file mode 100644 index 0000000000..0002362205 --- /dev/null +++ b/tests/capabilities/test_benchmark_terminal_followthrough.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import json +from datetime import datetime + +import pytest + +from loopx.capabilities.benchmark_toolkit import concurrency_envelope as ce +from loopx.capabilities.benchmark_toolkit import experiment_board as eb +from loopx.capabilities.benchmark_toolkit import terminal_followthrough as tf +from tests.source_cli import run_source_cli + + +@pytest.fixture +def campaign(tmp_path): + goal = "fixture-goal" + project = tmp_path / "project" + registry = project / ".loopx" / "registry.json" + registry.parent.mkdir(parents=True) + registry.write_text(json.dumps({"goals": [{"id": goal, "objective": "Evaluate a fixture.", "repo": str(project)}]})) + board = eb.default_benchmark_experiment_board_path(project=project, goal_id=goal) + envelope = ce.default_benchmark_concurrency_envelope_path(project=project, goal_id=goal) + ce.configure_benchmark_concurrency_envelope( + envelope, ce.build_benchmark_concurrency_config(max_active_cases=4), execute=True, + ) + row = { + "schema_version": "benchmark_experiment_board_row_v0", + "benchmark_id": "fixture", "study_id": "study", "case_id": "case-a", + "run_id": "run-a", "arm_id": "official", "arm_role": "baseline", "attempt": 1, + "status": "running", "observed_at": "2026-01-01T00:00:00Z", + "model_id": "model", "protocol_id": "protocol", "comparison_protocol_id": "comparison", + "claim_scope": "diagnostic_only", "primary_metric": "score", "metrics": {}, + "countability": {"integrity_qualified": False, "official_result_present": False, "score_countable": False}, + "treatment_fidelity": "not_applicable", "effort": {}, "insight": {"status": "pending"}, + } + eb.upsert_benchmark_experiment_board_row(board, row) + admission = ce.admit_benchmark_case( + envelope, run_id="run-a", case_id="case-a", arm_role="baseline", execute=True, + admitted_at="2026-01-01T00:00:00Z", + ) + admission["goal_id"] = goal + ce.admit_benchmark_case(envelope, run_id="other-run", case_id="other-case", arm_role="baseline", execute=True) + identity = {k: row[k] for k in ("benchmark_id", "study_id", "case_id", "run_id", "arm_id", "arm_role", "attempt")} + terminal = {**identity, "status": "completed", "observed_at": "2026-01-01T01:00:00Z", + "solver_terminal": True, "grading_drained": True} + return {"project": project, "goal_id": goal, "admission_receipt": admission, + "identity": identity, "terminal": terminal}, board, envelope, registry + + +def test_real_cli_terminal_release_replay_and_default_preview(campaign, tmp_path): + args, board, envelope, registry = campaign + request = tmp_path / "request.json" + request.write_text(json.dumps({k: args[k] for k in ("admission_receipt", "identity", "terminal")})) + cli = ["--registry", str(registry), "--format", "json", "benchmark", "terminal-reconcile", + "--goal-id", args["goal_id"], "--request-json", str(request)] + before = (board.read_bytes(), envelope.read_bytes()) + preview = json.loads(run_source_cli(cli).stdout) + assert preview["dry_run"] is True and preview["write_performed"] is False + markdown = run_source_cli([*cli, "--format", "markdown"]).stdout + assert "Run: `run-a`" in markdown and "Preview: `True`" in markdown + assert before == (board.read_bytes(), envelope.read_bytes()) + result = json.loads(run_source_cli([*cli, "--execute"]).stdout) + assert result["released"] and result["board_readback_verified"] and result["slot_readback_verified"] + assert result["row"]["status"] == "completed" + assert result["row"]["metrics"] == {} and result["row"]["countability"]["score_countable"] is False + assert [x["run_id"] for x in ce.read_benchmark_concurrency_envelope(envelope)["active_runs"]] == ["other-run"] + done = (board.read_bytes(), envelope.read_bytes()) + repeated = json.loads(run_source_cli([*cli, "--execute"]).stdout) + assert repeated["already_released"] and not repeated["write_performed"] + assert done == (board.read_bytes(), envelope.read_bytes()) + + +def test_delayed_terminal_and_original_running_shard_use_real_cli(campaign, tmp_path): + args, board, envelope, registry = campaign + running = eb.read_benchmark_experiment_board_rows(board)[0] + running["observed_at"] = "2026-01-01T02:00:00Z" + eb.upsert_benchmark_experiment_board_row(board, running) + shard = tmp_path / "provider.jsonl" + eb.upsert_benchmark_experiment_board_row(shard, running) + request = tmp_path / "request.json" + request.write_text(json.dumps({k: args[k] for k in ("admission_receipt", "identity", "terminal")})) + cli = ["--registry", str(registry), "--format", "json", "benchmark"] + terminal = json.loads(run_source_cli([ + *cli, "terminal-reconcile", "--goal-id", args["goal_id"], + "--request-json", str(request), "--execute", + ]).stdout) + assert terminal["released"] + assert datetime.fromisoformat(terminal["row"]["observed_at"]) > datetime.fromisoformat(running["observed_at"]) + merged = json.loads(run_source_cli([ + *cli, "experiment-board-reconcile", "--goal-id", args["goal_id"], + "--source-ledger", str(shard), "--execute", + ]).stdout) + assert merged["ok"] and merged["runs"][0]["status"] == "completed" + assert [x["run_id"] for x in ce.read_benchmark_concurrency_envelope(envelope)["active_runs"]] == ["other-run"] + + +@pytest.mark.parametrize("field,value", [("grading_drained", False), ("solver_terminal", False), + ("attempt", 2), ("run_id", "foreign"), ("case_id", "foreign")]) +def test_unknown_pending_or_wrong_identity_never_writes(campaign, field, value): + args, board, envelope, _ = campaign + args["terminal"][field] = value + before = (board.read_bytes(), envelope.read_bytes()) + with pytest.raises(ValueError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_runtime_unavailable_preserves_original_evidence(campaign, monkeypatch): + args, board, envelope, _ = campaign + before = (board.read_bytes(), envelope.read_bytes()) + def unavailable(*a, **kw): + raise tf.EffectRuntimeStartupError("fixture unavailable", diagnostic_code="fixture_unavailable") + monkeypatch.setattr(tf, "effect_runtime_result", unavailable) + with pytest.raises(ValueError, match="runtime unavailable; retry original"): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_old_admission_cannot_touch_replacement(campaign): + args, board, envelope, _ = campaign + ce.release_benchmark_case(envelope, run_id="run-a", execute=True) + ce.admit_benchmark_case(envelope, run_id="run-a", case_id="case-a", arm_role="baseline", + admitted_at="2026-01-01T00:30:00Z", execute=True) + before = (board.read_bytes(), envelope.read_bytes()) + with pytest.raises(ValueError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_same_slot_cannot_be_adopted_by_a_second_study(campaign): + args, board, envelope, _ = campaign + other = eb.read_benchmark_experiment_board_rows(board)[0] + other["study_id"] = "different-study" + eb.upsert_benchmark_experiment_board_row(board, other) + before = (board.read_bytes(), envelope.read_bytes()) + with pytest.raises(ValueError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_idempotent_admission_is_not_original_launch_evidence(campaign): + args, board, envelope, _ = campaign + receipt = ce.admit_benchmark_case(envelope, run_id="run-a", case_id="case-a", + arm_role="baseline", execute=True) + assert receipt["admitted"] and receipt["idempotent"] + args["admission_receipt"] = {**receipt, "goal_id": args["goal_id"]} + before = (board.read_bytes(), envelope.read_bytes()) + with pytest.raises(ValueError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_write_failure_keeps_original_slot_then_replay_preserves_new_metrics(campaign, monkeypatch): + args, board, envelope, _ = campaign + original = tf.atomic_write_json + def fail(*a, **kw): + raise OSError("injected release failure") + monkeypatch.setattr(tf, "atomic_write_json", fail) + with pytest.raises(OSError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert eb.read_benchmark_experiment_board_rows(board)[0]["status"] == "completed" + assert len(ce.read_benchmark_concurrency_envelope(envelope)["active_runs"]) == 2 + row = eb.read_benchmark_experiment_board_rows(board)[0] + row.update(observed_at="2026-01-01T02:00:00Z", metrics={"score": {"value": 9, "higher_is_better": True}}, + insight={"status": "complete"}) + eb.upsert_benchmark_experiment_board_row(board, row) + monkeypatch.setattr(tf, "atomic_write_json", original) + result = tf.reconcile_benchmark_terminal(**args, execute=True) + assert result["released"] and result["row"]["metrics"]["score"]["value"] == 9 + assert result["row"]["insight"]["status"] == "complete" + + +def test_board_failure_or_readback_failure_cannot_release(campaign, monkeypatch): + args, board, envelope, _ = campaign + before = envelope.read_bytes() + monkeypatch.setattr(tf, "_replace_domain_state_jsonl", lambda *a: None) + with pytest.raises(ValueError, match="board readback"): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert envelope.read_bytes() == before + + +def test_concurrent_admission_between_plan_and_commit_fails_closed(campaign, monkeypatch): + args, board, envelope, _ = campaign + original = tf.effect_runtime_result + def interleave(*a, **kw): + result = original(*a, **kw) + ce.admit_benchmark_case(envelope, run_id="parallel", case_id="parallel-case", arm_role="baseline", execute=True) + return result + monkeypatch.setattr(tf, "effect_runtime_result", interleave) + before = board.read_bytes() + with pytest.raises(ValueError, match="inputs changed"): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert board.read_bytes() == before + assert len(ce.read_benchmark_concurrency_envelope(envelope)["active_runs"]) == 3 + monkeypatch.setattr(tf, "effect_runtime_result", original) + tf.reconcile_benchmark_terminal(**args, execute=True) + assert {x["run_id"] for x in ce.read_benchmark_concurrency_envelope(envelope)["active_runs"]} == {"other-run", "parallel"} + + +def test_missing_ledger_is_not_successful_replay(campaign): + args, _, envelope, _ = campaign + envelope.unlink() + with pytest.raises(ValueError, match="ledger unavailable"): + tf.reconcile_benchmark_terminal(**args, execute=True) + + +def test_wrong_goal_never_writes(campaign): + args, board, envelope, _ = campaign + args["admission_receipt"] = {**args["admission_receipt"], "goal_id": "foreign"} + before = (board.read_bytes(), envelope.read_bytes()) + with pytest.raises(ValueError): + tf.reconcile_benchmark_terminal(**args, execute=True) + assert before == (board.read_bytes(), envelope.read_bytes()) + + +def test_managed_cli_cannot_route_an_unregistered_project(campaign, tmp_path): + args, board, envelope, _ = campaign + empty_registry = tmp_path / "empty.json" + empty_registry.write_text('{"goals": []}') + # Remove the fixture's local route as well as the supplied registry route. + (args["project"] / ".loopx" / "registry.json").unlink() + request = tmp_path / "request.json" + request.write_text(json.dumps({k: args[k] for k in ("admission_receipt", "identity", "terminal")})) + before = (board.read_bytes(), envelope.read_bytes()) + result = run_source_cli([ + "--registry", str(empty_registry), "--format", "json", "benchmark", "terminal-reconcile", + "--project", str(args["project"]), "--goal-id", args["goal_id"], + "--request-json", str(request), "--execute", + ], check=False) + assert result.returncode != 0 + assert before == (board.read_bytes(), envelope.read_bytes()) diff --git a/tests/control_plane_ts/benchmark_terminal_followthrough.test.ts b/tests/control_plane_ts/benchmark_terminal_followthrough.test.ts new file mode 100644 index 0000000000..f0e6b45bad --- /dev/null +++ b/tests/control_plane_ts/benchmark_terminal_followthrough.test.ts @@ -0,0 +1,171 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {prepareBenchmarkTerminalFollowthrough as prepare} from "../../loopx/control_plane/capabilities/benchmark_terminal_followthrough.ts"; +import {createEffectRuntimeHandlers, dispatchEffectRuntimeMethod} from "../../loopx/control_plane/effect_runtime_handlers.ts"; + +function fixture() { + const identity = { + benchmark_id: "synthetic", study_id: "study", case_id: "case", run_id: "run", + arm_id: "reference", arm_role: "baseline", attempt: 1, + }; + const run = {run_id: "run", case_id: "case", arm_role: "baseline", admitted_at: "2026-01-01T01:00:00Z"}; + const row = { + ...identity, + schema_version: "benchmark_experiment_board_row_v0", + status: "running", observed_at: "2026-01-01T01:01:00+00:00", + model_id: "model", protocol_id: "protocol", comparison_protocol_id: "comparison", + claim_scope: "matched_study", primary_metric: "score", guardrail_metrics: [], + metrics: {}, countability: {integrity_qualified: false, official_result_present: false, score_countable: false}, + treatment_fidelity: "not_applicable", effort: {}, insight: {status: "pending"}, + }; + return { + goal_id: "goal", + reconciled_at: "2026-01-01T01:04:00+00:00", + admission_receipt: {ok: true, admitted: true, dry_run: false, write_performed: true, idempotent: false, + goal_id: "goal", run}, + identity, + rows: [row], + active_runs: [run], + terminal: {...identity, status: "completed", observed_at: "2026-01-01T01:02:00+00:00", + solver_terminal: true, grading_drained: true}, + runtime_observation: { + schema_version: "benchmark_runtime_observation_v0", admission_active: true, job_receipt_state: "resolved", + runner_owner_state: "unknown", terminal_result_present: true, reconciliation_required: true, + classification: "terminal_pending_reconcile", recommended_transition: "write_terminal_then_release", + }, + }; +} + +test("prepares the terminal board transition and exact release through the Effect runtime", async () => { + const request = fixture(); + const before = structuredClone(request); + const handlers = createEffectRuntimeHandlers({fingerprint: "test", requestShutdown() {}}); + assert.deepEqual(await dispatchEffectRuntimeMethod(handlers, "benchmark.terminal-followthrough.prepare", request), { + row: {...request.rows[0], status: "completed", observed_at: request.reconciled_at}, + active_run: request.admission_receipt.run, + already_released: false, + }); + assert.deepEqual(request, before); +}); + +test("a confirmed drained runner-invalid attempt can close an existing planned row", () => { + const request = fixture(); + request.rows[0].status = "planned"; + request.terminal.status = "runner_invalid"; + const result = prepare(request); + assert.equal(result.row.status, "runner_invalid"); + assert.deepEqual(result.row.countability, request.rows[0].countability); + assert.deepEqual(result.active_run, request.admission_receipt.run); +}); + +test("replay preserves late score and insight enrichment with and without the admitted slot", () => { + const request = fixture(); + const row = { + ...request.rows[0], status: "completed", observed_at: "2026-01-01T01:03:00+00:00", + metrics: {score: {value: 0, unit: "points", higher_is_better: true}}, + countability: {integrity_qualified: true, official_result_present: true, score_countable: true}, + insight: {status: "complete"}, + }; + const replay = {...request, rows: [row], reconciled_at: row.observed_at}; + assert.deepEqual(prepare(replay), { + row, active_run: request.admission_receipt.run, already_released: false, + }); + const released = {...request.runtime_observation, admission_active: false, reconciliation_required: false, + classification: "not_admitted", recommended_transition: "none"}; + assert.deepEqual(prepare({...replay, active_runs: [], runtime_observation: released}), { + row, active_run: null, already_released: true, + }); + assert.throws(() => prepare({...replay, runtime_observation: released}), + /resolved terminal runtime evidence/); +}); + +test("terminal status alone cannot close or release a run with unknown runtime or grading", () => { + const request = fixture(); + for (const flags of [{solver_terminal: false}, {grading_drained: false}, {grading_drained: null}]) { + assert.throws(() => prepare({...request, terminal: {...request.terminal, ...flags}}), /terminal solver and drained grading/); + } + for (const evidence of [{job_receipt_state: "ambiguous"}, {terminal_result_present: false}, + {classification: "runner_lost_pending_reconcile", recommended_transition: "write_runner_invalid_then_release"}]) { + assert.throws(() => prepare({...request, runtime_observation: {...request.runtime_observation, ...evidence}}), + /resolved terminal runtime evidence/); + } +}); + +test("requires the original executed admission receipt for the same goal and run", () => { + const request = fixture(); + for (const fields of [{dry_run: true}, {admitted: false}, {goal_id: "other-goal"}, + {idempotent: true}, {write_performed: false}, {idempotent: true, write_performed: false}]) { + assert.throws(() => prepare({...request, admission_receipt: {...request.admission_receipt, ...fields}}), + /original executed admission for this goal/); + } + const run = {...request.admission_receipt.run, case_id: "other-case"}; + assert.throws(() => prepare({...request, admission_receipt: {...request.admission_receipt, run}}), /does not match the run identity/); +}); + +test("an enriched terminal row does not authorize releasing a reused or ambiguous slot", () => { + const request = fixture(); + request.rows[0].status = "completed"; + const [slot] = request.active_runs; + for (const replacement of [{admitted_at: "2026-01-01T01:00:00.000001Z"}, {case_id: "another-case"}, {arm_role: "explore"}]) { + assert.throws(() => prepare({...request, active_runs: [{...slot, ...replacement}]}), + {kind: "conflict", message: "active slot does not match the executed admission"}); + } + assert.throws(() => prepare({...request, active_runs: [slot, slot]}), /active slot does not match/); +}); + +test("rejects absent, mismatched and ambiguous board identities without creating a row", () => { + const request = fixture(); + const [row] = request.rows; + for (const rows of [[], [{...row, attempt: 2}], [{...row, arm_id: "other-arm"}], [row, {...row, arm_id: "other-arm"}]]) { + assert.throws(() => prepare({...request, rows}), /one existing exact board row/); + } + assert.throws(() => prepare({...request, terminal: {...request.terminal, attempt: 2}}), + /terminal evidence does not match the run identity/); +}); + +test("a goal slot cannot disambiguate the same run ID across studies or benchmarks", () => { + const request = fixture(); + const [row] = request.rows; + for (const scope of [{study_id: "other-study"}, {benchmark_id: "other-benchmark"}]) { + assert.throws(() => prepare({...request, rows: [row, {...row, ...scope}]}), /one existing exact board row/); + } + const other = {...row, study_id: "other-study", run_id: "other-run"}; + assert.deepEqual(prepare({...request, rows: [row, other]}).active_run, request.admission_receipt.run); +}); + +test("a missing slot cannot advance a board row and a conflicting terminal status cannot release a slot", () => { + const request = fixture(); + assert.throws(() => prepare({...request, active_runs: []}), /terminal board transition requires the admitted slot/); + assert.throws(() => prepare({...request, rows: [{...request.rows[0], status: "cancelled"}]}), + /existing board status conflicts/); +}); + +test("rejects terminal evidence older than the executed admission", () => { + const request = fixture(); + const terminal = {...request.terminal, observed_at: "2026-01-01T00:59:59.999999Z"}; + assert.throws(() => prepare({...request, terminal}), /terminal observation predates admission/); +}); + +test("delayed terminal closeout uses host time after the newer running snapshot", () => { + const request = fixture(); + const row = {...request.rows[0], observed_at: "2026-01-01T09:02:00.000001+08:00"}; + assert.deepEqual(prepare({...request, rows: [row]}), { + row: {...row, status: "completed", observed_at: request.reconciled_at}, + active_run: request.admission_receipt.run, already_released: false, + }); +}); + +test("a host timestamp at or before the running snapshot cannot order the terminal transition", () => { + const request = fixture(); + const row = {...request.rows[0], observed_at: "2026-01-01T09:02:00.000001+08:00"}; + for (const reconciled_at of ["2026-01-01T01:02:00.000001Z", "2026-01-01T01:02:00Z"]) { + assert.throws(() => prepare({...request, rows: [row], reconciled_at}), + /host reconciliation time must follow the existing board observation/); + } +}); + +test("provider proof cannot postdate the host reconciliation time", () => { + const request = fixture(); + const terminal = {...request.terminal, observed_at: "2026-01-01T01:04:00.000001Z"}; + assert.throws(() => prepare({...request, terminal}), /terminal observation follows host reconciliation time/); +}); From 29a81dddbfd34df82220684f3cb5c93b179385cd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:46:04 +0800 Subject: [PATCH 2/3] fix(benchmark): retain managed terminal receipt on atomic write failure Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/edgebench/managed_run.py | 14 ++++++- benchmark/edgebench/run.py | 1 - tests/benchmark/test_edgebench_managed_run.py | 37 +++++++++++++++++-- 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/benchmark/edgebench/managed_run.py b/benchmark/edgebench/managed_run.py index 34ac48f68b..212460b7fe 100644 --- a/benchmark/edgebench/managed_run.py +++ b/benchmark/edgebench/managed_run.py @@ -166,7 +166,19 @@ def retain_managed_solver(trial: Path, binding: dict, launch: dict, *, receipt: if (execution == result.solver_execution and _solver_ended( execution, admitted_at=admitted.timestamp(), observed_at=datetime.now(UTC).timestamp())): retained["solver_execution_sha256"] = digest - _write_runtime_receipt(trial, receipt, retained) + try: + _write_runtime_receipt(trial, receipt, retained) + except OSError: + # An error after replacement can leave the exact terminal record visible. + # An error before replacement allows one retry from the original launch. + observed, _ = _read_record(trial / RUNTIME_FILENAME, private=True) + expected = {**receipt, "managed_run": retained} + if json.dumps(observed) != json.dumps(expected): + if (observed.get("status") != "starting" or observed.get("managed_run") != launch + or _runtime_digest(observed) != launch["launch_runtime_sha256"] + or any(observed.get(key) != receipt.get(key) for key in ("run_id", "task", "feedback"))): + raise + _write_runtime_receipt(trial, receipt, retained) return "solver_execution_sha256" in retained diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index eb686ab247..5a37cb4853 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -586,7 +586,6 @@ def main(argv=None): try: solver_retained = retain_managed_solver(trial, managed_binding, managed_launch, receipt=receipt, result=result) except (OSError, TypeError, ValueError) as error: - receipt_path.write_text(json.dumps(receipt, indent=2)) logger.warning("Managed solver evidence unavailable (%s)", type(error).__name__) else: try: diff --git a/tests/benchmark/test_edgebench_managed_run.py b/tests/benchmark/test_edgebench_managed_run.py index cdcf888d3e..39c7b80bb6 100644 --- a/tests/benchmark/test_edgebench_managed_run.py +++ b/tests/benchmark/test_edgebench_managed_run.py @@ -16,7 +16,7 @@ import pytest from benchmark.edgebench import managed_run -from benchmark.edgebench.managed_run import BINDING_FILENAME, TERMINAL_FILENAME +from benchmark.edgebench.managed_run import BINDING_FILENAME, RUNTIME_FILENAME, TERMINAL_FILENAME from loopx.capabilities.benchmark_toolkit import ( BENCHMARK_EXPERIMENT_BOARD_ROW_SCHEMA_VERSION, build_benchmark_concurrency_config, @@ -140,7 +140,8 @@ def _replay(trial: Path, *, execute: bool = False) -> subprocess.CompletedProces capture_output=True, text=True, timeout=30) -@pytest.mark.parametrize("stage", ["drained", "pending", "write_failure"]) +@pytest.mark.parametrize("stage", ["drained", "pending", "write_failure", + "receipt_before_replace", "receipt_after_replace", "receipt_persistent"]) def test_managed_terminal_flows_from_runner_to_controller(tmp_path, entry, capsys, monkeypatch, stage): trial, invoke, drain = entry project = tmp_path / "controller-project" @@ -173,18 +174,48 @@ def controller(command: str, *args: str) -> dict: "insight": {"status": "pending"}, })["row"] + terminal_receipt_attempts = [] with monkeypatch.context() as patch: if stage == "write_failure": def fail_write(*args, **kwargs): raise OSError("fixture proof storage failure") patch.setattr(managed_run, "atomic_write_state_text", fail_write) + if stage.startswith("receipt_"): + write_json_atomic = managed_run.write_json_atomic + + def fail_terminal_receipt_once(path, payload, **kwargs): + terminal_receipt = path.name == RUNTIME_FILENAME and "solver_execution_sha256" in payload.get("managed_run", {}) + if terminal_receipt: + terminal_receipt_attempts.append(payload) + if len(terminal_receipt_attempts) == 1 or stage == "receipt_persistent": + if stage == "receipt_after_replace": + write_json_atomic(path, payload, **kwargs) + raise OSError("fixture terminal receipt storage failure") + return write_json_atomic(path, payload, **kwargs) + + patch.setattr(managed_run, "write_json_atomic", fail_terminal_receipt_once) assert invoke(_binding_file(tmp_path, admission), evidence=stage) == 0 runtime = json.loads(capsys.readouterr().out) assert runtime["status"] == "terminal" and runtime["best_score"] == 4 assert GOAL not in json.dumps(runtime) and str(project) not in json.dumps(runtime) proof = trial / TERMINAL_FILENAME - if stage != "drained": + if stage == "receipt_persistent": + retained = json.loads((trial / RUNTIME_FILENAME).read_bytes()) + assert retained["status"] == "starting" and retained["run_id"] == admission["run"]["run_id"] + assert "solver_execution_sha256" not in retained["managed_run"] + assert len(terminal_receipt_attempts) == 2 and not proof.exists() + assert _replay(trial, execute=True).returncode == 2 + assert read_benchmark_experiment_board_rows(board) == [running] + assert read_benchmark_concurrency_envelope(envelope)["active_runs"] == [admission["run"]] + return + if stage.startswith("receipt_"): + retained = json.loads((trial / RUNTIME_FILENAME).read_bytes()) + assert retained["managed_run"]["solver_execution_sha256"] + assert retained["run_id"] == admission["run"]["run_id"] and retained["status"] == "terminal" + assert len(terminal_receipt_attempts) == (2 if stage == "receipt_before_replace" else 1) + assert all(attempt == retained for attempt in terminal_receipt_attempts) + if stage in {"pending", "write_failure"}: assert not proof.exists() if stage == "pending": pending = _replay(trial, execute=True) From c7e63ca2fecbab9c17dcb6da40d3e5ce8318aa2b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 18:23:21 +0800 Subject: [PATCH 3/3] fix(benchmark): replay a completed attempt after receipt storage recovers Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/edgebench/README.md | 7 +++- benchmark/edgebench/managed_run.py | 38 +++++++++++++++++-- tests/benchmark/test_edgebench_managed_run.py | 37 +++++++++++++++--- 3 files changed, 70 insertions(+), 12 deletions(-) diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index 19e8a1f858..237e7e5841 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -883,8 +883,11 @@ python -m benchmark.edgebench.managed_run --trial "$ATTEMPT_DIR" --execute ``` The replay reads retained evidence. Pending, unknown, or mismatched native -evidence leaves the reservation active. Use the proof from the exact attempt on -the controller and retain the command's JSON receipt with its binding. +evidence leaves the reservation active. After a terminal receipt write fails, +replay can use the runner's completed result when it agrees with the original +launch pins, the exact native solver record, and the drained grading history. +Use the proof from the exact attempt on the controller and retain the command's +JSON receipt with its binding. ```sh loopx benchmark terminal-reconcile --goal-id "$GOAL_ID" --project "$PROJECT" \ diff --git a/benchmark/edgebench/managed_run.py b/benchmark/edgebench/managed_run.py index 212460b7fe..70b2da0a21 100644 --- a/benchmark/edgebench/managed_run.py +++ b/benchmark/edgebench/managed_run.py @@ -11,6 +11,7 @@ from datetime import UTC, datetime import hashlib import json +import math import os from pathlib import Path import stat @@ -26,6 +27,7 @@ from loopx.control_plane.runtime.document_io import atomic_write_state_text from loopx.file_lock import write_json_atomic from loopx.public_safe_text import LETTER_LED_COMPACT_TOKEN_PATTERN +from .result_publication import native_grading_complete BINDING_FILENAME = "managed-run.json" @@ -211,6 +213,22 @@ def _grading_drained(trial: Path, *, history_file: str, run_id: str, task_id: st and all(_terminal_history_entry(entry, task_id=task_id) for entry in entries)) +def _retained_completed_result(trial: Path, receipt: dict) -> dict | None: + """Read the runner's completion marker when its terminal receipt did not land.""" + try: + result, _ = _read_record(trial / "final_result.json") + except FileNotFoundError: + return None + duration = result.get("runtime_seconds") + if (any(result.get(field) != receipt.get(field) for field in ("run_id", "task", "model", "effort")) + or result.get("runtime_status", "terminal") != "terminal" + or result.get("evaluation_coverage") == "online_only" + or not native_grading_complete(result) + or type(duration) not in (int, float) or not math.isfinite(duration) or duration <= 0): + return None + return result + + def publish_managed_terminal(trial: Path, *, execute: bool = False) -> bool: """Check retained native evidence and optionally publish this attempt's proof.""" if (trial.parent.parent.name != "runs" or not trial.is_dir() @@ -224,16 +242,28 @@ def publish_managed_terminal(trial: Path, *, execute: bool = False) -> bool: raise ValueError("Managed runtime receipt differs from the selected attempt") status = _BOARD_STATUS.get(receipt.get("status")) retained = receipt.get("managed_run") - if status is None or not isinstance(retained, dict) or "solver_execution_sha256" not in retained: + if not isinstance(retained, dict): + return False + recovered_completion = None + if receipt.get("status") == "starting": + recovered_completion = _retained_completed_result(trial, receipt) + if recovered_completion is None: + return False + status = _BOARD_STATUS["terminal"] + elif status is None or "solver_execution_sha256" not in retained: return False history_file = retained.get("history_file") if history_file not in _NATIVE_HISTORY_FILES: raise ValueError("Retained managed grading history path differs") execution, solver_digest = _read_record(trial / "solver_execution.json") + if recovered_completion is None: + solver_matches = retained.get("solver_execution_sha256") == solver_digest + else: + solver_matches = json.dumps(execution, sort_keys=True) == json.dumps( + recovered_completion.get("solver_execution"), sort_keys=True) admitted = _timestamp(binding["admission_receipt"]["run"]["admitted_at"]) - if (retained.get("solver_execution_sha256") != solver_digest - or not _solver_ended(execution, admitted_at=admitted.timestamp(), - observed_at=datetime.now(UTC).timestamp())): + if not solver_matches or not _solver_ended(execution, admitted_at=admitted.timestamp(), + observed_at=datetime.now(UTC).timestamp()): raise ValueError("Retained solver termination evidence differs") if not _grading_drained(trial, history_file=history_file, run_id=run_id, task_id=task_id): return False diff --git a/tests/benchmark/test_edgebench_managed_run.py b/tests/benchmark/test_edgebench_managed_run.py index 39c7b80bb6..973d5d13dd 100644 --- a/tests/benchmark/test_edgebench_managed_run.py +++ b/tests/benchmark/test_edgebench_managed_run.py @@ -201,15 +201,40 @@ def fail_terminal_receipt_once(path, payload, **kwargs): assert GOAL not in json.dumps(runtime) and str(project) not in json.dumps(runtime) proof = trial / TERMINAL_FILENAME if stage == "receipt_persistent": - retained = json.loads((trial / RUNTIME_FILENAME).read_bytes()) + runtime_path = trial / RUNTIME_FILENAME + original_runtime = runtime_path.read_bytes() + retained = json.loads(original_runtime) assert retained["status"] == "starting" and retained["run_id"] == admission["run"]["run_id"] assert "solver_execution_sha256" not in retained["managed_run"] assert len(terminal_receipt_attempts) == 2 and not proof.exists() - assert _replay(trial, execute=True).returncode == 2 - assert read_benchmark_experiment_board_rows(board) == [running] - assert read_benchmark_concurrency_envelope(envelope)["active_runs"] == [admission["run"]] - return - if stage.startswith("receipt_"): + final_path = trial / "final_result.json" + original_final = final_path.read_bytes() + final = json.loads(original_final) + assert final["solver_execution"] == json.loads((trial / "solver_execution.json").read_bytes()) + final_path.unlink() + assert _replay(trial, execute=True).returncode == 2 and not proof.exists() + for change in ( + {"task": "other-task"}, + {"runtime_status": "cancelled"}, + {"solver_execution": final["solver_execution"] | {"status": "nonzero_exit"}}, + ): + final_path.write_text(json.dumps(final | change)) + assert _replay(trial, execute=True).returncode == 2 and not proof.exists() + final_path.write_bytes(original_final) + history = trial / "run_history.json" + original_history = history.read_bytes() + history.write_text(json.dumps(json.loads(original_history) | {"run_id": "different-run"})) + assert _replay(trial, execute=True).returncode == 2 and not proof.exists() + history.write_bytes(original_history) + runtime_path.write_text(json.dumps(retained | {"runner_commit": "different-runner"})) + assert _replay(trial, execute=True).returncode == 2 and not proof.exists() + runtime_path.write_bytes(original_runtime) + preview = _replay(trial) + assert preview.returncode == 0 and json.loads(preview.stdout) == {"ready": True, "dry_run": True} + assert runtime_path.read_bytes() == original_runtime and not proof.exists() + assert _replay(trial, execute=True).returncode == 0 + assert runtime_path.read_bytes() == original_runtime + elif stage.startswith("receipt_"): retained = json.loads((trial / RUNTIME_FILENAME).read_bytes()) assert retained["managed_run"]["solver_execution_sha256"] assert retained["run_id"] == admission["run"]["run_id"] and retained["status"] == "terminal"