feat(benchmark): checkpoint research participants and continue native sessions

benchmark/README.md

@@ -113,3 +113,140 @@ The [DeepSWE GPT xhigh v1 archive note](deepswe-gptxhigh-versions.md) registers
 preserved for inspection; the companion notice governs their current evidence
 status. Neither archive category is the architecture for new work or grants
 runner, verifier, submission, or publication authority.
+
+## Inert checkpoint capture
+
+This is a developer prerequisite for the [research checkpoint contract](../docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md#77-research-state-branching-and-matched-continuation),
+not a qualified research continuation. The EdgeBench CLI accepts an explicit
+`--checkpoint-request /path/to/request.json` **host** file and stages its exact
+bytes into the heartbeat worker's protected control directory. The declaration
+is validated before trial creation/registration and its digest appears in the
+runtime receipt. The programmatic `BenchmarkCodex` and heartbeat `SForgeWorker`
+adapters also accept `checkpoint_request_path` to an already staged normalized
+absolute **guest** file. Ambient host variables do not enable capture; omitting
+these options disables the hook. Official and native-goal profiles reject them.
+
+The request contains exactly these fields; select only authorized private
+artifacts and exclude credentials and hidden evaluation material:
+
+```json
+{
+  "destination": "/logs/agent/checkpoints/research-point",
+  "components": [
+    {"id": "code", "source_path": "/workspace/solution.py"},
+    {"id": "evidence", "source_path": "/control/source-evidence.json"}
+  ],
+  "point_metadata": {"purpose": "inert source capture"}
+}
+```
+
+The destination must be a new direct child of the trial log owner's
+`checkpoints` directory. Capture runs after a successful execute wake's original
+settlement, before the next wake. It uses the remaining phase budget with a
+30-second capture ceiling, preserves SQLite's committed WAL through the existing
+backup owner, rejects observed source changes/symlinks and never overwrites a
+destination. Capture failures are recorded separately in the wake receipt;
+they do not replay work or change its settled outcome. The ceiling bounds
+publication eligibility and RPC waits, not an operating-system I/O cancellation
+guarantee. Capture cost remains part of the trial's elapsed time.
+
+For readback, call the existing `effect_runtime_result` adapter with method
+`runtime.checkpoint_artifacts`, action `verify`, the captured `root` and the
+receipt's `expected_sha256`. Action `materialize` with those fields and a new
+`destination` verifies and copies the unit into isolation. Neither action adopts
+state or starts execution. The new `loopx_checkpoint_artifact_set_v0` manifest
+is an internal artifact contract; `point_metadata` is unverified evidence.
+`research_point_qualified`, `execution_authority_granted` and `privacy_certified`
+remain false. Integrity checks cover the declared members and observed stable
+source boundary, not adversarial filesystem mutation or a cross-owner transaction.
+Native session reconstruction, effect reconciliation, fresh authority admission
+and evaluator cooldown/opportunity inheritance must pass before matched continuation.
+
+For pre-decision capture at a **fresh replan projected at the start of a heartbeat
+wake**, add `"trigger": "fresh_replan"` and set `destination` to the trial's
+`checkpoints` directory itself. The Host preloads that wake's original quota guard
+once, the typed owner validates its replan/settlement binding, and the adapter
+captures from the latest successful, settled work wake before invoking the model.
+The model reads the complete preloaded guard and follows its existing selection,
+required-read, recovery and settlement contract. Each obligation has one immutable
+destination; a repeated read verifies that point instead of replacing it with later
+work. The after-wake hook cannot substitute a post-decision state. Capture failures
+remain explicit and do not settle or complete the new work.
+The first attempt is also retained: if it fails and the model proceeds, retries
+cannot substitute a later state for that missed pre-decision point.
+
+Add `"native_session": true` to include Codex's completed native cursor, all
+retained history pages, rollout/compaction records and native index/history SQLite
+stores. The adapter uses native readback before sealing these members and excludes
+credential files, logs, memories and the symlinked skill installation. Record the
+effective configuration digest and Host version; tools/skills and credentials need
+their existing installation/authentication owners. Native state is opaque and its
+storage layout remains a pinned provider qualification.
+
+`benchmark.runtime.native_checkpoint.fork_native_checkpoint` verifies a unit and
+reconstructs a distinct native session in a **new isolated environment preserving
+the captured guest paths**, with matching configuration and Host version. It checks
+the full history digest and cursor, starts no model and registers no LoopX binding.
+A failed or partial reconstruction is not an admitted session. This is a session
+participant prerequisite, not full research continuation or arbitrary path import.
+
+For the native feedback protocol with **no automatic evaluations**, start an
+explicit checkpoint-aware native judge:
+
+```sh
+python -m benchmark.edgebench.checkpoint_judge --checkpoint-capture \
+  --tasks-dir /path/to/tasks --log-dir /path/to/private-run-logs
+```
+
+Its default-off factory otherwise preserves the native routes. With
+`"evaluation_opportunity": true` in the worker declaration, the capture hook
+reads that registration's released history and complete public result details,
+next round counters, accepted submission clock, remaining cooldown, task/native
+source identity and original trial deadline. It also preserves the native-selected
+incumbent's opaque submission archive as `evaluation_incumbent`, separate from the
+current working candidate. Native ranking chooses its original released round;
+the result identity, archive digest and byte count bind its download to the same
+judge epoch and consumed-round state. Missing or moved selected bytes reject
+capture; no selected result is explicitly recorded as absent. The transport never
+extracts archive members, overwrites a prior participant or silently accepts a
+truncated response. Its checked file witness must match the descriptor before
+the artifact owner publishes the unit. The provider uses the existing
+native admission/history locks and rejects an outstanding grade, unreconciled
+consumed round, missing result details, automatic-evaluation history or closed
+registration. It never returns tokens or hidden task material and never grades,
+registers or refills an opportunity. Download is limited to the same registration
+and a matching capture digest; source movement requires a new owned boundary.
+The local `sforge_checkpoint_opportunity_v0`
+artifact records these facts; it grants no restored admission. Disabling capture
+means omitting both the judge flag and worker declaration option on a new run.
+
+For an isolated single-Goal trial, `"loopx_state": true` adds the existing
+state-backup owner's inventory: registered active state, project research state,
+runtime authority/effect/quota journals and registry. The declared worker and
+physical routes must match that registry; missing registered state, incomplete
+inventory, symlink routes and narrower overlapping components reject capture.
+File and SQLite authority providers remain inert artifacts after materialization;
+copying their old leases or settlement records grants no new execution authority.
+
+Declare authorized workspace code/research files, effective configuration and
+installed source/profile pins as non-overlapping components. The native and
+opportunity participants do not imply that omitted state was saved. Every source
+remains private; inspect the verified inventory before qualifying a complete point.
+
+The current Host hook cannot capture a new replan created **inside an already
+running model wake** before that decision. Those boundaries still need a qualified
+Host handoff; a later snapshot must not be labeled as the missed point. Complete
+LoopX research-state adoption, restored evaluator admission and useful continued work
+remain open. No automatic retention or extra evaluation opportunity is introduced.
+
+这是研究 checkpoint 的开发前置切片。EdgeBench CLI 的显式
+`--checkpoint-request` 会把 host JSON 原文安全放入 heartbeat worker，并在
+runtime receipt 绑定摘要；programmatic adapter 也可指定已放入 guest 的请求
+路径。省略选项即关闭，Official/native-goal 不接受。产物
+验证和隔离物化不授予执行权，不代表研究点已资格化。请求中的 metadata 只是
+证据，不能代替会话、效果、状态采用及评测机会的联合验收。
+
+显式 `"loopx_state": true` 复用现有 state-backup inventory，捕获单 Goal 试验的
+注册状态、项目研究状态、runtime journal 和 registry；File/SQLite authority
+物化后仍是惰性历史。缺少来源、共享注册表、路径别名或重叠声明会拒绝，捕获
+不授予新的租约、配额、效果或执行权，不能替代完整接续验收。

benchmark/edgebench/checkpoint_judge.py

@@ -0,0 +1,183 @@
+"""Opt-in read-only checkpoint participant over the native SForge judge.
+
+Native admission, grading, history and ranking retain their owners. This
+adapter exports a quiescent registration's public history and opportunity
+clock; it cannot register, submit, restore or extend a trial.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import stat
+import time
+import uuid
+from dataclasses import replace
+from pathlib import Path
+from benchmark.runtime.sforge_checkpoint import ARCHIVE_ROUTE, ROUTE, SCHEMA
+
+
+def _read_archive(root, run, task, round_id):
+    """Read an opaque native archive, never extract or inspect its members."""
+    root = root.resolve(strict=True)
+    path = root / "runs" / run / task / "submissions" / round_id / "submission.tar.gz"
+    if not path.is_relative_to(root) or path.resolve(strict=True) != path:
+        raise ValueError("Native incumbent archive must be a physical file below the log owner")
+    with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW), "rb") as source:
+        before = os.fstat(source.fileno())
+        if not stat.S_ISREG(before.st_mode):
+            raise ValueError("Native incumbent archive must be a regular file")
+        raw = source.read()
+        after = os.fstat(source.fileno())
+    observed = path.stat(follow_symlinks=False)
+    def signature(value):
+        return value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns
+    if (signature(before) != signature(after) or signature(after) != signature(observed) or
+            path.resolve(strict=True) != path or len(raw) != before.st_size):
+        raise ValueError("Native incumbent archive moved during capture")
+    return raw
+
+
+def create_app(config, *, checkpoint_capture=False):
+    if type(checkpoint_capture) is not bool:
+        raise ValueError("checkpoint_capture must be an explicit boolean")
+    from sforge.harness.judge_server import SubmissionStatus, create_app as native_app
+    app = native_app(config)
+    if not checkpoint_capture:
+        return app  # Preserve the native route/admission surface when disabled.
+    from fastapi import HTTPException
+    from fastapi.responses import Response
+    from pydantic import BaseModel, ConfigDict
+    import sforge
+
+    state = getattr(app.state, "judge", None)
+    if state is None or not all(hasattr(state, key) for key in (
+            "_admission_lock", "_outstanding_by_run", "_tokens_lock", "_history_lock")):
+        raise ValueError("Checkpoint capture requires a pinned SForge with its native admission owner exposed")
+    epoch = uuid.uuid4().hex
+    source = hashlib.sha256()
+    source_root = Path(sforge.__file__).parent
+    for path in sorted(source_root.rglob("*.py")):
+        source.update(path.relative_to(source_root).as_posix().encode() + b"\0")
+        source.update(hashlib.sha256(path.read_bytes()).digest())
+    tasks = {key: {
+        "native_source_sha256": source.hexdigest(),
+        "task_spec_sha256": hashlib.sha256((config.tasks_dir / f"{key}.json").read_bytes()).hexdigest(),
+        "judge_image_key": task.judge_image_key,
+        "selection": task.judge.selection, "score_direction": task.judge.score_direction,
+    } for key, task in state.tasks.items()}
+
+    class ReadRequest(BaseModel):
+        model_config = ConfigDict(extra="forbid")
+        token: str
+
+    class ArchiveRequest(ReadRequest):
+        capture_binding_sha256: str
+
+    # FastAPI resolves annotations in module globals; keep this factory's
+    # private request type concrete rather than a forward reference.
+    def snapshot_locked(token):
+        from sforge.harness.selection import select_best
+
+        # Reuse the native lock order and outstanding admission counter. A
+        # pending grade or unexplained consumed round is never a complete point.
+        # Called only under the native admission -> token -> history locks.
+        info = state.tokens.get(token)
+        if info is None:
+            raise HTTPException(403, "Unknown registration")
+        run, task = info["run_id"], info["task_id"]
+        if state._closed or state.tasks[task].game_mode:
+            raise HTTPException(409, "Checkpoint requires an open non-game registration")
+        if state._outstanding_by_run.get((run, task), 0):
+            raise HTTPException(409, "Accepted evaluation is still outstanding")
+        if info["next_auto"] != 1:
+            raise HTTPException(409, "This checkpoint profile requires no automatic evaluations")
+        history = state.run_history.get(state._run_key(run, task), [])
+        rounds = [entry.get("round") for entry in history]
+        expected = {f"agent-{index}" for index in range(1, info["next_agent"])}
+        if (len(rounds) != len(expected) or set(rounds) != expected or
+                any(entry.get("status") not in {SubmissionStatus.COMPLETED, SubmissionStatus.ERROR} for entry in history)):
+            raise HTTPException(409, "Consumed rounds and released history are not reconciled")
+        results = {}
+        for entry in history:
+            identifier = entry.get("submission_id")
+            native = state.submissions.get(identifier)
+            if (native is None or native.get("run_id") != run or native.get("task_id") != task or
+                    native.get("round") != entry["round"] or native.get("status") != entry["status"]):
+                raise HTTPException(409, "Released result details are unavailable or differ")
+            results[identifier] = {key: native.get(key) for key in ("status", "report", "error")}
+        # Reuse native ranking, including validity, direction and tie rules.
+        # The working candidate can differ from this older selected archive.
+        best = select_best(history, state.tasks[task].judge.score_direction,
+                           state.tasks[task].judge.selection)
+        incumbent = None
+        raw = None
+        if best["best_round"]:
+            selected = next(entry for entry in history if entry["round"] == best["best_round"])
+            try:
+                raw = _read_archive(config.log_dir, run, task, selected["round"])
+            except (OSError, ValueError) as error:
+                raise HTTPException(409, "Native selected incumbent archive is unavailable or moved") from error
+            incumbent = {"round": selected["round"], "submission_id": selected["submission_id"],
+                         "archive_sha256": hashlib.sha256(raw).hexdigest(), "archive_bytes": len(raw)}
+        observed = time.time()
+        last = info["last_agent_submit_at"]
+        cooldown = info["submission_cooldown"]
+        remaining = max(0, last + (cooldown or 0) - observed) if last else 0
+        packet = {
+            "schema_version": SCHEMA, "judge_epoch": epoch,
+            "run_id": run, "task_id": task, "evaluator": tasks[task],
+            "observed_at_epoch": observed, "last_agent_submit_at_epoch": last,
+            "submission_cooldown_seconds": cooldown,
+            "remaining_cooldown_seconds": remaining,
+            "max_agent_submissions": info["max_agent_submissions"],
+            "next_agent_round": info["next_agent"], "next_auto_round": info["next_auto"],
+            "released_history": history, "released_results": results, "outstanding_evaluations": 0,
+            "native_incumbent": incumbent,
+            "execution_authority_granted": False,
+        }
+        # Bind both reads to the same released state, not a wall clock. A new
+        # accepted round, source epoch or changed archive invalidates download.
+        binding = {key: value for key, value in packet.items()
+                   if key not in {"observed_at_epoch", "remaining_cooldown_seconds"}}
+        packet["capture_binding_sha256"] = hashlib.sha256(
+            json.dumps(binding, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+        return json.loads(json.dumps(packet)), raw  # No live references or tokens leave the lock.
+
+    def capture(request):
+        with state._admission_lock, state._tokens_lock, state._history_lock:
+            return snapshot_locked(request.token)[0]
+
+    def archive(request):
+        with state._admission_lock, state._tokens_lock, state._history_lock:
+            packet, raw = snapshot_locked(request.token)
+            incumbent = packet["native_incumbent"]
+            if packet["capture_binding_sha256"] != request.capture_binding_sha256 or incumbent is None:
+                raise HTTPException(409, "Native checkpoint basis moved or has no selected incumbent")
+            return Response(raw, media_type="application/octet-stream")
+    capture.__annotations__["request"] = ReadRequest
+    archive.__annotations__["request"] = ArchiveRequest
+    app.post(ROUTE)(capture)
+    app.post(ARCHIVE_ROUTE)(archive)
+    return app
+
+
+def main(argv=None):
+    import uvicorn
+    from sforge.harness.config import load_config
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--checkpoint-capture", action="store_true")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=8080)
+    parser.add_argument("--tasks-dir", type=Path)
+    parser.add_argument("--log-dir", type=Path)
+    args = parser.parse_args(argv)
+    config = load_config()
+    overrides = {key: getattr(args, key) for key in ("tasks_dir", "log_dir") if getattr(args, key) is not None}
+    uvicorn.run(create_app(replace(config, **overrides), checkpoint_capture=args.checkpoint_capture),
+                host=args.host, port=args.port, access_log=False)
+
+
+if __name__ == "__main__":
+    main()