fix(benchmark): isolate read-only native execution evidence by trial

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,
+        )

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:
 

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())