fix(checkpoint): qualify restored decisions and non-root capture

benchmark/README.md

@@ -348,22 +348,24 @@ python -m benchmark.edgebench.checkpoint_judge --checkpoint-capture \
   --tasks-dir /path/to/tasks --log-dir /path/to/private-run-logs
 ```
 
-This adapter's supported SForge source is the public compatibility fork at
-[`0e20fb499ee248f012d6467c00b3dec00a29fe85`](https://github.com/huangruiteng/EdgeBench/commit/0e20fb499ee248f012d6467c00b3dec00a29fe85).
-It exposes the existing native judge/admission owner; the upstream `sforge==1.1.0`
-distribution does not provide that seam. Enabling capture against an unsupported
-provider fails before adding routes. This pin qualifies the capture prerequisite,
-not an upstream release or restored admission. From this LoopX checkout, prepare
-an isolated Python 3.13 environment and verify the actual imported source:
+The supported public checkpoint provider is the compatibility fork at
+[`071a884012754127f28f5b84be10f533a42db527`](https://github.com/loopx-agent/EdgeBench/commit/071a884012754127f28f5b84be10f533a42db527).
+It contains `sforge.harness.checkpoint` and the native
+`create_app(..., checkpoints=True)` seam used by the controller below. The earlier
+`0e20fb499ee248f012d6467c00b3dec00a29fe85` compatibility pin and upstream
+`sforge==1.1.0` do not contain that API and cannot qualify joint continuation.
+Enabling capture against an unsupported provider fails before adding routes.
+This is pinned proposed source, not an upstream release. From this LoopX checkout,
+prepare an isolated Python 3.12 environment and verify the imported API:
 
 ```sh
 CKPT_PROVIDER_ROOT=$(mktemp -d)
-git clone https://github.com/huangruiteng/EdgeBench.git "$CKPT_PROVIDER_ROOT/provider"
-git -C "$CKPT_PROVIDER_ROOT/provider" checkout --detach 0e20fb499ee248f012d6467c00b3dec00a29fe85
-uv venv --python 3.13 "$CKPT_PROVIDER_ROOT/venv"
+git clone https://github.com/loopx-agent/EdgeBench.git "$CKPT_PROVIDER_ROOT/provider"
+git -C "$CKPT_PROVIDER_ROOT/provider" checkout --detach 071a884012754127f28f5b84be10f533a42db527
+uv venv --python 3.12 "$CKPT_PROVIDER_ROOT/venv"
 uv pip install --python "$CKPT_PROVIDER_ROOT/venv/bin/python" \
   -e '.[test]' -e "$CKPT_PROVIDER_ROOT/provider" 'httpx>=0.28,<1'
-"$CKPT_PROVIDER_ROOT/venv/bin/python" -c 'import sforge,loopx; print(sforge.__file__); print(loopx.__file__)'
+"$CKPT_PROVIDER_ROOT/venv/bin/python" -c 'import inspect,sforge,loopx,sforge.harness.checkpoint; from sforge.harness.judge_server import create_app; assert "checkpoints" in inspect.signature(create_app).parameters; print(sforge.__file__); print(loopx.__file__)'
 "$CKPT_PROVIDER_ROOT/venv/bin/python" -m pytest -q benchmark/tests/test_checkpoint_judge.py
 # Linux with an owned disposable Docker daemon; grades only synthetic fixtures.
 LOOPX_EDGEBENCH_DOCKER_SMOKE=1 "$CKPT_PROVIDER_ROOT/venv/bin/python" -m pytest -q \
@@ -517,10 +519,10 @@ continued research.
 ### Controller-only native capsule transport
 
 The read-only participant above does **not** implement the newer native capsule
-API. The separate controller adapter targets
-[EdgeBench PR #23](https://github.com/ByteDance-Seed/EdgeBench/pull/23), pinned at
-`57dce5dd1fded9d5d5cbd44fa59c59d27a068d62`; this is proposed source, not a released
-upstream version. Start that pinned native judge with `sforge serve --checkpoints`.
+API. The separate controller adapter targets the API proposed in
+[EdgeBench PR #23](https://github.com/ByteDance-Seed/EdgeBench/pull/23), using the
+public compatibility pin `071a884012754127f28f5b84be10f533a42db527` above.
+Start that pinned native judge with `sforge serve --checkpoints`.
 Its private controller configuration file has exactly `url`, `admin_secret` and
 `bank_root` (the native log directory's `checkpoints` directory). Keep this file,
 the bank and controller process outside every solver's filesystem/network authority.
@@ -636,6 +638,70 @@ Omit the controller file and `controller_handoff` on a new run to disable automa
 controller capture; removing the checkpoint request disables capture entirely.
 Existing artifacts remain inert, and the source trial is not restarted by disabling it.
 
+To reproduce **joint source capture and both fresh-guest receivers**, use
+[`qualify_checkpoint_joint.py`](tests/qualify_checkpoint_joint.py) from the current
+clean dedicated LoopX checkout; keep it frozen for the whole command. Run the
+controller on Linux as the ordinary owner of an explicitly
+owned disposable Docker daemon. Prepare the provider/venv above, install this
+checkout's Node dependencies with `npm ci`, and put a compatible **Host** Node on
+`PATH`. Separately supply Linux **amd64** binaries for `codex-cli 0.160.0` and Node
+`24.21.0`, and a caller-prepared `python:3.12-slim` image. These inputs are explicit;
+the qualification does not download a native CLI, pull an image or use model
+credentials. The receiver also imports `harbor==0.20.0` to exercise the actual
+phase-command builder; it does not launch a Harbor job. Build Python 3.12 guest
+dependencies from the pinned provider plus this qualification prerequisite:
+
+```sh
+set -eu
+"$CKPT_PROVIDER_ROOT/venv/bin/python" -c 'import sys,tomllib; from pathlib import Path; d=tomllib.loads((Path(sys.argv[1])/"pyproject.toml").read_text()); print("\n".join(d["project"]["dependencies"]+d["dependency-groups"]["dev"]+["harbor==0.20.0"]))' \
+  "$CKPT_PROVIDER_ROOT/provider" > "$CKPT_PROVIDER_ROOT/guest-requirements.txt"
+mkdir "$CKPT_PROVIDER_ROOT/guest-deps"
+docker image inspect python:3.12-slim >/dev/null
+docker run --rm --platform linux/amd64 --user "$(id -u):$(id -g)" \
+  -v "$CKPT_PROVIDER_ROOT:/qualification" python:3.12-slim \
+  python -m pip install --target /qualification/guest-deps \
+  -r /qualification/guest-requirements.txt
+# Set these to the explicitly prepared Linux amd64 binaries.
+CKPT_GUEST_CODEX=/absolute/path/to/codex
+CKPT_GUEST_NODE=/absolute/path/to/node
+"$CKPT_PROVIDER_ROOT/venv/bin/python" -m benchmark.tests.qualify_checkpoint_joint \
+  --provider-source "$CKPT_PROVIDER_ROOT/provider" \
+  --native-binary "$CKPT_GUEST_CODEX" --guest-node "$CKPT_GUEST_NODE" \
+  --guest-dependencies "$CKPT_PROVIDER_ROOT/guest-deps" \
+  --output "$CKPT_PROVIDER_ROOT/joint-result"
+```
+
+The public fixture runs six actual native work wakes, captures the pending
+pre-decision guard and continues the source; the ordinary and stride receivers
+then use separate fresh guests with no mounted source authority. It includes a
+whole declared workspace directory and a read-only Git member. Receivers run as
+the ordinary non-root owner, select the workspace by its sealed source path and
+materialize it through the typed artifact owner. Copied source registry/state
+files remain historical: current authority uses a fresh external registry/runtime,
+and the old Goal cannot be admitted through it. Only the model
+response stream and grading/backend fixtures are synthetic; the native CLI,
+canonical SQLite authority, worker, supervisor and capsule transport are real.
+Check `result.json`, `source.log`, `ordinary.log` and `stride.log` in the private
+output directory; failure is an error, including missing prerequisites, and a
+skipped test cannot produce a passed receipt. A second attempt needs a new output
+directory. Keep its controller config, native
+stores, reports and logs private. Cleanup removes only containers owned by this
+invocation. This reproduces mechanical continuation; the receipt remains
+`research_point_qualified=false`. A real later research point, scientific policy
+benefit and the packaged frontend restore journey still need separate acceptance.
+
+An enabled checkpoint rendezvous returns an explicit failed quota readback to
+its original caller without interrupting the writer or attempting capture.
+The original failure and recovery requirements remain unchanged; delivery of
+that error grants no work or settlement authority. Successful, foreign and
+malformed readbacks retain the existing typed capture checks. This contains
+optional capture failure; it does not repair the underlying quota error.
+
+The qualifier exports only the source's digest-bound native configuration through
+the existing Docker archive transport into a controller-owned private directory.
+It works with a root source guest and ordinary controller/receivers, preserves
+source permissions, and never exports the source authentication files.
+
 An explicit **SForge phase-clock participant** preserves a declared Host file's
 remaining duration alongside native opportunities. Add
 `--phase-deadline-file /private/source-phase-deadline` to capture, and
@@ -706,7 +772,7 @@ runtime receipt 绑定摘要；programmatic adapter 也可指定已放入 guest
 已放入 guest 的声明文件丢失、权限失败或损坏只记录捕获失败，原工作仍按
 完整的原始 quota guard 继续；准入 owner 不可用仍停止。首次 replan 捕获
 失败后，修复声明不得把后来的状态补记成同一原点。评测参与者使用上文明确的
-公开 SForge fork/pin 和 Python 3.13；普通 upstream 安装缺少该 seam 时明确
+公开 SForge fork/pin 和 Python 3.12；普通 upstream 安装缺少该 seam 时明确
 拒绝。Codex 原生会话的真实分页/compaction 与隔离 fork 须单独验证。
 评测机会及 incumbent 参与者也分配未占用的 ID，并在封存的
 `point_metadata.evaluation_opportunity` 中记录实际映射；无 incumbent 时对应值为 null。
@@ -733,8 +799,15 @@ worker 保留原 guard，controller 暂停全部容器写入者，核对仍有
 文件 witness，再封存原生评测机会、incumbent 及剩余预算。源变化、过期或失去
 交接会拒绝；可恢复的捕获失败不取消原研究，也不得改用稍后的状态。解除暂停后，
 源会话继续使用原 packet 和 deadline。
-真实 native worker、Docker 与 native HTTP 的合成验证覆盖了联合捕获和源接续；
-产物仍是 `research_point_qualified=false`。独立目标环境的新权限准入、有效工作、
-联合丢回复恢复及后续评测尚待一体验证，不能据此重启研究实验。新运行省略
+上文的公开复现命令使用固定 provider `071a884`、Python 3.12、显式 Linux amd64
+Codex/Node 及调用者准备的 Docker 镜像。非 root controller 从六个真实工作 wake
+形成模型决策前捕获，保留完整声明的目录及只读 Git 对象；普通与 stride receiver
+以非 root 身份在两个全新 guest 按封存源路径定位任务目录，由类型化 owner 物化；
+原注册表/状态只是历史，新的外置 registry/runtime 不授予原 Goal 权限。
+分别验证新权限准入、完整历史、剩余预算/冷却及实际接续。模型响应和评分为合成
+fixture，原生 CLI、worker、supervisor、SQLite 和 capsule transport 使用真实路径。
+结果及三份日志只存私有输出，失败或缺少前置条件明确报错，重试须使用新目录；
+产物仍是 `research_point_qualified=false`。真实中后期研究点、策略净收益及安装后的
+frontend restore 旅程仍须另行验收。新运行省略
 controller 文件及 handoff 开关即可关闭；移除 checkpoint 请求关闭全部捕获，
 已有产物保持惰性。

benchmark/runtime/checkpoint.py

@@ -295,6 +295,13 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
             "action": "select_replan", "quota": guard, "goal_id": env["LOOPX_GOAL_ID"],
             "agent_id": env["LOOPX_AGENT_ID"], "turn_instance_id": receipt["turn_id"],
         })
+        from .checkpoint_continuation import restored_capture_gate
+        restore_gate = restored_capture_gate(env, intent.get("obligation_id"))
+        receipt["checkpoint_restore"] = {k: v for k, v in restore_gate.items() if k != "instructions"}
+        preloaded_body = restore_gate.get("instructions", "") + preloaded_body
+        if not restore_gate["capture_allowed"]:
+            receipt["checkpoint"]["reason"] = restore_gate["reason"]
+            return preloaded_body
         root = Path(env["LOOPX_WAKE_LOG_DIR"]).parent / "checkpoints"
         if declaration_error is not None:
             if intent["capture"]:
@@ -341,10 +348,11 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                     "point_metadata": {**request["point_metadata"], "replan": intent,
                                        "capture_order": "before_model_replan_decision"}}
                 effective_request.pop("trigger")
-                capture_after_wake(env, prior, supplied_request=(effective_request, raw))
+                capture_after_wake(env, prior, supplied_request=(effective_request, raw), pending_guard=guard_path)
                 receipt["checkpoint"].update(prior["checkpoint"], phase="before_replan")
                 receipt["checkpoint"]["source_turn_id"] = prior["turn_id"]
     except Exception as error:
+        receipt["checkpoint_restore"] = {"capture_allowed": False, "reason": "capture_basis_unavailable"}
         receipt["checkpoint"].update(reason="capture_failed", error_kind=type(error).__name__, error=str(error)[:500])
     finally:
         receipt["checkpoint"]["elapsed_seconds"] = time.monotonic() - started
@@ -354,7 +362,8 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
 
 
 def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
-                       supplied_request: tuple[dict[str, Any], bytes] | None = None) -> None:
+                       supplied_request: tuple[dict[str, Any], bytes] | None = None,
+                       pending_guard: Path | None = None) -> None:
     request_path = env.get("LOOPX_CHECKPOINT_REQUEST")
     if not request_path:
         return  # Feature-off path performs no additional read or write.
@@ -381,6 +390,12 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             request, raw = read_checkpoint_request(request_path)
         else:
             request, raw = supplied_request
+        if supplied_request is None:
+            from .checkpoint_continuation import restored_capture_gate
+            restore_gate = restored_capture_gate(env)
+            if not restore_gate["capture_allowed"]:
+                receipt["checkpoint"]["reason"] = restore_gate["reason"]
+                return
         if request.get("trigger") == "fresh_replan":
             receipt["checkpoint"]["reason"] = "pre_replan_hook_required"
             return
@@ -414,6 +429,12 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             "request_sha256": hashlib.sha256(raw).hexdigest(),
         }
         components = request["components"]
+        if pending_guard is not None:
+            decision_id = _participant_id(components, "held_decision")
+            components = [*components, {"id": decision_id, "source_path": str(pending_guard)}]
+            metadata.update(held_decision_component_id=decision_id,
+                decision_sha256=hashlib.sha256(pending_guard.read_bytes()).hexdigest(),
+                replan=request["point_metadata"]["replan"], capture_order="before_model_replan_decision")
         if request.get("loopx_state"):
             evidence = Path(env["LOOPX_WAKE_LOG_DIR"]) / receipt["turn_id"] / "checkpoint-research-context.json"
             components, research = research_participants(env, components, metadata, destination, evidence, deadline)
@@ -555,19 +576,27 @@ def remaining():
         if hashlib.sha256(raw).hexdigest() != message["sha256"]:
             raise ValueError("decision handoff digest mismatch")
         quota = json.loads(raw)
+        # This is transport dispatch, not capture admission: an explicit quota
+        # failure still belongs to its original caller's recovery path. Do not
+        # signal that writer or submit its failure to the successful-readback
+        # selector. Successful and foreign/malformed packets retain typed checks.
+        failed_readback = isinstance(quota, dict) and quota.get("ok") is False and quota.get("goal_id") == goal_id
         interrupted = None
-        if native_process is not None and interrupt_when is None:
+        if not failed_readback and native_process is not None and interrupt_when is None:
             from .worker import interrupt_checkpoint_process
 
             # Interrupt before the typed selector: even this read can outlive a
             # native tool yield. Keep original bytes for same-identity recovery.
             interrupted = interrupt_checkpoint_process(native_process, timeout=min(interrupt_timeout, remaining()))
-        intent = effect_runtime_result("runtime.checkpoint_artifacts", {
-            "action": "select_replan", "quota": quota, "goal_id": goal_id,
-            "agent_id": agent_id, "turn_instance_id": (turn_instance_id if turn_instance_id is not None
-                else (quota.get("heartbeat_receipt") or {}).get("turn_instance_id")),
-        }, timeout=remaining())
-        if native_process is not None and interrupt_when is not None and interrupt_when(intent):
+        if failed_readback:
+            intent = {"capture": False, "reason": "quota_readback_failed", "execution_authority_granted": False}
+        else:
+            intent = effect_runtime_result("runtime.checkpoint_artifacts", {
+                "action": "select_replan", "quota": quota, "goal_id": goal_id,
+                "agent_id": agent_id, "turn_instance_id": (turn_instance_id if turn_instance_id is not None
+                    else (quota.get("heartbeat_receipt") or {}).get("turn_instance_id")),
+            }, timeout=remaining())
+        if not failed_readback and native_process is not None and interrupt_when is not None and interrupt_when(intent):
             from .worker import interrupt_checkpoint_process
 
             interrupted = interrupt_checkpoint_process(native_process, timeout=min(interrupt_timeout, remaining()))