diff --git a/benchmark/edgebench/execution_commitment.py b/benchmark/edgebench/execution_commitment.py new file mode 100644 index 0000000000..4f3066c4c1 --- /dev/null +++ b/benchmark/edgebench/execution_commitment.py @@ -0,0 +1,192 @@ +"""Native SForge IO for the toolkit's existing duration and Goal acceptance owners. + +The host selects exact solver receipts and supplies real process liveness. No +clock or continuation state is added to the worker or its checkpoint capsule. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import datetime, timezone +import hashlib +import json +import math +from pathlib import Path +import shutil +import subprocess +import time + + +class NativeExecutionEvidence: + def __init__( + self, + *, + run_id: str, + task_id: str, + trial: Path, + target_seconds: int, + runner_active: Callable[[], bool], + node: str | None = None, + previous: tuple[tuple[Path, str], ...] = (), + ): + if type(target_seconds) is not int or target_seconds <= 0: + raise ValueError("A finite positive solver allowance is required") + self.run_id, self.task_id = run_id, task_id + self.trial, self.target_seconds = Path(trial), target_seconds + self.runner_active, self.previous = runner_active, previous + self.node = node or shutil.which("node") + if not self.node: + raise ValueError("The duration owner requires the qualified Node runtime") + + def fulfilled(self) -> bool | None: + """Project native intervals through the same owner used by campaign reports.""" + now = time.time() + intervals = [] + active = self.runner_active() + if type(active) is not bool: + return None + sources = [*self.previous, (self.trial / "solver_execution.json", None)] + for path, digest in sources: + if path.is_symlink() or not path.is_file() or path.stat().st_size > 65536: + return None + raw = path.read_bytes() + if digest is not None and hashlib.sha256(raw).hexdigest() != digest: + return None + execution = json.loads(raw) + if ( + not isinstance(execution, dict) + or execution.get("schema_version") != 1 + or execution.get("status") == "unknown" + ): + return None + segments = execution.get("segments") + if not isinstance(segments, list) or not segments: + return None + for segment in segments: + if not isinstance(segment, dict): + return None + start, end = segment.get("started_at"), segment.get("finished_at") + if end is None: + if ( + digest is not None + or not active + or segment.get("status") != "pending" + ): + return None + end = now + if any( + type(value) not in (int, float) or not math.isfinite(value) + for value in (start, end) + ): + return None + if not 0 <= start <= end <= now: + return None + # The typed owner parses observed_at at millisecond precision. + # Round inward so an active tail is never in the future and + # fractional setup/teardown time cannot become execution credit. + first, last = math.ceil(start * 1000), math.floor(end * 1000) + if first > last: + continue # A sub-millisecond segment contributes no credit. + intervals.append( + dict( + start_ms=first, + end_ms=last, + evidence_ref="native:solver-execution", + ) + ) + request = dict( + schema_version="benchmark_campaign_report_input_v1", + observed_at=datetime.fromtimestamp(now, timezone.utc).isoformat(), + title="Execution commitment", + case_ids=[self.task_id], + arms=[dict(arm_id="worker", label="Worker")], + selection=[ + dict( + arm_id="worker", + case_id=self.task_id, + run_id=self.run_id, + status="running" if active else "completed", + score=None, + protocol_id="native", + exclude_reason=None, + rerun_reason=None, + active_intervals=intervals, + ) + ], + incidents=[], + notes=[], + execution_audit=[], + duration_policy=dict( + target_ms=self.target_seconds * 1000, allow_goal_completion=[] + ), + ) + owner = ( + Path(__file__).resolve().parents[2] / "scripts/benchmark-campaign-report.ts" + ) + code = ( + "import {readFileSync} from 'node:fs'; import {buildReport} from " + + json.dumps(owner.as_uri()) + + "; console.log(JSON.stringify(buildReport(" + "JSON.parse(readFileSync(0,'utf8'))).duration_coverage[0]));" + ) + result = subprocess.run( + [self.node, "--input-type=module", "-e", code], + input=json.dumps(request), + capture_output=True, + text=True, + timeout=2, + check=True, + ) + coverage = json.loads(result.stdout) + return None if coverage["active_ms"] is None else coverage["remaining_ms"] == 0 + + +def install_execution_commitment( + app, *, run_id: str, task_id: str, observe: Callable[[], bool | None] +) -> None: + """Install an explicit read-only extension on one native feedback relay. + + This packet is local to the SForge provider. It adds no Core vocabulary or + scoring operation; unrelated registrations and default-off apps keep their + original routes and behavior. + """ + from fastapi import HTTPException + + if not run_id or not task_id or not callable(observe): + raise ValueError("An exact native identity and observer are required") + identity = (run_id, task_id) + observers = getattr(app.state, "execution_commitment_observers", None) + if observers is not None: + previous = observers.get(identity) + if previous is not None and previous != observe: + raise ValueError("Native execution observer is already installed") + observers[identity] = observe + return + observers = {identity: observe} + app.state.execution_commitment_observers = observers + + @app.get("/api/v1/feedback/{relay_id}/execution-commitment") + def execution_commitment(relay_id: str): + try: + _, info = app.state.judge.resolve_feedback_relay(relay_id) + except KeyError: + raise HTTPException( + status_code=404, detail="Feedback relay not found" + ) from None + observer = observers.get((info["run_id"], info["task_id"])) + if observer is None: + raise HTTPException(status_code=404, detail="Feedback relay not found") + try: + fulfilled = observer() + except (OSError, ValueError, subprocess.SubprocessError): + fulfilled = None + if fulfilled is not None and type(fulfilled) is not bool: + raise HTTPException( + status_code=503, detail="Native execution observation unavailable" + ) + return dict( + schema_version="sforge_execution_commitment_observation_v0", + task_id=info["task_id"], + observed_at_epoch=time.time(), + fulfilled=fulfilled, + ) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 65ebdd8b4d..f2aa6517d8 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -385,6 +385,41 @@ Budget exhaustion is distinct from task completion: this clock does not override provider failures, Stop, blocked planning, or a valid terminal LoopX decision. Campaign duration coverage must still classify an early exit as missing time. +### Optional native execution evidence + +A benchmark host can expose finalized and live solver intervals through the +existing campaign duration owner. This read-only provider does not declare +Goal requirements, configure acceptance checks or decide whether to continue. +The Agent owns semantic requirements in the shared first-planning checkpoint; +independent executable checks remain optional. A maximum timeout alone never +creates a minimum execution commitment. + +```python +from benchmark.edgebench.execution_commitment import ( + NativeExecutionEvidence, install_execution_commitment, +) + +evidence = NativeExecutionEvidence( + run_id=run_id, task_id=task_id, trial=trial, + target_seconds=explicit_caller_target_seconds, + runner_active=lambda: worker is not None and worker.poll() is None, +) +install_execution_commitment( + app, run_id=run_id, task_id=task_id, observe=evidence.fulfilled, +) +``` + +Use the native trial directory containing `solver_execution.json` and the exact +worker process. For continuation, supply finalized earlier receipts as explicit +`(Path, sha256)` pairs in `previous`. Setup, grading and downtime contribute no +execution credit; missing, altered or unfinished previous evidence stays unknown. +The scoped feedback route resolves the exact `(run_id, task_id)` pair. Reinstalling +the same observer is idempotent; replacing it is rejected. The command adapter +`python -m benchmark.runtime.execution_commitment probe|check --task-id TASK` +reads the explicitly selected native relay and rejects stale or malformed data. +Apps without this extension and ordinary worker entry/restore behavior remain +unchanged. No default semantic requirement or automatic check is installed. + To compare entry policies, hold the execution mode, session policy, model, effort, tools, feedback and total budget fixed, and use separate trials: diff --git a/benchmark/runtime/execution_commitment.py b/benchmark/runtime/execution_commitment.py new file mode 100644 index 0000000000..f425151d28 --- /dev/null +++ b/benchmark/runtime/execution_commitment.py @@ -0,0 +1,90 @@ +"""Read a scoped commitment observation from the selected native controller. + +No clock, scheduler or continuation policy lives here. The native provider +supplies execution evidence; Core owns Goal acceptance and subsequent work. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import time +from urllib.parse import urlsplit +from urllib.request import ProxyHandler, Request, build_opener + + +def validate_feedback_endpoint(endpoint: str) -> str: + url = urlsplit(endpoint) + if ( + url.scheme not in {"http", "https"} + or not url.netloc + or url.query + or url.fragment + or url.username + or url.password + or not url.path.startswith("/api/v1/feedback/") + or not url.path.removeprefix("/api/v1/feedback/").strip("/") + ): + raise ValueError("A scoped native feedback endpoint is required") + return endpoint.rstrip("/") + + +def observe(task_id: str, endpoint: str | None = None) -> bool | None: + endpoint = validate_feedback_endpoint( + endpoint if endpoint is not None else os.environ.get("SFORGE_FEEDBACK_URL", "") + ) + if not task_id: + raise ValueError("A native task identity is required") + request = Request( + endpoint.rstrip("/") + "/execution-commitment", + headers={"Cache-Control": "no-cache"}, + ) + # Native judge traffic already has its own admitted network boundary. Do + # not forward its scoped relay capability through the model API proxy. + with build_opener(ProxyHandler({})).open(request, timeout=3) as response: + data = response.read(8193) + if len(data) > 8192: + raise ValueError("Execution observation is oversized") + value = json.loads(data) + if not isinstance(value, dict): + raise ValueError("Execution observation must be an object") + observed = value.get("observed_at_epoch") + if ( + value.get("schema_version") != "sforge_execution_commitment_observation_v0" + or value.get("task_id") != task_id + or type(observed) not in (int, float) + or not math.isfinite(observed) + or abs(time.time() - observed) > 30 + or value.get("fulfilled") is not None + and type(value["fulfilled"]) is not bool + ): + raise ValueError( + "Execution observation is unavailable or belongs to another task" + ) + return value["fulfilled"] + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=("probe", "check")) + parser.add_argument("--task-id", required=True) + parser.add_argument( + "--feedback-url", help="Host-pinned native relay; not model API routing" + ) + args = parser.parse_args(argv) + try: + fulfilled = observe(args.task_id, args.feedback_url) + except (OSError, ValueError, KeyError, TypeError): + print(json.dumps({"available": False})) + return 1 + if args.action == "probe": + print(json.dumps({"available": fulfilled is not None})) + return 0 if fulfilled is not None else 1 + print(json.dumps({"commitment_fulfilled": fulfilled is True})) + return 0 if fulfilled is True else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmark/tests/test_execution_commitment.py b/benchmark/tests/test_execution_commitment.py new file mode 100644 index 0000000000..2bb234c09b --- /dev/null +++ b/benchmark/tests/test_execution_commitment.py @@ -0,0 +1,153 @@ +"""A native observation can attest execution; elapsed downtime cannot.""" + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +import subprocess +import sys +import threading +import time + +import pytest + +from benchmark.runtime.execution_commitment import observe + + +@pytest.fixture +def native(monkeypatch): + state = { + "schema_version": "sforge_execution_commitment_observation_v0", + "task_id": "synthetic-task", + "observed_at_epoch": time.time(), + "fulfilled": False, + } + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + assert self.path == "/api/v1/feedback/only-this-trial/execution-commitment" + raw = state["raw"] if "raw" in state else json.dumps(state).encode() + self.send_response(200) + self.end_headers() + self.wfile.write(raw) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + monkeypatch.setenv( + "SFORGE_FEEDBACK_URL", + f"http://127.0.0.1:{server.server_port}/api/v1/feedback/only-this-trial", + ) + try: + yield state + finally: + server.shutdown() + server.server_close() + thread.join(5) + + +def cli(action): + return subprocess.run( + [ + sys.executable, + "-m", + "benchmark.runtime.execution_commitment", + action, + "--task-id", + "synthetic-task", + ], + capture_output=True, + text=True, + timeout=5, + ) + + +def test_completion_requires_an_independent_native_true(native): + assert observe("synthetic-task") is False + assert cli("probe").returncode == 0 + assert cli("check").returncode == 1 + native["fulfilled"] = None + assert observe("synthetic-task") is None + assert cli("check").returncode == 1 + native["fulfilled"] = True + accepted = cli("check") + assert accepted.returncode == 0 + assert json.loads(accepted.stdout) == {"commitment_fulfilled": True} + + +@pytest.mark.parametrize( + "change", + [ + {"task_id": "another-trial"}, + {"schema_version": "unknown"}, + {"observed_at_epoch": time.time() - 120}, + {"observed_at_epoch": True}, + {"observed_at_epoch": float("nan")}, + {"fulfilled": 1}, + {"fulfilled": "true"}, + ], +) +def test_wrong_stale_and_malformed_observations_never_accept(native, change): + native.update(change) + with pytest.raises(ValueError): + observe("synthetic-task") + result = cli("check") + assert result.returncode == 1 + assert json.loads(result.stdout) == {"available": False} + + +@pytest.mark.parametrize("raw", [b"[]", b"null", b"not-json", b"{}", b"x" * 8193]) +def test_missing_or_invalid_provider_is_not_a_pass(native, raw): + native["raw"] = raw + assert cli("probe").returncode == 1 + assert cli("check").returncode == 1 + + +def test_native_capability_bypasses_model_api_proxy(native, monkeypatch): + monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:1") + monkeypatch.setenv("NO_PROXY", "") + assert cli("probe").returncode == 0 + + +def test_host_pinned_relay_survives_a_stale_inherited_service_environment( + native, monkeypatch +): + endpoint = os.environ["SFORGE_FEEDBACK_URL"] + monkeypatch.setenv( + "SFORGE_FEEDBACK_URL", "http://127.0.0.1:1/api/v1/feedback/stale" + ) + assert observe("synthetic-task", endpoint) is False + result = subprocess.run( + [ + sys.executable, + "-m", + "benchmark.runtime.execution_commitment", + "probe", + "--task-id", + "synthetic-task", + "--feedback-url", + endpoint, + ], + capture_output=True, + text=True, + timeout=5, + ) + assert result.returncode == 0 and json.loads(result.stdout) == {"available": True} + + + + +@pytest.mark.parametrize( + "endpoint", + [ + "", + "file:///tmp/state", + "http://127.0.0.1/other", + "http://127.0.0.1/api/v1/feedback/key?secret=other", + ], +) +def test_unscoped_endpoints_are_rejected(monkeypatch, endpoint): + monkeypatch.setenv("SFORGE_FEEDBACK_URL", endpoint) + assert cli("probe").returncode == 1 diff --git a/benchmark/tests/test_native_execution_commitment.py b/benchmark/tests/test_native_execution_commitment.py new file mode 100644 index 0000000000..f26560638b --- /dev/null +++ b/benchmark/tests/test_native_execution_commitment.py @@ -0,0 +1,244 @@ +"""The native provider and report share one duration classification owner.""" + +import hashlib +import json +from types import SimpleNamespace + +import pytest + +from benchmark.edgebench.execution_commitment import ( + NativeExecutionEvidence, + install_execution_commitment, +) + + +def execution(path, segments, status="normal_exit"): + path.write_text( + json.dumps(dict(schema_version=1, status=status, segments=segments)) + ) + + +def segment(start, end=None): + return dict( + started_at=start, + **({"finished_at": end} if end is not None else {}), + status="normal_exit" if end is not None else "pending", + ) + + +def test_gaps_are_excluded_and_restore_reuses_pinned_native_intervals( + tmp_path, monkeypatch +): + monkeypatch.setattr( + "benchmark.edgebench.execution_commitment.time.time", lambda: 210 + ) + old = tmp_path / "previous.json" + execution(old, [segment(100, 104)]) + current = tmp_path / "solver_execution.json" + execution(current, [segment(200, 204)]) + owner = NativeExecutionEvidence( + run_id="run", + task_id="task", + trial=tmp_path, + target_seconds=10, + runner_active=lambda: False, + previous=((old, hashlib.sha256(old.read_bytes()).hexdigest()),), + ) + assert owner.fulfilled() is False # The 96-second outage is not useful execution. + execution(current, [segment(200, 206)]) + assert owner.fulfilled() is True + execution(old, [segment(100, 110)]) + assert owner.fulfilled() is None # Retained evidence cannot silently change. + + +def test_pending_native_interval_requires_exact_runner_liveness(tmp_path, monkeypatch): + now = [109] + monkeypatch.setattr( + "benchmark.edgebench.execution_commitment.time.time", lambda: now[0] + ) + execution(tmp_path / "solver_execution.json", [segment(100)], status="pending") + live = [True] + owner = NativeExecutionEvidence( + run_id="run", + task_id="task", + trial=tmp_path, + target_seconds=10, + runner_active=lambda: live[0], + ) + assert owner.fulfilled() is False + now[0] = 110 + assert owner.fulfilled() is True + live[0] = False + assert owner.fulfilled() is None + now[0] = 10000 + assert owner.fulfilled() is None + + +def test_unknown_missing_or_symlinked_solver_evidence_never_attests(tmp_path): + owner = NativeExecutionEvidence( + run_id="run", + task_id="task", + trial=tmp_path, + target_seconds=10, + runner_active=lambda: True, + ) + assert owner.fulfilled() is None + path = tmp_path / "solver_execution.json" + execution(path, [segment(100, 200)], status="unknown") + assert owner.fulfilled() is None + path.unlink() + other = tmp_path / "other.json" + execution(other, [segment(100, 200)]) + path.symlink_to(other) + assert owner.fulfilled() is None + + +@pytest.mark.parametrize("fraction", [0.0001, 0.0006, 0.0016, 0.9996]) +def test_fractional_native_clock_does_not_create_a_future_tail( + tmp_path, monkeypatch, fraction +): + monkeypatch.setattr( + "benchmark.edgebench.execution_commitment.time.time", lambda: 110 + fraction + ) + execution(tmp_path / "solver_execution.json", [segment(100)], status="pending") + owner = NativeExecutionEvidence( + run_id="run", + task_id="task", + trial=tmp_path, + target_seconds=10, + runner_active=lambda: True, + ) + assert owner.fulfilled() is True + + +def test_short_native_segment_is_known_zero_credit(tmp_path, monkeypatch): + monkeypatch.setattr( + "benchmark.edgebench.execution_commitment.time.time", lambda: 110 + ) + execution(tmp_path / "solver_execution.json", [segment(100.0001, 100.0002)]) + owner = NativeExecutionEvidence( + run_id="run", + task_id="task", + trial=tmp_path, + target_seconds=10, + runner_active=lambda: False, + ) + assert owner.fulfilled() is False + + +def test_read_only_route_is_absent_by_default_and_scoped_to_original_relay(): + FastAPI = pytest.importorskip("fastapi").FastAPI + pytest.importorskip("httpx") + TestClient = pytest.importorskip("fastapi.testclient").TestClient + app = FastAPI() + with TestClient(app) as client: + assert ( + client.get("/api/v1/feedback/own/execution-commitment").status_code == 404 + ) + calls = [] + + def resolve(relay): + if relay == "missing": + raise KeyError(relay) + return "private-token", dict( + run_id="run" if relay == "own" else "other", task_id="task" + ) + + app.state.judge = SimpleNamespace(resolve_feedback_relay=resolve) + install_execution_commitment( + app, run_id="run", task_id="task", observe=lambda: calls.append(True) or False + ) + with TestClient(app) as client: + for relay in ["missing", "foreign"]: + assert ( + client.get(f"/api/v1/feedback/{relay}/execution-commitment").status_code + == 404 + ) + assert calls == [] + response = client.get("/api/v1/feedback/own/execution-commitment") + assert response.status_code == 200 and response.json()["fulfilled"] is False + assert "private-token" not in response.text + assert ( + client.post("/api/v1/feedback/own/execution-commitment").status_code == 405 + ) + assert calls == [True] + + +@pytest.mark.parametrize("order", [("a", "b", "c"), ("c", "b", "a")]) +def test_multiple_native_trials_keep_independent_routes_and_public_probes(order): + """A later valid registration must not be captured by the first route.""" + import socket + import subprocess + import sys + import threading + import time + from urllib.error import HTTPError + from urllib.request import urlopen + + FastAPI = pytest.importorskip("fastapi").FastAPI + uvicorn = pytest.importorskip("uvicorn") + app = FastAPI() + # A/B share a task; B/C share a run. Neither field alone identifies a trial. + identities = {"a": ("run-a", "task-a"), "b": ("run-b", "task-a"), + "c": ("run-b", "task-c"), "uninstalled": ("run-x", "task-x")} + states = {"a": False, "b": True, "c": None} + calls = [] + + def resolve(relay): + run, task = identities[relay] + return "private-token", dict(run_id=run, task_id=task) + + app.state.judge = SimpleNamespace(resolve_feedback_relay=resolve) + observers = {name: (lambda name=name: calls.append(name) or states[name]) + for name in order} + for name in (*order, *order): + run, task = identities[name] + install_execution_commitment(app, run_id=run, task_id=task, + observe=observers[name]) + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + server = uvicorn.Server(uvicorn.Config(app, log_level="error", access_log=False)) + thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True) + thread.start() + try: + deadline = time.monotonic() + 5 + while not server.started and thread.is_alive() and time.monotonic() < deadline: + time.sleep(0.01) + assert server.started + root = f"http://127.0.0.1:{sock.getsockname()[1]}/api/v1/feedback" + for name in order: + response = urlopen(f"{root}/{name}/execution-commitment", timeout=5) + packet = json.load(response) + assert packet["task_id"] == identities[name][1] + assert packet["fulfilled"] is states[name] + assert "private-token" not in json.dumps(packet) + for action in ("probe", "check"): + result = subprocess.run( + [sys.executable, "-m", "benchmark.runtime.execution_commitment", action, + "--task-id", identities[name][1], "--feedback-url", f"{root}/{name}"], + capture_output=True, text=True, timeout=5) + accepted = states[name] is not None if action == "probe" else states[name] is True + assert result.returncode == (0 if accepted else 1), result.stdout + previous_calls = list(calls) + for name in ("missing", "uninstalled"): + with pytest.raises(HTTPError) as error: + urlopen(f"{root}/{name}/execution-commitment", timeout=5) + assert error.value.code == 404 + assert calls == previous_calls + assert sorted(calls) == sorted(list(order) * 3) + with pytest.raises(ValueError, match="already installed"): + install_execution_commitment(app, run_id="run-b", task_id="task-a", + observe=lambda: False) + routes = [r for r in app.routes if r.path.endswith("/execution-commitment")] + assert len(routes) == 1 + # One app cannot inherit another app's registrations. + isolated = FastAPI() + isolated.state.judge = app.state.judge + TestClient = pytest.importorskip("fastapi.testclient").TestClient + with TestClient(isolated) as client: + assert client.get("/api/v1/feedback/a/execution-commitment").status_code == 404 + finally: + server.should_exit = True + thread.join(5) + sock.close() + assert not thread.is_alive()