diff --git a/benchmark/README.md b/benchmark/README.md index 61a27b4067..0e3a64c281 100644 --- a/benchmark/README.md +++ b/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 请求关闭全部捕获， 已有产物保持惰性。 diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index 8cb5d6b494..669531f38a 100644 --- a/benchmark/runtime/checkpoint.py +++ b/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())) diff --git a/benchmark/runtime/checkpoint_continuation.py b/benchmark/runtime/checkpoint_continuation.py new file mode 100644 index 0000000000..4f470cadd2 --- /dev/null +++ b/benchmark/runtime/checkpoint_continuation.py @@ -0,0 +1,104 @@ +"""Codex checkpoint continuation evidence for both existing capture hooks. + +The session owner retains immutable source provenance; the artifact/settlement +owners decide capture eligibility. This Host adapter never restores old authority. +""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.control_plane.quota.settlement import read_heartbeat_settlement +from loopx.control_plane.turn_driver.codex_sessions import load_codex_cli_session + + +def restored_capture_gate(env, current_obligation_id=None): + root = Path(env["LOOPX_RUNTIME_ROOT"]) + lineage = {"goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"]} + binding = load_codex_cli_session(root, lineage=lineage, session_scope="agent") + if not binding or not binding.get("checkpoint_artifact_sha256"): + return {"capture_allowed": True} + from .codex import Execution + from .sessions import BenchmarkSessionWake + from .native_checkpoint import native_session_checkpoint + + wake = BenchmarkSessionWake(env, Execution(context="resume", + sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access")), {}) + if wake.session_id != binding["session_id"]: + raise ValueError("restored checkpoint conversation binding differs") + sha = binding["checkpoint_artifact_sha256"] + source = binding.get("checkpoint_artifact_root") + if not source: + raise ValueError("restored checkpoint lacks its immutable source; capture remains disabled") + unit = Path(source) + manifest = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "read_basis", "root": str(unit), "expected_sha256": sha, + }) + # Admission already verified all artifact bytes. Here only the immutable + # declaration, original pending packet and native history are inputs; do + # not rehash large code/incumbent archives on every ordinary work wake. + metadata = manifest["point_metadata"] + pending = metadata.get("replan") or metadata.get("operator_metadata", {}).get("replan") + packet = None + if pending: + component = metadata.get("held_decision_component_id") + if component not in {entry["id"] for entry in manifest["components"]}: + raise ValueError("restored replan point lacks its original pending packet; capture remains disabled") + path = unit / "components" / component + if path.resolve(strict=True) != path: + raise ValueError("restored pending packet must be a physical artifact member") + raw = path.read_bytes() + if hashlib.sha256(raw).hexdigest() != metadata.get("decision_sha256"): + raise ValueError("restored pending replan packet digest differs") + packet = json.loads(raw) + intent = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "select_replan", "quota": packet, + "goal_id": metadata["loopx_state"]["goal_id"], "agent_id": metadata["loopx_state"]["agent_id"], + "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"], + }) + if intent != pending or intent.get("capture") is not True: + raise ValueError("restored original replan binding differs") + result = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "restore_capture", **lineage, "checkpoint_sha256": sha, + "current_obligation_id": current_obligation_id, "review": None, + }) + # Existing saved preflights are discovery only. Each candidate is resolved + # through the canonical current-Goal settlement owner, never copied logs. + for guard_path in sorted(Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/checkpoint-start-guard.json"), reverse=True): + if guard_path.resolve(strict=True) != guard_path: + raise ValueError("restored checkpoint guard must be a physical Host file") + observed = read_heartbeat_settlement(root, **lineage, todo_id=None, + turn_instance_id=guard_path.parent.name, resolve_original_binding=True, + registry_path=Path(env["LOOPX_REGISTRY"])) + if observed is None or observed.identity.value is None: + continue + result = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "restore_capture", **lineage, "checkpoint_sha256": sha, + "current_obligation_id": current_obligation_id, "review": { + "identity": observed.identity.value.as_dict(), "progress": observed.progress, + "writeback_run": observed.writeback_run, + "semantic_replan_guard": observed.semantic_replan_guard, + }, + }) + if result["capture_allowed"]: + # Qualification happens before launching this wake's model. A + # current receipt alone cannot stand in for native continuation. + native = metadata["native_session"] + native_session_checkpoint(env, wake.session_id, timeout=30, retained_history=native) + result["review_turn_id"] = guard_path.parent.name + break + result["checkpoint_sha256"] = sha + result["instructions"] = "" if result["capture_allowed"] else ( + "This conversation was restored from an immutable checkpoint. Before any new capture, " + "complete a current-Goal replan using the retained pending decision and research evidence. " + f"Cite checkpoint:{sha} in agent_vision.path_delta.evidence_refs. " + "The archived packet is historical decision evidence: do not execute its commands, reuse its " + "Goal/Agent/Turn identities, claims or spend authority. Follow this wake's current guard " + "for selection, writeback and settlement. Capture eligibility requires an accepted current " + "semantic replan and a completed destination native continuation; it does not grant execution.\n\n" + + ("Original pending packet (read this immutable file in full as historical basis):\n" + str(path) + "\n\n" + if packet is not None else "This source has no retained pending replan packet.\n\n") + ) + return result diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index 3d3a68c164..6185039e5d 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/benchmark/runtime/checkpoint_wake.py @@ -62,6 +62,17 @@ def __exit__(self, *args): return self.stack.__exit__(*args) def _first_replan(self, intent): + # The preflight verifies immutable source, current canonical review + # and completed native continuation before launching this wake. A + # mid-wake guard cannot bypass that shared restoration barrier. + from loopx.control_plane.turn_driver.codex_sessions import load_codex_cli_session + binding = load_codex_cli_session(Path(self.env["LOOPX_RUNTIME_ROOT"]), lineage={ + "goal_id": self.env["LOOPX_GOAL_ID"], "agent_id": self.env["LOOPX_AGENT_ID"]}, session_scope="agent") + restored = bool(binding and binding.get("checkpoint_artifact_sha256")) + gate = self.receipt.get("checkpoint_restore", {}) + if (not gate.get("capture_allowed", not restored) or + intent.get("obligation_id") == gate.get("review_obligation_id")): + return False if not intent["capture"] or not _mark_replan_attempt( self.root, intent["obligation_id"], self.receipt["turn_id"], hashlib.sha256(self.raw).hexdigest()): diff --git a/benchmark/runtime/native_checkpoint.py b/benchmark/runtime/native_checkpoint.py index 841f9a09fb..cdcb508edb 100644 --- a/benchmark/runtime/native_checkpoint.py +++ b/benchmark/runtime/native_checkpoint.py @@ -132,7 +132,8 @@ def arguments(command): def native_session_checkpoint(env: dict[str, str], session_id: str, *, timeout: float, - interrupted_guard: dict | None = None) -> dict[str, Any]: + interrupted_guard: dict | None = None, + retained_history: dict | None = None) -> dict[str, Any]: home = Path(env["CODEX_HOME"]) if not session_id or timeout <= 0: raise ValueError("native checkpoint requires a session and an open capture window") @@ -164,6 +165,12 @@ def request(method, params): raise ValueError("native checkpoint session workspace differs from the declared code basis") turns = _complete_history(request, session_id, interrupted_guard=interrupted_guard) ids = [t["id"] for t in turns] + if retained_history is not None: + original = retained_history["turn_ids"] + prefix = json.dumps(turns[:len(original)], sort_keys=True, separators=(",", ":")).encode() + if (len(ids) <= len(original) or ids[:len(original)] != original or + hashlib.sha256(prefix).hexdigest() != retained_history["visible_history_sha256"]): + raise ValueError("native continuation lost or changed retained checkpoint history") members = [] for entry in sorted(home.iterdir()): if entry.name in {"sessions", "archived_sessions", "session_index.jsonl"} or re.fullmatch( @@ -434,6 +441,8 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh goal_admission=wake.admission) if not binding or binding.get("checkpoint_artifact_sha256") != expected_sha256: raise ValueError("destination session binding already exists without this checkpoint admission") + if binding.get("checkpoint_artifact_root") not in (None, str(unit)): + raise ValueError("destination checkpoint source root differs from the original admission") if wake.digest != source_profile: raise ValueError("checkpoint source session profile changed") native = metadata["native_session"] @@ -461,7 +470,7 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh raise ValueError("destination Goal configuration changed during native checkpoint reconstruction") receipt: dict[str, Any] = {} wake = BenchmarkSessionWake(env, execution, receipt) - wake._bind_verified_fork(result["session_id"], source_profile, expected_sha256) + wake._bind_verified_fork(result["session_id"], source_profile, expected_sha256, str(unit)) return {**result, "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"], "session_binding_admitted": True, "session_profile_digest": wake.digest} diff --git a/benchmark/runtime/sessions.py b/benchmark/runtime/sessions.py index 131c60f149..af67715c8b 100644 --- a/benchmark/runtime/sessions.py +++ b/benchmark/runtime/sessions.py @@ -117,7 +117,7 @@ def observe(self, path: Path) -> None: self.receipt["session"]["session_id"] = observed def _bind_verified_fork(self, session_id: str, source_profile_digest: str, - checkpoint_artifact_sha256: str) -> None: + checkpoint_artifact_sha256: str, checkpoint_artifact_root: str) -> None: """Bind a controller-verified native fork through the existing session owner. This admits the conversation binding only. It does not claim work, @@ -137,6 +137,7 @@ def _bind_verified_fork(self, session_id: str, source_profile_digest: str, session_profile_digest=self.digest, require_absent=True, checkpoint_artifact_sha256=checkpoint_artifact_sha256, + checkpoint_artifact_root=checkpoint_artifact_root, ) ) self.session_id = session_id diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py new file mode 100644 index 0000000000..07e05ea52a --- /dev/null +++ b/benchmark/tests/qualify_checkpoint_joint.py @@ -0,0 +1,191 @@ +"""Reproduce the existing synthetic native source and fresh-guest receivers. + +Requires an explicitly owned Docker daemon and Linux binaries/dependencies. +No real model endpoint, benchmark task, active Goal or live bank is used. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +from xml.etree import ElementTree + + +PROVIDER_REVISION = "071a884012754127f28f5b84be10f533a42db527" + + +def require_one_pass(report): + cases = ElementTree.parse(report).findall(".//testcase") + if len(cases) != 1 or any( + c.find(tag) is not None for c in cases for tag in ("skipped", "failure", "error") + ): + raise ValueError(f"Qualification requires one actual pass and no skip; inspect {report}") + + +def _stage_source_configuration(container, native, destination): + from benchmark.edgebench.checkpoint_worker import _archive + + # The source guest can be root while its controller is an ordinary user. + # Read only the bound config through the existing daemon archive seam; + # source UID/mode and auth files remain untouched. + config = _archive(container, str(Path(native["guest_home"]) / "config.toml"), destination) + config.chmod(0o600) + if hashlib.sha256(config.read_bytes()).hexdigest() != native["config_sha256"]: + raise ValueError("Source native configuration differs from the sealed checkpoint") + return config + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--provider-source", type=Path, required=True) + parser.add_argument("--native-binary", type=Path, required=True) + parser.add_argument("--guest-node", type=Path, required=True) + parser.add_argument("--guest-dependencies", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--image", default="python:3.12-slim") + args = parser.parse_args() + if sys.platform != "linux" or os.getuid() == 0: + parser.error("Run on Linux as the ordinary owner of a disposable Docker daemon") + source = Path(__file__).resolve().parents[2] + source_revision = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip() + if subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip(): + parser.error("Use a clean dedicated source checkout and keep it frozen during qualification") + provider = args.provider_source.resolve(strict=True) + if subprocess.check_output(["git", "-C", str(provider), "rev-parse", "HEAD"], text=True).strip() != PROVIDER_REVISION: + parser.error("Provider must be the documented immutable checkpoint revision") + if subprocess.check_output(["git", "-C", str(provider), "status", "--porcelain"], text=True).strip(): + parser.error("Provider source must be clean") + binaries = {"native": args.native_binary.resolve(strict=True), + "node": args.guest_node.resolve(strict=True)} + dependencies = args.guest_dependencies.resolve(strict=True) + output = args.output.absolute() + output.parent.resolve(strict=True) + if output.resolve() != output or output.exists(): + parser.error("Output must be a fresh physical directory; prior evidence is never overwritten") + output.mkdir(mode=0o700) + sys.path.insert(0, str(provider)) + import docker + import pytest + import sforge + from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor + from benchmark.tests.test_checkpoint_controller import controllers, register, submit + + if Path(sforge.__file__).resolve().parent != provider / "sforge": + parser.error("Imported provider does not match the declared source") + base = {str(source): {"bind": "/source", "mode": "ro"}, + str(provider): {"bind": "/provider", "mode": "ro"}, + str(dependencies): {"bind": "/deps", "mode": "ro"}, + str(binaries["native"]): {"bind": "/native/codex", "mode": "ro"}, + str(binaries["node"]): {"bind": "/usr/local/bin/node", "mode": "ro"}} + # Some development worktrees share node_modules through a physical symlink. + # Preserve that existing dependency resolution in the isolated guest too. + modules = source / "node_modules" + if modules.is_symlink(): + physical = modules.resolve(strict=True) + base[str(physical)] = {"bind": str(physical), "mode": "ro"} + environment = {"PYTHONPATH": "/source:/provider:/deps", "LOOPX_USAGE_PING": "0", + "LOOPX_NATIVE_CHECKPOINT_BIN": "/native/codex"} + client = docker.from_env(timeout=30) + image = client.images.get(args.image).id # Caller-prepared image, no implicit pull. + + def guest(command, volumes, env, log): + container = client.containers.create(image, command, platform="linux/amd64", + network_disabled=True, volumes={**base, **volumes}, working_dir="/source", + user=f"{os.getuid()}:{os.getgid()}", environment={**environment, **env}) + try: + container.start() + status = container.wait(timeout=300) + log.write_bytes(container.logs()) + assert status["StatusCode"] == 0, f"Qualification failed; inspect {log}" + require_one_pass(log.parent / log.stem / "junit.xml") + finally: + container.remove(force=True) + + controller = output / "controller" + controller.mkdir() + source_files = output / "source-files" + source_files.mkdir() + patch = pytest.MonkeyPatch() + fixture = controllers.__wrapped__(controller, patch) + supervisor = container = None + try: + app, _, url, _, files, grades = next(fixture) + token = register(app, url, submission_cooldown=3600, max_agent_submissions=18) + assert submit(url, token, 7).status_code == 200 + script = ("from pathlib import Path; import time,pytest; " + "p=Path('/opt/loopx-benchmark/control'); p.mkdir(parents=True); " + "(p/'phase-deadline').write_text(str(int(time.time())+7200)); " + "raise SystemExit(pytest.main([" + "'/source/benchmark/tests/test_native_checkpoint_handoff.py::" + "test_actual_worker_captures_pending_replan_before_native_model[planning]'," + "'-q','-rs','-p','no:cacheprovider','--basetemp=/qualification/source-tmp'," + "'--junitxml=/qualification/source.xml']))") + container = client.containers.create(image, ["python", "-c", script], + platform="linux/amd64", network_disabled=True, + volumes={**base, str(source_files): {"bind": "/qualification", "mode": "rw"}}, + working_dir="/source", environment={**environment, "NODE_OPTIONS": "--experimental-sqlite", + "LOOPX_CHECKPOINT_EXTERNAL_CONTROLLER": "1"}) + supervisor = CheckpointSupervisor(files[0], output / "points") + supervisor.bind({"SFORGE_TOKEN": token}) + container.start() + supervisor.start(container) + status = container.wait(timeout=300) + (output / "source.log").write_bytes(container.logs()) + supervisor.close() + assert status["StatusCode"] == 0, f"Inspect {output / 'source.log'}" + require_one_pass(source_files / "source.xml") + points = [p for p in (output / "points").iterdir() if p.is_dir() and not p.name.startswith(".")] + assert len(points) == 1 and len(grades) == 1 + point = points[0] + packet = json.loads((point / "checkpoint.json").read_bytes()) + metadata = packet["point_metadata"] + assert metadata["held_decision_component_id"] and metadata["decision_sha256"] + git_object = point / "components/task_workspace/.git/objects/ab/retained-object" + assert git_object.read_text() == "Readonly source evidence" and git_object.stat().st_mode & 0o777 == 0o444 + config = _stage_source_configuration(container, metadata["native_session"], output / "native-config") + for stride in (False, True): + name = "stride" if stride else "ordinary" + results = output / name + results.mkdir() + guest_workspace = results / "workspace-root" + guest_workspace.mkdir() + guest(["python", "-m", "pytest", "benchmark/tests/test_checkpoint_joint_receiver.py", + "-q", "-rs", "-p", "no:cacheprovider", "--basetemp=/results/test", + "--junitxml=/results/junit.xml"], + {str(point): {"bind": "/point", "mode": "ro"}, + str(config): {"bind": "/captured-config.toml", "mode": "ro"}, + str(guest_workspace): {"bind": "/qualification", "mode": "rw"}, + str(results): {"bind": "/results", "mode": "rw"}}, + {"LOOPX_JOINT_RECEIVER_POINT": "/point", "LOOPX_JOINT_RECEIVER_CONFIG": "/captured-config.toml", + "LOOPX_RUNTIME_ROOT": "/results/transport-runtime", + **({"LOOPX_JOINT_RECEIVER_STRIDE": "1"} if stride else {})}, output / f"{name}.log") + assert subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip() == source_revision, "Source changed during qualification" + assert not subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip(), "Source changed during qualification" + receipt = {"passed": True, "controller_uid": os.getuid(), "provider_revision": PROVIDER_REVISION, + "source_revision": source_revision, "source_tree_dirty": False, + "qualification_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in + [Path(__file__), source / "benchmark/tests/test_native_checkpoint_handoff.py", + source / "benchmark/tests/test_checkpoint_joint_receiver.py"]}, + "image_id": image, "point_sha256": packet["sha256"], + "binary_sha256": {k: hashlib.sha256(p.read_bytes()).hexdigest() for k, p in binaries.items()}, + "ordinary_and_stride": True, "research_point_qualified": False} + (output / "result.json").write_text(json.dumps(receipt, indent=2)) + print(json.dumps(receipt)) + finally: + try: + if supervisor is not None: + supervisor.close() + finally: + if container is not None: + container.remove(force=True) + client.close() + fixture.close() + patch.undo() + + +if __name__ == "__main__": + main() diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py index 0e2dacc8fa..e487184ba6 100644 --- a/benchmark/tests/test_checkpoint_handoff.py +++ b/benchmark/tests/test_checkpoint_handoff.py @@ -1,6 +1,8 @@ """Private source rendezvous never substitutes a late or foreign point.""" import hashlib import json +import os +import sys import threading import time @@ -55,3 +57,106 @@ def test_surviving_request_is_not_overwritten(tmp_path): await_controller_capture(tmp_path / "checkpoints", [], {}, {"source_sha256": "later", "source_witness": {}}, time.time() + 1) assert ready.read_bytes() == b"original uncertain request" + + +@pytest.mark.skipif(sys.platform != "linux", reason="Qualifies the Linux writer process group") +@pytest.mark.parametrize("use_selector", [False, True]) +def test_failed_guard_returns_to_original_writer_without_interrupt(tmp_path, use_selector): + """Optional capture must not abort the caller's quota failure recovery.""" + from pathlib import Path + import socket + import sys + import tempfile + + from benchmark.runtime.checkpoint import held_quota_decision + from benchmark.runtime.worker import child_process + project = tmp_path / "source" + project.mkdir() + capture_root = project / "captures" + capture_root.mkdir(mode=0o700) + env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2])} + with tempfile.TemporaryDirectory(prefix="failed-guard-", dir="/tmp") as short: + socket_path = Path(short).resolve() / "host.sock" + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener: + listener.bind(str(socket_path)) + listener.listen(1) + listener.settimeout(30) + failed = {"ok": False, "mode": "should-run", "goal_id": "failed-source", + "decision": "skip", "should_run": False, + "error_code": "quota_unexpected_collection_error"} + packet = capture_root / "original" + packet.mkdir() + decision = packet / "decision.json" + raw = json.dumps(failed).encode() + decision.write_bytes(raw) + # The transport must release this failed packet without invoking + # capture admission or interrupting the independent original writer. + code = ( + "import socket,json; from pathlib import Path; " + "s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM); " + f"s.connect({str(socket_path)!r}); " + f"s.sendall({(json.dumps({'path': str(decision), 'sha256': hashlib.sha256(raw).hexdigest()}).encode() + bytes([10]))!r}); " + "d=json.loads(s.makefile('rb').readline()); assert d['released'] is True; " + f"assert d['sha256']=={hashlib.sha256(raw).hexdigest()!r}; " + f"Path('observed-failure.json').write_bytes(Path({str(decision)!r}).read_bytes()); " + "Path('source-work').write_text('original writer recovered')" + ) + with child_process([sys.executable, "-c", code], env={**env, "LOOPX_PROJECT": str(project)}, + stdout=None, stderr=None) as process: + process.stdin.close() + with held_quota_decision(listener, capture_root, goal_id="failed-source", + agent_id="failed-agent", turn_instance_id="failed-guard", + native_process=process, + interrupt_when=(lambda _: False) if use_selector else None) as held: + assert held["quota"]["ok"] is False + assert held["intent"]["capture"] is False + assert held["intent"]["execution_authority_granted"] is False + assert "native_process_return_code" not in held + assert process.poll() is None and not (project / "source-work").exists() + assert process.wait(timeout=30) == 0 + assert (project / "source-work").read_text() == "original writer recovered" + assert (project / "observed-failure.json").read_bytes().strip() == raw + assert held["path"].read_bytes() == raw + + +@pytest.mark.skipif(os.name != "posix", reason="Requires the Host Unix rendezvous") +@pytest.mark.parametrize("ok,goal", [(False, "foreign"), ("false", "source"), (None, "source")]) +def test_failure_dispatch_does_not_relax_foreign_or_malformed_readback(tmp_path, ok, goal): + import socket + import tempfile + from pathlib import Path + + from benchmark.runtime.checkpoint import held_quota_decision + from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError + + root = tmp_path / "decisions" + packet = root / "original" + packet.mkdir(parents=True) + decision = packet / "decision.json" + raw = json.dumps({"ok": ok, "goal_id": goal}).encode() + decision.write_bytes(raw) + responses = [] + with tempfile.TemporaryDirectory(prefix="invalid-guard-", dir="/tmp") as short: + address = str(Path(short).resolve() / "host.sock") + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener: + listener.bind(address) + listener.listen(1) + + def caller(): + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: + channel.connect(address) + channel.sendall(json.dumps({"path": str(decision), + "sha256": hashlib.sha256(raw).hexdigest()}).encode() + b"\n") + responses.append(channel.recv(4096)) + + thread = threading.Thread(target=caller) + thread.start() + try: + with pytest.raises(EffectRuntimeRemoteError): + with held_quota_decision(listener, root, goal_id="source", + agent_id="agent", turn_instance_id="turn"): + pytest.fail("invalid readback cannot enter a capture boundary") + finally: + thread.join(5) + assert not thread.is_alive() and responses == [b""] + assert decision.read_bytes() == raw diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index 7f6c5926bc..7db1aa7772 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/benchmark/tests/test_checkpoint_joint_receiver.py @@ -9,6 +9,7 @@ import os from pathlib import Path import shlex +import shutil import subprocess import sys import threading @@ -28,6 +29,7 @@ def test_captured_point_continues_with_fresh_authority( tmp_path, monkeypatch, capsys, request ): + assert os.getuid() != 0, "Workspace continuation must qualify the actual non-root writer" from benchmark.edgebench.checkpoint_controller import restore from benchmark.runtime.codex import Execution from benchmark.runtime.native_checkpoint import ( @@ -44,6 +46,9 @@ def test_captured_point_continues_with_fresh_authority( ) from loopx.control_plane.quota.settlement import read_heartbeat_settlement + # Each variant uses its own fresh guest because the native fork retains + # fixed captured workspace/home paths. The default variant stays unchanged. + stride_continuation = os.environ.get("LOOPX_JOINT_RECEIVER_STRIDE") == "1" unit = Path(os.environ["LOOPX_JOINT_RECEIVER_POINT"]) manifest = json.loads((unit / "checkpoint.json").read_bytes()) digest, metadata = manifest["sha256"], manifest["point_metadata"] @@ -53,17 +58,45 @@ def test_captured_point_continues_with_fresh_authority( if "completed_turn_id" in native: assert native["completed_turn_id"] == native["turn_ids"][-1] else: - assert native["interrupted_guard"]["decision_sha256"] == metadata["decision_sha256"] + assert ( + native["interrupted_guard"]["decision_sha256"] + == metadata["decision_sha256"] + ) config = Path(os.environ["LOOPX_JOINT_RECEIVER_CONFIG"]) workspace, home = Path(native["guest_workspace"]), Path(native["guest_home"]) assert not workspace.exists() and not home.exists(), ( "receiver must have fresh fixed guest paths" ) - workspace.mkdir(parents=True) + # Select the sealed declared workspace by its exact source path. Actual + # workers declare a directory with caller-owned IDs, not a single code file. + roots = [entry for entry in snapshot["components"] + if entry["source_path"] == str(workspace)] + assert len(roots) == 1, "Requires one declared full native workspace" + materialized = tmp_path / "materialized-point" + verified = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "materialize", "root": str(unit), "expected_sha256": digest, + "destination": str(materialized), + }) + assert verified["materialized"] and not verified["activation_performed"] + source_workspace = materialized / "components" / roots[0]["id"] + assert source_workspace.is_dir() + shutil.copytree(source_workspace, workspace) monkeypatch.chdir(workspace) candidate = workspace / "candidate.txt" - candidate.write_bytes((unit / "components/code").read_bytes()) captured_candidate = candidate.read_text() + # Retained source registry/state files are historical inputs. Current CLI + # authority must use a fresh registry/runtime outside this copied tree. + historical_registry = workspace / ".loopx/registry.json" + historical_bytes = historical_registry.read_bytes() + historical = json.loads(historical_bytes) + original_runtime = Path(historical["common_runtime_root"]) + assert not original_runtime.exists() + source_goal = metadata["loopx_state"]["goal_id"] + assert historical["goals"][0]["id"] == source_goal + assert (workspace / "evidence-6.json").read_bytes() == (source_workspace / "evidence-6.json").read_bytes() + retained_object = workspace / ".git/objects/ab/retained-object" + assert retained_object.read_text() == "Readonly source evidence" + assert retained_object.stat().st_mode & 0o777 == 0o444 registry, runtime = tmp_path / "registry.json", tmp_path / "runtime" goal, agent, turn = "joint-receiver", "joint-researcher", "joint-work" prefix = [ @@ -105,8 +138,27 @@ def cli(*args): agent, "--agent-work-mode", agent + "=active", + *( + [ + "--execution-stride-strategy", + "evidence_conditioned_v1", + "--execution-replan-after-turns", + "1", + "--explore-mode", + "planning", + ] + if stride_continuation + else [] + ), "--execute", ) + assert historical_registry.read_bytes() == historical_bytes + assert json.loads(registry.read_bytes())["goals"][0]["id"] == goal + assert runtime != original_runtime and not original_runtime.exists() + rejected = main([*prefix, "quota", "should-run", "--goal-id", source_goal, + "--agent-id", agent, "--turn-instance-id", "foreign-source"]) + capsys.readouterr() + assert rejected != 0, "Source Goal must not acquire destination authority" todo = cli( "todo", "add", @@ -221,15 +273,19 @@ def lose_binding_reply(self, *args): assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] _, branch, _, url, controller_files, grades = request.getfixturevalue("controllers") capsule = unit / "components" / snapshot["native_capsule_component_id"] - opportunity = json.loads((capsule / "checkpoint.json").read_bytes())["point_metadata"][ - "native_reference" - ]["opportunity"] + opportunity = json.loads((capsule / "checkpoint.json").read_bytes())[ + "point_metadata" + ]["native_reference"]["opportunity"] remaining_cooldown = opportunity["remaining_cooldown_seconds"] # Explicit fixture boundaries: either already eligible at capture, or a # cooldown longer than this entire 60-second native work window. A short # partial cooldown would make a timing-dependent expectation ambiguous. assert remaining_cooldown == 0 or remaining_cooldown > 60 cooldown_pending = remaining_cooldown > 60 + if stride_continuation and not cooldown_pending: + pytest.skip( + "Stride receiver qualification requires a captured pending cooldown" + ) phase = tmp_path / "phase-deadline" fresh = restore( controller_files[1], @@ -262,7 +318,25 @@ def lose_binding_reply(self, *args): ) (workspace / "judge.json").chmod(0o600) script = workspace / "continue.py" - script.write_text("""import json,time,requests + if stride_continuation: + script.write_text( + """import json,requests +from pathlib import Path +b=json.loads(Path('judge.json').read_text()); url,submission_token=b['url'],b['token'] +h=requests.get(url+'/api/v1/history',params={'token':submission_token},timeout=5).json() +assert h['best_score']==7 and h['agent_submissions']==1 +assert Path('candidate.txt').read_text()==CAPTURED_CANDIDATE +source=json.loads(Path(CONTEXT).read_text()) +assert source +Path('restoration-evidence.json').write_text(json.dumps({'code_unchanged':True,'public_incumbent':7,'consumed_submissions':1,'source_context_read':True})) +print('Restored state verified; source code unchanged, incumbent and consumed opportunities retained.') +""".replace("CAPTURED_CANDIDATE", repr(captured_candidate)).replace( + "CONTEXT", repr(adopted["artifact"]) + ) + ) + else: + script.write_text( + """import json,time,requests from pathlib import Path b=json.loads(Path('judge.json').read_text()); url,submission_token=b['url'],b['token'] h=requests.get(url+'/api/v1/history',params={'token':submission_token},timeout=5).json() @@ -283,9 +357,19 @@ def lose_binding_reply(self, *args): time.sleep(.01) assert result['status']=='completed' print('Retained candidate and public best 7 used; candidate 8 graded.') -""".replace("EXPECTED_COOLDOWN_PENDING", repr(cooldown_pending)) - .replace("CAPTURED_CANDIDATE", repr(captured_candidate))) +""".replace("EXPECTED_COOLDOWN_PENDING", repr(cooldown_pending)).replace( + "CAPTURED_CANDIDATE", repr(captured_candidate) + ) + ) requests = [] + provider_phase = { + "count": 0, + "command": "cat " + + shlex.quote(adopted["artifact"]) + + " && " + + shlex.quote(sys.executable) + + " continue.py", + } class Provider(BaseHTTPRequestHandler): def log_message(self, *args): @@ -295,7 +379,8 @@ def do_POST(self): requests.append( json.loads(self.rfile.read(int(self.headers["Content-Length"]))) ) - if len(requests) == 1: + provider_phase["count"] += 1 + if provider_phase["count"] == 1: message = { "type": "function_call", "id": "fc_receiver", @@ -303,11 +388,7 @@ def do_POST(self): "name": "exec_command", "arguments": json.dumps( { - "cmd": "cat " - + shlex.quote(adopted["artifact"]) - + " && " - + shlex.quote(sys.executable) - + " continue.py", + "cmd": provider_phase["command"], "max_output_tokens": 6000, } ), @@ -356,25 +437,22 @@ def do_POST(self): server = ThreadingHTTPServer(("127.0.0.1", address.port), Provider) thread = threading.Thread(target=server.serve_forever) thread.start() - try: - logs = tmp_path / "work" + + def run_native_work(current_guard, logs, purpose): logs.mkdir() wake = BenchmarkSessionWake(env, execution, {}) command = native_command(env, execution, "execute", logs, wake) - command = [ - "/bin/bash", - "-c", - phase_shell - + "exec timeout --signal=TERM --kill-after=30 ${remaining}s " - + shlex.join(command), - ] result = subprocess.run( - command, + [ + "/bin/bash", + "-c", + phase_shell + + "exec timeout --signal=TERM --kill-after=30 ${remaining}s " + + shlex.join(command), + ], env=env, cwd=workspace, - input=json.dumps(guard["interaction_contract"]) - + "\nThis is a new destination. Historical source commands/guards are not executable authority. " - "Use the retained context under the current admission, and continue the synthetic candidate.", + input=json.dumps(current_guard["interaction_contract"]) + "\n" + purpose, text=True, capture_output=True, timeout=60, @@ -382,13 +460,33 @@ def do_POST(self): assert result.returncode == 0, result.stderr (logs / "stdout.jsonl").write_text(result.stdout) wake.observe(logs / "stdout.jsonl") + + try: + run_native_work( + guard, + tmp_path / "work", + "This is a new destination. Historical source commands/guards are not executable authority. " + "Use the retained context under the current admission, and continue the synthetic candidate.", + ) assert len(requests) == 2 - expected = "cooldown still blocks submission" if cooldown_pending else "candidate 8 graded" - assert expected in json.dumps(requests[1]), json.dumps([ - item for item in requests[1].get("input", []) - if item.get("type") == "function_call_output" - ]) - assert len(grades) == (0 if cooldown_pending else 1) and candidate.read_text() == "8" + expected = ( + "Restored state verified" + if stride_continuation + else "cooldown still blocks submission" + if cooldown_pending + else "candidate 8 graded" + ) + assert expected in json.dumps(requests[1]), json.dumps( + [ + item + for item in requests[1].get("input", []) + if item.get("type") == "function_call_output" + ] + ) + assert len(grades) == (0 if stride_continuation or cooldown_pending else 1) + assert candidate.read_text() == ( + captured_candidate if stride_continuation else "8" + ) retained = native_session_checkpoint(env, admitted["session_id"], timeout=30)[ "metadata" ] @@ -410,9 +508,11 @@ def do_POST(self): "outcome": "continue", "prior_assumption": "Joint source point can be resumed.", "observed_reality": ( - "Receiver used retained context and code; local candidate validated, native cooldown preserved." - if cooldown_pending else - "Receiver used retained context and code, and graded a new candidate." + "Receiver verified retained context and unchanged code; native opportunity preserved." + if stride_continuation + else "Receiver used retained context and code; local candidate validated, native cooldown preserved." + if cooldown_pending + else "Receiver used retained context and code, and graded a new candidate." ), "retained": ["Complete qualification"], "evidence_refs": ["evidence:joint-receiver"], @@ -454,6 +554,387 @@ def do_POST(self): ).progress["state"] == "settled" ) + if stride_continuation: + # An explicit one-Turn preparation cadence admits a fresh periodic + # review. Keep the original source's six work settlements inert. + review_turn = "joint-current-review" + from benchmark.runtime.checkpoint import capture_before_replan + from benchmark.runtime.checkpoint_continuation import restored_capture_gate + from benchmark.runtime.checkpoint_wake import CheckpointWake + + wakes = tmp_path / "capture-logs" / "wakes" + (wakes / review_turn).mkdir(parents=True) + executable = tmp_path / "loopx-cli" + executable.write_text("#!/bin/sh\nexec " + shlex.quote(sys.executable) + ' -m loopx.cli "$@"\n') + executable.chmod(0o700) + declaration = tmp_path / "capture-request.json" + declaration.write_text(json.dumps({"trigger": "fresh_replan", + "destination": str(wakes.parent / "checkpoints"), "native_session": True, + "loopx_state": True, "components": [{"id": "code", "source_path": str(candidate)}], + "point_metadata": {}})) + env.update(LOOPX_CLI=str(executable), LOOPX_WAKE_LOG_DIR=str(wakes), + LOOPX_CHECKPOINT_REQUEST=str(declaration)) + pending_receipt = {"mode": "heartbeat", "stage": "execute", "turn_id": review_turn} + pending_body = capture_before_replan(env, pending_receipt, "Current destination contract.") + assert pending_receipt["checkpoint"]["reason"] == "restored_replan_pending", pending_receipt + assert not pending_receipt["checkpoint"]["captured"] + assert "Original pending packet" in pending_body and "checkpoint:" + digest in pending_body + # Both stop locations share the same gate. Repeated/restarted + # capture attempts neither interrupt the fork nor publish a point. + for _ in range(2): + gate = restored_capture_gate(env) + assert not gate["capture_allowed"] + hook = CheckpointWake(env, dict(pending_receipt)) + assert not hook._first_replan({"capture": True, "obligation_id": "replan-new-destination"}) + assert not (wakes.parent / "checkpoints").exists() + reviewed = cli( + "quota", + "should-run", + "--goal-id", + goal, + "--agent-id", + agent, + "--turn-instance-id", + review_turn, + "--runtime-profile", + "generic_cli", + "--available-capability", + "shell", + ) + assert reviewed["interaction_contract"]["mode"] == "autonomous_replan" + obligation = reviewed["autonomous_replan_obligation"] + assert any( + t["kind"] == "periodic_review_due" for t in obligation["triggers"] + ), obligation + reviewed = cli( + "quota", + "should-run", + "--goal-id", + goal, + "--agent-id", + agent, + "--todo-id", + todo, + "--turn-instance-id", + review_turn, + "--runtime-profile", + "generic_cli", + "--available-capability", + "shell", + ) + control = reviewed["replan_action_packet"]["stride_control"] + decision = json.loads(json.dumps(control["minimal_example"])) + decision.update( + basis="supported_progress", + action="grow", + axes={ + "work_slice": { + "plan": "Complete candidates 9 and 10 with local evidence in one work wake.", + "reason": "One real recovery check verified retained code, research context and public opportunity.", + }, + "route_distance": { + "plan": "Retain the synthetic increment method.", + "reason": "This qualifies restoration and decision adoption, not scientific strategy quality.", + }, + "validation_scope": { + "plan": "Check both candidate evidence files, final value 10 and unchanged native opportunity.", + "reason": "A complete larger work unit still needs its full postcondition.", + }, + }, + next_probe={ + "scope": "One two-candidate synthetic continuation.", + "prediction": "Both candidates are validated locally; captured cooldown remains binding.", + "adoption_criterion": "Evidence files and final value agree without resetting opportunity.", + "cost_bound": "One native work wake under the original remaining deadline.", + "minimum_evidence": "Actual native tool output, candidate bytes, current settlement and original native history.", + }, + ) + if "explore_review" in decision: + decision["explore_review"].update( + disposition="no_applicable_evidence", + considered_refs=[], + reason="The fresh graph contains no findings; retained source context is inert evidence, not reactivated graph authority.", + ) + identity = reviewed["heartbeat_receipt"]["settlement_identity"] + review_binding = ( + [ + "--goal-id", + goal, + "--agent-id", + agent, + "--turn-instance-id", + review_turn, + "--todo-id", + identity["todo_id"], + ] + if identity.get("todo_id") + else [ + "--goal-id", + goal, + "--agent-id", + agent, + "--turn-instance-id", + review_turn, + "--replan-obligation-id", + identity["replan_obligation_id"], + ] + ) + review_vision = tmp_path / "review-vision.json" + review_vision.write_text( + json.dumps( + { + "schema_version": "goal_vision_replan_contract_v0", + "state": "vision_patch_proposed", + "vision_patch": { + "acceptance_summary": "A current review must lead to a complete native experiment and preserve remaining opportunity." + }, + "path_delta": { + "outcome": "replan", + "prior_assumption": "A restored local candidate is only a continuation prerequisite.", + "observed_reality": "An actual destination recovery check verified unchanged code, retained research and native opportunity.", + "changed": [ + "Adopt a complete two-candidate work slice under current destination authority." + ], + "evidence_refs": ["evidence:restoration-check", "checkpoint:" + digest], + }, + "stride_decision": decision, + } + ) + ) + cli( + "refresh-state", + *review_binding, + "--autonomous-replan-recorded", + "--classification", + "joint_stride_review", + "--delivery-outcome", + "outcome_progress", + "--agent-vision-json", + str(review_vision), + "--no-global-sync", + "--suppress-external-sinks", + ) + # Interruption/lost delivery after the accepted writeback still + # owes this original Turn's settlement, so recapture stays closed. + unconsumed = restored_capture_gate(env) + assert not unconsumed["capture_allowed"] + original_packet = unit / "components" / metadata["held_decision_component_id"] + assert str(original_packet) in unconsumed["instructions"] + assert str(wakes / review_turn / "checkpoint-start-guard.json") not in unconsumed["instructions"] + cli( + "quota", + "spend-slot", + *review_binding, + "--source", + "heartbeat", + "--execute", + ) + reopened = restored_capture_gate(env) + assert reopened["capture_allowed"], reopened + assert not reopened["instructions"] # Do not repeat the historical packet after consumption. + assert not restored_capture_gate(env, obligation["obligation_id"])["capture_allowed"] + configured = cli( + "configure-goal", + "--goal-id", + goal, + "--execution-replan-after-turns", + "6", + "--execute", + ) + assert ( + configured["after"]["execution_profile"]["replan_after_effective_turns"] + == 6 + ) + larger = todo + revision = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)[ + "authority_read" + ]["provider_revision"] + cli( + "todo", + "update", + "--goal-id", + goal, + "--todo-id", + todo, + "--agent-id", + agent, + "--text", + "Complete synthetic candidates 9 and 10 as one larger experiment and preserve native opportunity.", + "--update-operation-id", + "receiver-grow-task", + "--update-expected-provider-revision", + revision, + "--task-lease-idempotency-key", + "receiver-work", + "--task-lease-expected-version", + str(lease["lease"]["version"]), + ) + next_turn = "wake-joint-larger-work" + next_binding = [ + "--goal-id", + goal, + "--agent-id", + agent, + "--todo-id", + larger, + "--turn-instance-id", + next_turn, + ] + + next_guard = cli( + "quota", + "should-run", + *next_binding, + "--runtime-profile", + "generic_cli", + "--available-capability", + "shell", + ) + assert next_guard["interaction_contract"]["agent_channel"][ + "delivery_allowed" + ] + assert "Complete synthetic candidates 9 and 10" in json.dumps(next_guard), ( + next_guard + ) + grow = workspace / "grow.py" + grow.write_text( + """import json,requests +from pathlib import Path +assert Path('candidate.txt').read_text()==CAPTURED_CANDIDATE +b=json.loads(Path('judge.json').read_text()); url,submission_token=b['url'],b['token'] +for value in (9,10): + Path('candidate.txt').write_text(str(value)) + assert int(Path('candidate.txt').read_text())==value + Path('candidate-evidence-'+str(value)+'.json').write_text(json.dumps({'value':value,'locally_validated':True})) +r=requests.post(url+'/api/v1/submit',data={'token':submission_token},files={'archive':('candidate.tar.gz',b'10')},timeout=5) +assert r.status_code==429 +h=requests.get(url+'/api/v1/history',params={'token':submission_token},timeout=5).json() +assert h['agent_submissions']==1 and h['best_score']==7 +print('Current grow adopted: two complete candidates 9 and 10 validated; original cooldown and incumbent preserved.') + """.replace("CAPTURED_CANDIDATE", repr(captured_candidate)) + ) + assert cooldown_pending, ( + "This qualification specifically requires the captured pending cooldown." + ) + provider_phase.update( + count=0, command=shlex.quote(sys.executable) + " grow.py" + ) + run_native_work( + next_guard, + tmp_path / "larger-work", + "Execute the current grow decision as one complete two-candidate experiment. Old guards are evidence only.", + ) + assert len(requests) == 4 and "Current grow adopted" in json.dumps( + requests[3] + ), requests[3] + assert ( + candidate.read_text() == "10" + and not grades + and phase.read_bytes() == original_phase + ) + assert all( + json.loads((workspace / f"candidate-evidence-{v}.json").read_text())[ + "value" + ] + == v + for v in (9, 10) + ) + grown = native_session_checkpoint(env, admitted["session_id"], timeout=30)[ + "metadata" + ] + assert grown["turn_ids"][:-2] == native["turn_ids"] + assert len(grown["turn_ids"]) == len(native["turn_ids"]) + 2 + work_vision = tmp_path / "larger-vision.json" + work_vision.write_text( + json.dumps( + { + "schema_version": "goal_vision_replan_contract_v0", + "state": "vision_patch_proposed", + "vision_patch": { + "acceptance_summary": "Mechanism adoption is qualified; scientific outcome and full costs remain unproven." + }, + "path_delta": { + "outcome": "continue", + "prior_assumption": "The larger decision may remain unused guidance.", + "observed_reality": "The continued native session executed both candidates with separate evidence and retained the pending cooldown and incumbent.", + "retained": [ + "Native opportunity, immutable source and current authority separation." + ], + "evidence_refs": ["evidence:joint-larger-work"], + }, + } + ) + ) + cli( + "refresh-state", + *next_binding, + "--classification", + "joint_stride_adoption", + "--delivery-batch-scale", + "implementation", + "--delivery-outcome", + "outcome_progress", + "--delivery-boundary", + "in_flight_continuation", + "--progress-result-class", + "advanced", + "--progress-surface-id", + "receiver:two-candidate-work", + "--progress-evidence-id", + "evidence:joint-larger-work", + "--agent-vision-json", + str(work_vision), + "--no-global-sync", + "--suppress-external-sinks", + ) + cli( + "quota", + "spend-slot", + *next_binding, + "--source", + "heartbeat", + "--execute", + ) + final = cli( + "quota", + "should-run", + "--goal-id", + goal, + "--agent-id", + agent, + "--turn-instance-id", + "joint-next-audit", + "--runtime-profile", + "generic_cli", + "--available-capability", + "shell", + ) + assert final["interaction_contract"]["mode"] == "bounded_delivery" + # The next genuinely new decision boundary can capture changed + # research/code/native history, after actual useful continuation. + # This partial point deliberately makes no evaluator qualification. + prior_wake = wakes / next_turn + prior_wake.mkdir() + (prior_wake / "receipt.json").write_text(json.dumps({"turn_id": next_turn, + "mode": "heartbeat", "stage": "execute", "ok": True, + "session": {"binding_scope": "agent", "action": "resume", "session_id": admitted["session_id"]}})) + cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute") + boundary_turn = "joint-next-capture" + (wakes / boundary_turn).mkdir() + boundary = {"turn_id": boundary_turn, "mode": "heartbeat", "stage": "execute", "ok": False} + capture_before_replan(env, boundary, "Current new decision contract.") + if boundary["checkpoint"].get("reason") == "no_fresh_replan": + # Capture eligibility does not manufacture a quota obligation. + # This source may not have another due review yet. + assert boundary["checkpoint_restore"]["capture_allowed"] + else: + assert boundary["checkpoint"]["captured"], json.dumps(boundary, indent=2) + recaptured = Path(boundary["checkpoint"]["destination"]) + assert (recaptured / "components/code").read_text() == "10" + next_metadata = json.loads((recaptured / "checkpoint.json").read_text())["point_metadata"] + assert next_metadata["replan"]["obligation_id"] != obligation["obligation_id"] + assert next_metadata["held_decision_component_id"] + assert next_metadata["native_session"]["turn_ids"] == grown["turn_ids"] effect_runtime_result( "runtime.checkpoint_artifacts", {"action": "verify", "root": str(unit), "expected_sha256": digest}, diff --git a/benchmark/tests/test_checkpoint_qualification.py b/benchmark/tests/test_checkpoint_qualification.py new file mode 100644 index 0000000000..b0e4eb04b7 --- /dev/null +++ b/benchmark/tests/test_checkpoint_qualification.py @@ -0,0 +1,22 @@ +"""Qualification must distinguish actual execution from pytest's zero exit.""" +import pytest + +from benchmark.tests.qualify_checkpoint_joint import require_one_pass + + +@pytest.mark.parametrize("cases,passes", [ + ('', True), + ('', False), + ('', False), + ('', False), + ('', False), + ('', False), +]) +def test_qualification_requires_one_executed_success(tmp_path, cases, passes): + report = tmp_path / "junit.xml" + report.write_text(f"{cases}") + if passes: + require_one_pass(report) + else: + with pytest.raises(ValueError, match="actual pass and no skip"): + require_one_pass(report) diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index 404542a0fd..302d7efb37 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/benchmark/tests/test_checkpoint_worker.py @@ -2,6 +2,7 @@ from __future__ import annotations import io +import hashlib import json import os import sqlite3 @@ -76,6 +77,30 @@ def counter(container): return int(archive.extractfile("counter").read()) +def test_ordinary_controller_stages_root_private_native_configuration(writer, tmp_path): + from benchmark.tests.qualify_checkpoint_joint import _stage_source_configuration + + assert os.getuid() != 0, "Qualify the ordinary controller, not a privileged reader" + container, _ = writer + raw = b'model = "fixture"\n' + command = ("from pathlib import Path; p=Path('/private-codex'); p.mkdir(mode=0o700); " + f"f=p/'config.toml'; f.write_bytes({raw!r}); f.chmod(0o600)") + assert container.exec_run(["python", "-c", command], user="root").exit_code == 0 + assert container.exec_run(["stat", "-c", "%u %a", "/private-codex", + "/private-codex/config.toml"]).output == b"0 700\n0 600\n" + rejected = container.exec_run(["python", "-c", + "from pathlib import Path; Path('/private-codex/config.toml').read_bytes()"], user=str(os.getuid())) + assert rejected.exit_code != 0 and b"PermissionError" in rejected.output + native = {"guest_home": "/private-codex", "config_sha256": hashlib.sha256(raw).hexdigest()} + config = _stage_source_configuration(container, native, tmp_path / "staged") + assert config.read_bytes() == raw and config.stat().st_uid == os.getuid() + assert config.stat().st_mode & 0o777 == 0o600 + with pytest.raises(ValueError, match="differs from the sealed checkpoint"): + _stage_source_configuration(container, {**native, "config_sha256": "f" * 64}, tmp_path / "wrong") + assert container.exec_run(["stat", "-c", "%u %a", "/private-codex", + "/private-codex/config.toml"]).output == b"0 700\n0 600\n" + + def test_archive_keeps_private_directory_modes_without_special_bits(writer, tmp_path): from benchmark.edgebench.checkpoint_worker import _archive container, _ = writer diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index 529417f404..6f8e765432 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/benchmark/tests/test_native_checkpoint_handoff.py @@ -350,7 +350,9 @@ def test_checkpoint_interrupt_timeout_reaps_without_qualifying_capture(tmp_path) def test_midwake_native_profile_refusal_precedes_interruption(tmp_path, monkeypatch, platform, native, interrupt): from benchmark.runtime.checkpoint_wake import CheckpointWake - wake = CheckpointWake({"LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes")}, {"turn_id": "fixture-turn"}) + wake = CheckpointWake({"LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"), + "LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), "LOOPX_GOAL_ID": "fixture", + "LOOPX_AGENT_ID": "researcher"}, {"turn_id": "fixture-turn"}) wake.request, wake.raw = {"native_session": native}, b"fixture declaration" monkeypatch.setattr(sys, "platform", platform) intent = {"capture": True, "obligation_id": "replan-0123456789abcdef"} @@ -389,6 +391,10 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp logs = Path("/logs/agent/wakes") candidate = project / "candidate.txt" candidate.write_text("0") + git_object = project / ".git" / "objects" / "ab" / "retained-object" + git_object.parent.mkdir(parents=True) + git_object.write_text("Readonly source evidence") + git_object.chmod(0o444) executable = tmp_path / "loopx-cli" executable.write_text("#!/bin/sh\nexec " + shlex.quote(sys.executable) + ' -m loopx.cli "$@"\n') executable.chmod(0o700) @@ -501,7 +507,8 @@ def settle_work(result, expected): assert len(set(work_turns)) == 6 and len(calls) == 13 request = tmp_path / "request.json" request.write_text(json.dumps({"destination": str(logs.parent / "checkpoints"), - "components": [{"id": "code", "source_path": str(candidate)}], "point_metadata": {}, + "components": [{"id": "task_workspace", "source_path": str(project)}], + "point_metadata": {}, "trigger": "fresh_replan", "native_session": True, "loopx_state": True, "evaluation_opportunity": True, "controller_handoff": True})) phase.update(kind="review", tool_sent=False) diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md index a17c860fe5..13103ea9a9 100644 --- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md +++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md @@ -1008,6 +1008,70 @@ mid-wake native capture before signalling and releases the original tool respons Linux interrupted and completed-wake capture retain their separate qualification; strict history, ambiguity, source and deadline predicates remain unchanged. +An explicit destination preparation can reuse the existing cadence owner for +its first review: configure `--execution-replan-after-turns 1`, perform and settle +one actual restoration check, then admit the newly due periodic review. That +check reads retained research/code and public opportunity without changing the +candidate or submitting it. The review uses current Goal/Todo/lease and fresh +context identities; source settlements remain historical evidence. After the +review settles, restore the preregistered normal cadence through +`configure-goal` before admitting research work. The isolated native receiver +qualification records independent grow axes, adopts one two-candidate complete +slice and verifies unchanged deadline, pending cooldown, consumed rounds and +incumbent. The default receiver path is exercised separately. The public +[`qualify_checkpoint_joint.py`](../../../benchmark/tests/qualify_checkpoint_joint.py) +entry reproduces the six-work-wake source with a non-root controller, then runs +ordinary and stride receivers in separate fresh guests; the +[benchmark procedure](../../../benchmark/README.md#controller-only-native-capsule-transport) +pins the checkpoint provider, prepares guest dependencies and lists explicit +native CLI/Node inputs. Missing prerequisites fail rather than skip qualification. +Its private receipt and logs retain input provenance; read-only Git members +exercise the controller boundary. The source now declares its whole workspace +directory with a caller-owned component ID. Non-root receivers select that +directory by the sealed original workspace path and use the typed materialization +owner; retained source registry/state files remain historical beside a fresh +external registry/runtime. Admission of the source Goal through the fresh registry +is rejected, while retained evidence and actual candidate writes remain usable. +This qualifies the directory/writer seam, not a real task dataset or scientific +continuation result. This is an +explicit tested composition, not an automatic restore hook or a new default. +A Todo-successor repair alone is not a spendable periodic strategy review. +Matched branches must declare the same preparation/cadence transition and charge +its full cost. Scripted native-provider execution qualifies transport and +mechanical adoption, not model judgment, restored Explore graph authority or +scientific benefit; later-point selection and real branch admission remain open. + +Re-enabling capture on a restored conversation requires a destination decision, +not merely a new Goal/Agent or a locally unseen replan ID. Both pre-model and +mid-wake hooks refuse capture until the canonical current-Goal settlement +contains an accepted semantic replan whose Vision evidence references +`checkpoint:`, and the destination native history has completed +a continuation with the original full history prefix intact. The consumed +obligation itself remains ineligible; a subsequent genuinely new node can be +captured. Retry, restart and a lost admission response preserve the original +source-root/digest binding. This is a machine-enforced capture barrier, not +additional execution or budget authority. + +Completed pre-model points now retain the original pending guard as a sealed +participant, as interrupted points already do. The pending packet is read as +historical decision evidence under the destination's current guard; source +commands, claims, identities and effects must not execute again. Older pending +points without that participant cannot re-enable capture. After consumption the +Host stops repeating this context. Continuation reads the digest-bound manifest +basis and verifies the particular pending packet and native history it uses; +it does not rehash large code/incumbent archives every work wake or claim that +this limited read is fresh full-member verification. Native admission still +requires full artifact verification. Capture alone never automatically restores +a branch. An external restore-every-new-point orchestrator must exercise the same +barrier and current admission; its existence or successful scientific outcome +does not follow from this receiver qualification. + +The artifact copy also preserves read-only file modes without requesting write +access merely to sync the copied bytes. A non-root controller capture of a +`0444` participant failed before this fix and now succeeds; Git object stores +are an ordinary source of such members. Root-only container qualification was +insufficient for this Host boundary. Source permissions and bytes remain unchanged. + Committed native session admission now retains the captured artifact digest in the original atomic session binding. Exact controller retries after a lost binding reply verify that receipt, current Goal/Agent/profile and unchanged complete diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md index 063a397057..be7d9ed88c 100644 --- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md +++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md @@ -817,6 +817,50 @@ pending command；延迟重读及仅向主进程发信号均未恢复。因此 前明确拒绝中途 native 捕获，继续原工具响应。Linux interrupted 与 completed-wake 路径分别保留验收，不放宽历史、歧义、源或期限检查。 +目标侧显式恢复准备可复用既有 cadence owner 来安排首次复盘：配置 +`--execution-replan-after-turns 1`，完成并结算一次真实恢复检查，再准入新到期的 +periodic review。检查读取保留的研究/代码与公开评测机会，不改候选、不提交。 +复盘使用当前 Goal/Todo/lease 及新 context 身份；源工作结算仅作历史证据。 +复盘结算后，通过 `configure-goal` 恢复预注册的正常频率，再准入研究工作。 +隔离的 native receiver 验证已记录三个轴独立计划的 grow，实际采用一次包含 +两个候选的完整工作切片，并检查 deadline、未到期冷却、已用轮次和 incumbent +不变；默认 receiver 路径单独验证。公开的 +[`qualify_checkpoint_joint.py`](../../../benchmark/tests/qualify_checkpoint_joint.py) +入口先由非 root controller 捕获六个真实工作 wake 后的源状态，再在两个全新 +guest 分别运行普通和 stride receiver； +[benchmark 操作步骤](../../../benchmark/README.md#controller-only-native-capsule-transport) +固定 checkpoint provider、准备 guest 依赖并列出显式 Codex/Node 输入。 +缺少前置条件会报错，不以 skip 充当资格通过。私有 receipt 与日志保留输入来源， +只读 Git 对象覆盖 controller 的权限边界。 +源声明整个 workspace 目录并保留调用者的 component ID；非 root receiver 按 +封存的原 workspace 路径定位，由类型化 owner 物化。复制的源 registry/state +只是历史，新外置 registry/runtime 拒绝原 Goal 准入，同时保留证据并完成候选 +写入。这资格化目录与写入者衔接，不证明真实题目数据集或科研接续收益。 +这是显式且经测试的组合，不是恢复时自动 +触发的 hook，也不改变默认行为。补后继 Todo 的生命周期修复本身不等于可记账的 +periodic 策略复盘。匹配分支须声明相同的准备/频率迁移，并计入完整成本。 +脚本式 native provider 验证传输与机械采用，不证明模型判断、源 Explore graph +权限恢复或科研收益；中后期研究点选择与真实分支准入仍待完成。 + +恢复会话重新开启捕获须完成目标侧决策，换一个 Goal/Agent 或本地尚未见过的 +replan ID 都不够。模型前与 wake 中两处 hook 共用门槛：当前 Goal 的 canonical +结算须接受一个语义 replan，其 Vision 证据引用 `checkpoint:`， +且目标原生历史已完成接续、保留原完整历史前缀。刚消费的义务本身仍不可捕获； +后续真正的新节点才可捕获。重试、重启或准入回复丢失保留原 source-root/digest +绑定。这是机器执行的捕获门槛，不增加执行或预算权限。 + +模型前的 completed 点现在也将原 pending guard 作为封存参与方，中断点已有 +此参与方。目标在当前 guard 下将原包读作历史决策依据，不重跑源命令、claim、 +身份与 effect。旧 pending 点缺少此参与方时不能重新开启捕获。消费后 Host 不再 +重复注入这段上下文。接续读取 digest 绑定的 manifest 依据，核验实际使用的 +pending 包与原生历史，不在每个工作 wake 重算大型代码/incumbent archive 的哈希， +也不把这个有限读取称为新的全成员验证；原生准入仍须完整验证产物。捕获本身 +不自动恢复分支。外层“每个新点都恢复”的编排也须通过同一门槛与当前准入； +其是否存在以及科研收益不能由本 receiver 验证推导。 + +产物复制保留只读文件模式，同步复制的字节时不额外索取写权限。非 root 控制器 +捕获 `0444` 参与方此前失败，修复后通过；Git object store 是这类成员的常见来源。 +仅在 root 容器中验证不足以覆盖此 Host 边界。源权限和字节保持不变。 inert verification unit 不授予执行权。目标侧使用新的实验/Host 身份，通过既有 Goal/session/state admission 进入。历史租约、quota、timer、pending effect 和 diff --git a/loopx/control_plane/runtime/checkpoint_artifacts.ts b/loopx/control_plane/runtime/checkpoint_artifacts.ts index fc799a856e..21ecfbefab 100644 --- a/loopx/control_plane/runtime/checkpoint_artifacts.ts +++ b/loopx/control_plane/runtime/checkpoint_artifacts.ts @@ -7,7 +7,7 @@ import {createReadStream} from "node:fs"; import {copyFile, lstat, mkdir, mkdtemp, open, readFile, readdir, rm} from "node:fs/promises"; import {dirname, isAbsolute, join, resolve, sep} from "node:path"; import type {JsonObject} from "../effect_program.ts"; -import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {jsonObject, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; import {canonicalAuthoritySha256, hasExactAuthorityKeys} from "../coordination/authority_store_codec.ts"; import {snapshotSqliteDatabase} from "../coordination/sqlite_backup.ts"; import {durableWriteJson, publishNewDirectory} from "../effect_runtime_io.ts"; @@ -15,6 +15,7 @@ import {settlementIdentity} from "../effect_program.ts"; import {normalizeHeartbeatReplanObligationId, normalizeHeartbeatTodoId} from "../quota/heartbeat_receipt_identity.ts"; import {EffectRuntimeConflictError} from "../effect_runtime_errors.ts"; import {EffectiveAction} from "../quota/effective_action.generated.ts"; +import {BARE_SHA256_PATTERN} from "../content_digest.ts"; const SCHEMA = "loopx_checkpoint_artifact_set_v0"; type Component = {id: string; source_path: string}; @@ -134,7 +135,9 @@ async function copyComponent(source: string, destination: string, snapshotDataba // SQLite's existing physical snapshot owner seals committed WAL contents. if (snapshotDatabases && await sqliteFile(source)) await snapshotSqliteDatabase(source, destination); else await copyFile(source, destination); - const handle = await open(destination, "r+"); + // copyFile preserves a read-only participant's mode. Syncing its bytes + // needs no write access; r+ fails for an ordinary non-root controller. + const handle = await open(destination, "r"); try {await handle.sync();} finally {await handle.close();} } else throw new Error("checkpoint source contains an unsupported file type"); } @@ -166,6 +169,42 @@ async function verify(root: string, expected: unknown): Promise { } export async function checkpointArtifacts(value: unknown): Promise { const request = requireJsonObject(value, "checkpoint artifacts request"); + if (request.action === "restore_capture") { + if (!hasExactAuthorityKeys(request, ["action", "checkpoint_sha256", "goal_id", "agent_id", "current_obligation_id", "review"])) { + throw new Error("checkpoint restored capture fields are invalid"); + } + const sha = requireNonEmptyString(request.checkpoint_sha256, "restored checkpoint digest"); + if (!BARE_SHA256_PATTERN.test(sha)) throw new Error("invalid restored checkpoint digest"); + const goal = requireNonEmptyString(request.goal_id, "checkpoint destination Goal"); + const agent = requireNonEmptyString(request.agent_id, "checkpoint destination Agent"); + if (request.current_obligation_id !== null && !normalizeHeartbeatReplanObligationId(request.current_obligation_id)) { + throw new Error("restored checkpoint current replan identity is invalid"); + } + const review = jsonObject(request.review), identity = jsonObject(review?.identity); + const progress = jsonObject(review?.progress), run = jsonObject(review?.writeback_run); + const guard = jsonObject(review?.semantic_replan_guard), ack = jsonObject(run?.autonomous_replan_ack); + const delta = jsonObject(ack?.semantic_delta), path = jsonObject(jsonObject(run?.agent_vision)?.path_delta); + const id = normalizeHeartbeatReplanObligationId(delta?.obligation_id); + const todo = normalizeHeartbeatTodoId(identity?.todo_id); + const boundReplan = normalizeHeartbeatReplanObligationId(identity?.replan_obligation_id); + // The adapter must obtain these records from the existing canonical + // settlement owner. Copied source runs, slot debit alone, a neutral ACK or + // the still-pending decision being captured cannot release this barrier. + const accepted = !!identity && !!run && identity.goal_id === goal && identity.agent_id === agent && + run.goal_id === goal && run.agent_id === agent && run.turn_instance_id === identity.turn_instance_id && + normalizeHeartbeatTodoId(run.todo_id) === todo && + normalizeHeartbeatReplanObligationId(run.replan_obligation_id) === boundReplan && + typeof identity.turn_instance_id === "string" && identity.turn_instance_id.length > 0 && + identity.effect_id === settlementIdentity({goal_id: goal, agent_id: agent, todo_id: todo, + replan_obligation_id: boundReplan, turn_instance_id: identity.turn_instance_id}).effect_id && + progress?.state === "settled" && ack?.schema_version === "autonomous_replan_ack_v0" && ack.recorded === true && + delta?.schema_version === "replan_semantic_delta_v0" && delta.accepted === true && !!id && + guard?.selected_obligation_id === id && request.current_obligation_id !== id && + Array.isArray(path?.evidence_refs) && path.evidence_refs.includes(`checkpoint:${sha}`); + return {capture_allowed: accepted, reason: accepted ? "restored_replan_completed" : "restored_replan_pending", + ...(accepted ? {review_obligation_id: id} : {}), + execution_authority_granted: false}; + } if (request.action === "select_replan") { if (!hasExactAuthorityKeys(request, ["action", "quota", "goal_id", "agent_id", "turn_instance_id"])) { throw new Error("checkpoint replan selection fields are invalid"); @@ -209,6 +248,17 @@ export async function checkpointArtifacts(value: unknown): Promise { return {capture: true, obligation_id: id, settlement_effect_id: identity.effect_id, execution_authority_granted: false}; } + if (request.action === "read_basis") { + if (!hasExactAuthorityKeys(request, ["action", "root", "expected_sha256"])) throw new Error("checkpoint basis fields are invalid"); + const root = absolute(request.root, "checkpoint root"); + await physical(join(root, "checkpoint.json")); + const manifest = manifestBody(JSON.parse(await readFile(join(root, "checkpoint.json"), "utf8"))); + if (manifest.sha256 !== request.expected_sha256) throw new Error("checkpoint reviewed digest mismatch"); + // This reads a digest-bound declaration after admission's full member + // verification. It does not claim current integrity of every component. + return {point_metadata: manifest.point_metadata, components: manifest.components, + integrity_verified: false, execution_authority_granted: false}; + } const fields = request.action === "probe" ? ["action", "components"] : request.action === "verify" ? ["action", "root", "expected_sha256"] : request.action === "capture" ? ["action", "components", "destination", "point_metadata", "expected_source_sha256"] : diff --git a/loopx/control_plane/turn_driver/codex_sessions.py b/loopx/control_plane/turn_driver/codex_sessions.py index 7381765d09..cf89b51682 100644 --- a/loopx/control_plane/turn_driver/codex_sessions.py +++ b/loopx/control_plane/turn_driver/codex_sessions.py @@ -177,9 +177,14 @@ def _store_codex_cli_session( operation_reasoning_effort: str | None = None, require_absent: bool = False, checkpoint_artifact_sha256: str | None = None, + checkpoint_artifact_root: str | None = None, ) -> None: if checkpoint_artifact_sha256 is not None and not re.fullmatch(r"[0-9a-f]{64}", checkpoint_artifact_sha256): raise ValueError("checkpoint session source digest is invalid") + if checkpoint_artifact_root is not None and ( + checkpoint_artifact_sha256 is None or not Path(checkpoint_artifact_root).is_absolute() or + str(Path(checkpoint_artifact_root).resolve()) != checkpoint_artifact_root): + raise ValueError("checkpoint session source root must be physical and digest-bound") path = _session_path(runtime_root, lineage, session_scope) with exclusive_file_lock(path): if require_absent and (path.exists() or path.is_symlink()): @@ -188,15 +193,22 @@ def _store_codex_cli_session( # but cannot migrate to a replacement session, profile or Goal lifetime. prior = load_codex_cli_session(runtime_root, lineage=lineage, session_scope=session_scope) retained_source = None + retained_root = None if prior and (prior.get("session_id"), prior.get("goal_ref"), prior.get("session_profile_digest")) == ( session_id, dict(goal_ref) if goal_ref is not None else None, session_profile_digest ): retained_source = prior.get("checkpoint_artifact_sha256") + retained_root = prior.get("checkpoint_artifact_root") if retained_source is not None: if not isinstance(retained_source, str) or not re.fullmatch(r"[0-9a-f]{64}", retained_source): raise ValueError("stored checkpoint session source digest is invalid") if checkpoint_artifact_sha256 not in (None, retained_source): raise ValueError("checkpoint session source digest cannot change") + if retained_root is not None and (not isinstance(retained_root, str) or + not Path(retained_root).is_absolute() or str(Path(retained_root).resolve()) != retained_root): + raise ValueError("stored checkpoint session source root is invalid") + if retained_root is not None and checkpoint_artifact_root not in (None, retained_root): + raise ValueError("checkpoint session source root cannot change") _write_codex_cli_session( runtime_root, lineage=lineage, @@ -208,6 +220,7 @@ def _store_codex_cli_session( operation_model=operation_model, operation_reasoning_effort=operation_reasoning_effort, checkpoint_artifact_sha256=checkpoint_artifact_sha256 or retained_source, + checkpoint_artifact_root=checkpoint_artifact_root or retained_root, ) @@ -223,6 +236,7 @@ def _write_codex_cli_session( operation_model: str | None = None, operation_reasoning_effort: str | None = None, checkpoint_artifact_sha256: str | None = None, + checkpoint_artifact_root: str | None = None, ) -> None: normalized_session_id = _valid_session_id(session_id) if not normalized_session_id: @@ -253,6 +267,8 @@ def _write_codex_cli_session( payload["goal_ref"] = dict(goal_ref) if checkpoint_artifact_sha256 is not None: payload["checkpoint_artifact_sha256"] = checkpoint_artifact_sha256 + if checkpoint_artifact_root is not None: + payload["checkpoint_artifact_root"] = checkpoint_artifact_root if operation_profile_digest is not None: payload["operation_transport"] = "app-server-operation-tools-v0" payload["operation_profile_digest"] = operation_profile_digest diff --git a/tests/control_plane/test_checkpoint_restore_capture.py b/tests/control_plane/test_checkpoint_restore_capture.py new file mode 100644 index 0000000000..ff05b25b2a --- /dev/null +++ b/tests/control_plane/test_checkpoint_restore_capture.py @@ -0,0 +1,70 @@ +"""A restored fork needs a current, settled semantic decision before recapture.""" +import copy + +import pytest + +from loopx.control_plane.effect_program import SettlementIdentity +from loopx.control_plane.effect_runtime import effect_runtime_result + + +@pytest.mark.parametrize("fault", [None, "pending", "neutral", "source_actor", "unlinked", "same_node", "wrong_binding"]) +def test_restored_capture_requires_current_consumption_not_local_dedup(fault): + sha = "a" * 64 + identity = SettlementIdentity("destination", "researcher", "todo_probe", "decision-1").as_dict() + obligation = "replan-" + "1" * 16 + review = {"identity": identity, "progress": {"state": "settled"}, + "semantic_replan_guard": {"schema_version": "semantic_replan_guard_v0", "selected_obligation_id": obligation}, + "writeback_run": {"goal_id": "destination", "agent_id": "researcher", "todo_id": "todo_probe", + "turn_instance_id": "decision-1", "autonomous_replan_ack": { + "schema_version": "autonomous_replan_ack_v0", "recorded": True, + "semantic_delta": {"schema_version": "replan_semantic_delta_v0", "accepted": True, + "obligation_id": obligation}}, + "agent_vision": {"path_delta": {"evidence_refs": ["checkpoint:" + sha]}}}} + current = "replan-" + "2" * 16 + if fault == "pending": + review["progress"]["state"] = "settlement_pending" + elif fault == "neutral": + review["writeback_run"]["autonomous_replan_ack"]["semantic_delta"]["accepted"] = False + elif fault == "source_actor": + review["writeback_run"]["goal_id"] = "source" + elif fault == "unlinked": + review["writeback_run"]["agent_vision"]["path_delta"]["evidence_refs"] = ["checkpoint:" + "b" * 64] + elif fault == "same_node": + current = obligation + elif fault == "wrong_binding": + review["writeback_run"]["todo_id"] = "todo_other" + request = {"action": "restore_capture", "checkpoint_sha256": sha, + "goal_id": "destination", "agent_id": "researcher", "current_obligation_id": current, + "review": review} + result = effect_runtime_result("runtime.checkpoint_artifacts", request) + assert result["capture_allowed"] is (fault is None) + assert result["execution_authority_granted"] is False + # A fresh Goal/Agent cannot inherit even a valid destination review. This + # is the outer restore-every-point counterexample to local-ID dedup. + another = copy.deepcopy(request) + another.update(goal_id="another-destination", agent_id="another-researcher") + assert not effect_runtime_result("runtime.checkpoint_artifacts", another)["capture_allowed"] + + +def test_readonly_participant_capture_and_limited_basis_read(tmp_path): + source = tmp_path / "code" + source.write_text("retained evidence") + source.chmod(0o444) + components = [{"id": "code", "source_path": str(source)}] + probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) + unit = tmp_path / "point" + captured = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components, + "destination": str(unit), "point_metadata": {}, "expected_source_sha256": probe["source_sha256"]}, retry_safe=False) + copied = unit / "components" / "code" + assert copied.read_text() == source.read_text() and copied.stat().st_mode & 0o777 == 0o444 + copied.chmod(0o644) + copied.write_text("changed member") + # Digest-bound declaration readback is explicitly weaker than full member + # verification. Native admission must continue using the latter. + basis = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "read_basis", "root": str(unit), + "expected_sha256": captured["sha256"]}) + assert basis["integrity_verified"] is False and basis["execution_authority_granted"] is False + from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError + with pytest.raises(EffectRuntimeRemoteError): + effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(unit), + "expected_sha256": captured["sha256"]}) diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts index 8d049af41f..25c24a4179 100644 --- a/tests/control_plane_ts/content_digest_single_owner.test.ts +++ b/tests/control_plane_ts/content_digest_single_owner.test.ts @@ -132,6 +132,7 @@ const CANONICAL_CONSUMERS = [ "control_plane/presentation/action_review_plan.ts", "control_plane/quota/refresh_external_delivery.ts", "control_plane/quota/turn_envelope.ts", + "control_plane/runtime/checkpoint_artifacts.ts", "control_plane/runtime/execution_identity.ts", "control_plane/runtime/usage_statistics_cycles.ts", "control_plane/runtime/usage_statistics_goal_contract.ts",