From dee5bef25aeb36171222ce500297b5d2eee1c8be Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:13:18 +0800 Subject: [PATCH] fix(benchmark): preserve complete LoopX state on warm replacement Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 134 ++++----- benchmark/edgebench/run.py | 9 +- benchmark/runtime/checkpoint_continuation.py | 13 +- .../runtime/checkpoint_session_refresh.py | 5 +- benchmark/runtime/harbor.py | 9 +- benchmark/runtime/layout.py | 2 + benchmark/runtime/sforge.py | 11 +- benchmark/runtime/sforge_warm_start.py | 261 +++++++----------- benchmark/runtime/worker.py | 3 - benchmark/tests/test_native_checkpoint.py | 136 +++------ benchmark/tests/test_sforge_warm_start.py | 61 ++-- docs/reference/research-checkpoint.md | 30 +- .../turn_driver/codex_agent_session.py | 9 +- loopx/hosts/codex_checkpoint.py | 36 ++- loopx/hosts/research_checkpoint.py | 45 ++- .../test_full_loopx_checkpoint_restore.py | 231 ++++++++++++++++ .../test_warm_continuation_readback.py | 203 -------------- 17 files changed, 599 insertions(+), 599 deletions(-) create mode 100644 tests/control_plane/test_full_loopx_checkpoint_restore.py delete mode 100644 tests/control_plane/test_warm_continuation_readback.py diff --git a/benchmark/README.md b/benchmark/README.md index 6aa4cb2ed8..1e8edc0e35 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1156,35 +1156,42 @@ controller 文件及 handoff 开关即可关闭；移除 checkpoint 请求关闭 The native heartbeat launcher accepts a complete point through `--restore-checkpoint POINT --restore-sha256 SHA --restore-config CONFIG ---restore-goal NEW_GOAL --restore-agent NEW_AGENT --restore-source-todo SOURCE_TODO`. +--restore-goal SOURCE_GOAL --restore-agent SOURCE_AGENT --restore-source-todo SOURCE_TODO`. All six inputs are required together. Keep the existing native `--eval-interval 0`, full `--capture-before-replan` handoff and `--checkpoint-controller-file` configuration. The native provider must expose `RestoredRegistration`; an older provider fails instead of cold-registering. -This adapter composes existing artifact, Codex session, Todo-continuation and +This adapter composes existing artifact, canonical state, Codex session and native-judge owners. A captured parent directory may contain the native task workspace: the receiver resolves exactly one physical task subtree from the verified component and restores only that subtree. Captured siblings remain inert unless separately selected as research data; absent, ambiguous or noncanonical roots reject before the receiver workspace changes. -It preserves the captured workspace and effective Codex configuration, admits -a distinct native session and fresh Goal/Agent, and adopts -retained research as evidence on a current preparation Todo. Historical registry, -claims, leases and settlement records are not installed as current authority. -If the sealed context contains Explore events, adoption also seeds the fresh -Goal's ordinary Explore log. Nodes, findings, scoped counterexamples, topology, -original attribution and timestamps survive, with each event bound to the -checkpoint digest and source event through its opaque run reference. Operational -research-observation lineage stays in the archive; it is not destination -execution proof. Recovery accepts only the same immutable initial prefix, may -finish a partial prefix, and never reapplies old statuses after newer evidence. -Malformed, foreign or interleaved destination history rejects before model work. -Legacy context-only points keep their original continuation behavior. -Fresh credentials come from the ordinary runner credential source. The native -activation's absolute deadline includes receiver installation and preparation; -restarting the entry cannot grant another full budget or submission allowance. +Warm replacement preserves the captured workspace, single-Goal registry and +complete LoopX runtime, including canonical SQLite/File state, Todo identities, +dependencies, receipts, quota counters, Vision and Explore history. The original +Goal/Agent identities are required. It forks the retained native conversation +and replaces only the exact captured session binding through current Goal +admission and compare-and-swap. A changed binding, occupied runtime or existing +registry rejects; preserve a failed attempt and use a new isolated receiver. +Stop the source worker before replacement. No active user runtime is overwritten. + +There is no preparation-Todo fallback or temporary replan-cadence override. +The restored configuration is read back against the declared treatment without +bootstrapping a new Goal, reseeding the task queue or rewriting its receipts. +Current guards, leases, recovery, user gates and stops decide execution. Closed +Turns remain settled; restoration grants no extra budget, opportunity or capture. +The absolute native deadline still includes receiver installation. Fresh +credentials come from the ordinary runner credential source. + +热恢复接续原 Goal/Agent、完整 SQLite/File 状态、任务依赖、结算、额度和 +Vision/Explore 历史；不另建 preparation Todo，不临时调整重规划频率。 +必须使用已停止源的全新隔离接收端；占用、分歧或部分恢复均拒绝覆盖。 +已有结算不能重放，当前 guard/lease、用户 gate 和原始预算仍然生效。 +需要跨 Goal 复制研究证据时，另用 Core 的 `research-checkpoint adopt-context`； +资料导入不构成热恢复。 To continue with a different model or reasoning effort, retain the exact sealed source settings in `--restore-config` and explicitly pass @@ -1233,10 +1240,9 @@ Native conversation admission and fresh-replan revalidation allow up to 90 secon per read, capped by the original phase deadline, while preserving full-history, cursor and identity checks. Core research-context adoption also gives complete artifact and native-history reads up to 90 seconds each (previously the generic -10-second artifact RPC and 30-second native read defaults). Warm preparation -passes its original phase deadline through adoption; expiration rejects the next -write, and uncertain prior writes must replay their original operation. Ordinary -Core callers remain lease-governed and may supply an absolute adoption deadline. +10-second artifact RPC and 30-second native read defaults). Ordinary Core import +callers remain lease-governed and may supply an absolute adoption deadline; this +separate import path does not configure a benchmark warm replacement. These allowances grant no extra execution time and do not establish sustained performance for growing checkpoints. @@ -1246,72 +1252,31 @@ credentials; an existing bank member must match the sealed bytes. Conflicting or partial members fail closed. A lock timeout leaves the attempt for inspection; it does not authorize reusing a failed trial directory. -Both comparison arms first settle actual preparation at cadence one and consume -the retained decision through a current semantic replan. The existing joint -continuation gate must accept that replan before the declared normal cadence -returns. Preparation and context-reading cost belong in the comparison. Omit all -restore flags for the unchanged cold entry; never reuse an ambiguous trial -folder as a new attempt. - -Warm preparation retains the original evidence-reading task on its existing Todo; -it does not add a prompt paragraph interpreting historical deadlines. Current -Goal/guard authority, task time constraints, native cooldowns and user stops -remain binding. Preparation reads the current claimed Todo and actual quota -result before reporting `preparation_actionable=true` in the existing `checkpoint-context.json` -receipt. Closed or unavailable work and a deadline that expires during readback -fail before model execution; an old ready plan is not current admission. This -readback neither grants execution nor qualifies subsequent research or recapture. -It applies only to the existing explicitly admitted complete-checkpoint path; -ordinary terminal archives are not converted into complete checkpoints. - -The receiver brackets the guard with the exact preparation Todo's canonical -authority revision. Completion, transfer or another canonical edit during that -observation rejects readiness; a later required review cannot revive an earlier -open snapshot. The existing context receipt includes `preparation_readback` with -that revision, the selected Todo and current replan identity, so an operator can -inspect what was actually observed instead of relying on a readiness boolean. -This is a bounded observation, not an execution lease; the worker must still use -fresh ordinary admission. Readiness is validated before changing cadence, so a -rejected preparation does not change the Goal's cadence. An ambiguous cadence -write must be inspected, not blindly rolled back across an owed current review. -Creation retries read the original `checkpoint-preparation` receipt and replay -its exact accepted task through the canonical create owner, including receivers -created with the earlier longer task. They never mint a second operation or Todo. -This covers partial creation, not arbitrary retries after context adoption or -worker execution. Such uncertain operations retain their original receipt, -revision and lease recovery boundary; actual model/research adoption is separate. - -Warm 准备保留既有 Todo 中原本的证据读取任务，不再增加解释历史截止时间的提示段落。 -当前 Goal、guard、题目时限、原生冷却、用户门槛及停止权限仍有效。 -准备步骤须读回当前 claimed Todo 和实际 quota 结果，才在既有 -`checkpoint-context.json` 回执中报告 `preparation_actionable=true`；工作已闭合、不可用或读回 -期间期限耗尽时，在模型执行前失败。旧 ready 计划不等于当前准入；此读回不授予 -执行权，也不认证后续研究或再次捕获，仅作用于显式准入的完整检查点路径。 -普通终态资料归档不会因此被升级为完整检查点。 - -接收端用准确准备 Todo 的规范权威修订号夹住 guard 读取；期间完成、转交或其他规范 -修改都会拒绝 ready，后来的 required review 不能复活较早的 open 快照。既有上下文 -回执的 `preparation_readback` 返回该修订、当前选中 Todo 与 replan 身份，便于检查 -实际观察依据；它不是执行租约，worker 仍须重新走普通准入。先验证准备资格，再改 -cadence，因此拒绝准备不会改变 Goal 的 cadence；不确定的配置写入须检查，不能盲目 -回滚而抹掉欠下的当前复盘。创建重试读回原 `checkpoint-preparation` 回执，通过规范 -创建 owner 重放当时接受的准确任务（兼容此前较长任务），不创建第二个操作或 Todo。 -此兼容范围是部分创建，不是上下文采用或 worker 执行后的任意重试；不确定的操作仍须 -保留原回执、修订与租约的恢复边界，真实模型/研究采用也需单独验证。 +Both comparison arms retain the original task graph and replan cadence. The +existing joint continuation gate still requires a current checkpoint-linked +semantic review before recapture; it does not create a preparation Todo or +force a review every work Turn. State verification and context-reading cost +belong in the comparison. Omit restore flags for the cold entry; never reuse an +ambiguous receiver as a new attempt. A complete point must contain canonical +LoopX state; a context-only archive is not a warm-replacement fallback. + +两组接续原任务队列与重规划频率；再次捕获仍须通过已有的当前复盘与结算校验， +不新增 preparation Todo，不强制每轮复盘。状态验证和上下文读取计入完整成本。 +仅有资料的归档不能冒充完整热恢复点，失败接收端须保留现场并使用新隔离实例。 To test a new decision from the same pre-replan state, use `--restore-fresh-replan`. The point must contain a verified pending replan packet. -The first destination model call starts a fresh native conversation, before -preparation and the first current-Goal replan. It reads the original task, +The first destination model call starts a fresh native conversation before +the retained pending replan, preserving the complete Goal and Todo state. It reads the original task, retained research evidence, code and incumbent; no post-checkpoint decision or -old native message prefix is supplied. Preparation remains charged work, and +old native message prefix is supplied. Context-reading remains charged work, and capture remains closed until the ordinary current review/settlement gate passes. Omit the flag for retained-history continuation. Change treatments only in a new attempt; the same checkpoint budget, opportunities and cooldown remain binding. 同一个决策前 checkpoint 的重新决策实验使用 `--restore-fresh-replan`：首次模型调用 -就在 fresh 原生会话中准备并完成当前目标的 replan，读取原任务、研究证据、代码和最佳 -候选，不预置 checkpoint 之后的决定或旧原生消息前缀。准备成本仍计入原预算，后续捕获 +就在 fresh 原生会话中接续已有任务状态并处理待完成的 replan，读取原任务、研究证据、代码和最佳 +候选，不预置 checkpoint 之后的决定或旧原生消息前缀。上下文读取成本仍计入原预算，后续捕获 仍须通过当前复盘和结算校验。省略该选项保持历史接续。 必须用新 attempt 改变处理方式，不能重置预算、机会或冷却。 @@ -1375,10 +1340,13 @@ accepted round counts, cooldown and scoring remain unchanged. Omit the option for the original-budget behavior; a changed budget requires a fresh destination, not a rewrite of a running study or its original results. -The ordinary installed worker has passed a synthetic joined qualification: -native session continuation, current preparation and accepted replan, restored -normal cadence, candidate work and released native feedback under the original -remaining allowance. It reused one immutable captured prefix and preserved its +The earlier import-based worker passed a synthetic joined qualification: +native session continuation, a new-Goal preparation/review, candidate work and +released native feedback under the original remaining allowance. That fixture +does not qualify the current full-state replacement. The full-state tests use +real File/SQLite owners and an actual Codex Host with local deterministic +inference to check retained work selection, unchanged settlement history, +current user gates and binding compare-and-swap. It reused one immutable captured prefix and preserved its source. Native app-server-generated skills are preserved outside discovery before linking the current installed skills; ordinary skill identity checks still apply. This used real Codex processes and native HTTP with deterministic local inference diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 2068e743f7..a614af4c7e 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -153,7 +153,7 @@ def main(argv=None): parser.add_argument("--checkpoint-request", type=Path, help="Explicit host JSON declaration staged into a heartbeat worker; omitted disables capture") parser.add_argument("--restore-checkpoint", type=Path, - help="Explicit complete checkpoint to continue under fresh native/Goal identities") + help="Explicit complete checkpoint to continue the retained Goal/Agent in an isolated replacement") parser.add_argument("--restore-sha256") retention = parser.add_mutually_exclusive_group() retention.add_argument("--checkpoint-retention-policy", type=Path, @@ -169,7 +169,7 @@ def main(argv=None): parser.add_argument("--restore-research-component", action="append", default=[], help="Opt-in captured file component ID to restore at its original research path; repeat per component") parser.add_argument("--restore-fresh-replan", action="store_true", - help="Start a fresh conversation before checkpoint preparation and the first current replan") + help="Start a fresh conversation before the retained pending replan while preserving full Goal state") parser.add_argument("--restore-phase-extension-seconds", type=int, default=0, help="Explicit authorized extra continuation time; omitted preserves the checkpoint's remaining phase") parser.add_argument("--eval-interval", type=int, @@ -238,7 +238,7 @@ def main(argv=None): restore_values = (args.restore_sha256, args.restore_config, args.restore_goal, args.restore_agent, args.restore_source_todo) if bool(args.restore_checkpoint) != all(restore_values) or (not args.restore_checkpoint and any(restore_values)): - parser.error("Warm continuation requires checkpoint, digest, config, fresh Goal/Agent and source Todo together") + parser.error("Warm continuation requires checkpoint, digest, config, source Goal/Agent and source Todo together") if (not 0 <= args.restore_phase_extension_seconds < 2 ** 53 or (args.restore_phase_extension_seconds and not args.restore_checkpoint)): parser.error("A non-negative phase extension requires an explicit warm checkpoint continuation") @@ -435,7 +435,8 @@ def main(argv=None): **({"restore_fresh_replan": True} if args.restore_fresh_replan else {}), **({"restored_checkpoint_sha256": warm_start.sha256, "original_deadline_epoch": warm_start.deadline_epoch, - "receiver_preparation_cadence": 1, "normal_replan_cadence": warm_start.normal_cadence} + "restored_loopx_state": True, + "normal_replan_cadence": warm_start.normal_cadence} if warm_start is not None else {}), **({"online_admission_epoch": sampler.epoch, "offline_scoring_complete": False} if sampler is not None else {}), diff --git a/benchmark/runtime/checkpoint_continuation.py b/benchmark/runtime/checkpoint_continuation.py index d4a606866f..60035e4cc5 100644 --- a/benchmark/runtime/checkpoint_continuation.py +++ b/benchmark/runtime/checkpoint_continuation.py @@ -127,12 +127,21 @@ def _restored_context(env, current_obligation_id=None, *, verify_native, include result["review_turn_id"] = guard_path.parent.name break result["checkpoint_sha256"] = sha + source = metadata.get("loopx_state", {}) + same_lineage = all(source.get(key) == value for key, value in lineage.items()) + lineage_instructions = ( + "The retained Goal/Agent/Todo identities and settlement history are continuing state. " + "Do not execute archived commands or replay settled effects. Current guard/recovery and " + "lease owners decide which original or new Turn may execute and settle. " + if same_lineage else + "The archived packet is historical decision evidence: do not execute its commands, reuse its " + "Goal/Agent/Turn identities, claims or spend authority. " + ) 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 " + + lineage_instructions + "Follow this wake's current guard " "for selection, writeback and settlement. Creating a successor Todo alone does not record " "this checkpoint review. If the current guard provides original-Turn closeout, include the " "checkpoint-linked agent_vision in that writeback and finish its settlement. If the Turn " diff --git a/benchmark/runtime/checkpoint_session_refresh.py b/benchmark/runtime/checkpoint_session_refresh.py index c898238811..d64d7cf808 100644 --- a/benchmark/runtime/checkpoint_session_refresh.py +++ b/benchmark/runtime/checkpoint_session_refresh.py @@ -83,8 +83,9 @@ def _prepare_fresh_replan(env, session, wake, declaration): "No earlier native messages or decision made after that checkpoint are supplied. " "Read the original task objective and scoring/acceptance rules from the current task document, " "then the retained research evidence, failed alternatives, code and best validated candidate. " - "Historical plans are evidence to reconsider, not a decision to inherit. Complete the current " - "preparation and replan under the current guard before choosing research work. " + "Historical plans are evidence to reconsider, not a decision to inherit. The retained Goal, " + "Todos, dependencies and settlement history continue unchanged. Follow the current guard " + "to complete the pending replan or its original-Turn recovery before choosing research work. " "Keep the original submission opportunities, cooldown and cumulative deadline. " "Do not resume archived commands or consult archived native conversation messages to choose " "the route; use the checkpoint research evidence and original objective.\n\n" diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index e5f355fb05..8b8597418a 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -22,7 +22,7 @@ from .codex_offline import MODEL_CATALOG_REMOTE, CodexOffline, stage_portable_python from .codex import DEFAULT_REPLAN_AFTER_TURNS, Execution, turn_envelope_receipt -from .layout import _ROOT, _SRC, _CONTROL, _CODEX_HOME +from .layout import _ROOT, _SRC, _CONTROL, _CODEX_HOME, _REGISTRY, _LOOPX_RUNTIME _PYTHON = f"{_ROOT}/python" @@ -32,8 +32,6 @@ _SHARED_CODEX_HOME = f"{_PROFILE}/codex-home" _SHARED_SKILLS = f"{_SHARED_CODEX_HOME}/skills" _CLI = f"{_PROFILE}/bin/loopx" -_REGISTRY = f"{_CONTROL}/registry.json" -_LOOPX_RUNTIME = f"{_ROOT}/state/runtime" _SCHEDULER_STATE = f"{_CONTROL}/scheduler-state.json" _TASK_DOC = f"{_CONTROL}/task.md" _BASH_ENV = f"{_CONTROL}/bash-env" @@ -544,6 +542,11 @@ async def _prepare_phase( if self.execution.task_entry == "seeded-todo": await self._seed_phase(environment, cwd=cwd) + await self._verify_phase_configuration(environment, cwd=cwd) + + async def _verify_phase_configuration(self, environment: BaseEnvironment, *, cwd: str) -> None: + """Read the treatment without reconfiguring a restored Goal or its queue.""" + key, _, expected = self._replan_configuration() cadence = await self._loopx( environment, ["configure-goal", "--goal-id", self.goal_id], cwd=cwd, ) diff --git a/benchmark/runtime/layout.py b/benchmark/runtime/layout.py index d263180418..f8dec8ceee 100644 --- a/benchmark/runtime/layout.py +++ b/benchmark/runtime/layout.py @@ -4,3 +4,5 @@ _SRC = f"{_ROOT}/source" _CONTROL = f"{_ROOT}/control" _CODEX_HOME = f"{_ROOT}/codex-home" +_REGISTRY = f"{_CONTROL}/registry.json" +_LOOPX_RUNTIME = f"{_ROOT}/state/runtime" diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index 634647543d..c3b3a325c0 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -450,8 +450,15 @@ def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resu raise RuntimeError("Native task prompt is unavailable") self.runtime._phase_number = 1 if self.runtime.execution.uses_loopx: - asyncio.run(self.runtime._prepare_phase(self.environment, result.stdout, cwd=self.cwd)) - if self.profile == "heartbeat-explore": + if self.warm_start is not None: + # The sealed Goal, task graph, review counters and strategy + # are the continuing state. Do not bootstrap, seed another + # task, or reset an existing configuration during restore. + asyncio.run(self.runtime._write_task_document(self.environment, result.stdout)) + asyncio.run(self.runtime._verify_phase_configuration(self.environment, cwd=self.cwd)) + else: + asyncio.run(self.runtime._prepare_phase(self.environment, result.stdout, cwd=self.cwd)) + if self.profile == "heartbeat-explore" and self.warm_start is None: observed = asyncio.run(self.runtime._loopx(self.environment, [ "configure-goal", "--goal-id", self.runtime.goal_id, "--explore-graph-enabled", "--explore-harness-enabled", diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index f70155500f..f7f48fe0ab 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -1,7 +1,7 @@ """Explicit SForge receiver composition; existing owners retain all authority. This is an operator transport, not a checkpoint format or a second admission -rule. The artifact, native session, Todo handoff and judge each verify their own +rule. The artifact, canonical state, native session and judge each verify their own participant before the ordinary worker can start. """ from __future__ import annotations @@ -19,19 +19,10 @@ import tomllib from loopx.control_plane.effect_runtime import effect_runtime_result -from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _SRC +from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _SRC, _REGISTRY, _LOOPX_RUNTIME _CHECKPOINT_VERIFY_TIMEOUT_SECONDS = 60 -_PREPARATION_TASK = ( - "Read the full retained checkpoint research and pending decision, verify the current code, " - "incumbent and remaining native opportunities. Do not modify candidates or submit during " - "this preparation. Settle actual findings under current authority, then complete the first " - "current replan before choosing a complete research experiment." -) -# Compatibility for the already accepted longer preparation request. Never -# submit that guidance to a new receiver or rewrite an existing create receipt. -_LEGACY_PREPARATION_TEXT_SHA256 = "c1480d8fcec20909567f39bac1b0af2a0a3d3703c56556f4356e5b3e43e0ba45" @dataclass(frozen=True) @@ -61,7 +52,7 @@ def _verify(self, workspace: str) -> dict: or self.deadline_epoch <= time.time() or any(not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", v) for v in (self.goal_id, self.agent_id, self.source_todo_id))): - raise ValueError("Warm start needs fresh explicit identities and its original open deadline") + raise ValueError("Warm start needs explicit source identities and its original open deadline") paths = (self.root, self.config) if self.destination_config is not None: paths += (self.destination_config,) @@ -77,9 +68,11 @@ def _verify(self, workspace: str) -> dict: timeout=min(_CHECKPOINT_VERIFY_TIMEOUT_SECONDS, remaining), large_local_snapshot=True) metadata = json.loads((self.root / "checkpoint.json").read_bytes())["point_metadata"] native, source = metadata["native_session"], metadata["loopx_state"] + same = source["goal_id"] == self.goal_id and source["agent_id"] == self.agent_id if (native["guest_workspace"] != workspace or native["guest_home"] != _CODEX_HOME - or source["goal_id"] == self.goal_id or source["agent_id"] == self.agent_id): + or not same): raise ValueError("Warm start paths, configuration or destination lineage differ") + _loopx_components(self.root, metadata, Path(workspace)) from loopx.hosts.codex_checkpoint import checkpoint_destination_config checkpoint_destination_config(self.config, native["config_sha256"], self.destination_config) _workspace_component(self.root, metadata, Path(workspace)) @@ -174,9 +167,10 @@ async def _stage(self, environment, workspace: str, *, if result.return_code: raise RuntimeError("Could not stage the verified checkpoint participants") # The root-only materializer preserves sealed modes. Its receiver must - # transfer these selected data copies to the worker before preparation; + # transfer these selected data copies to the worker before continuation; # making private directories world-readable would change the artifact. - return [str(target) for _, _, target in research] + state = _loopx_components(self.root, metadata, Path(workspace)) + return [str(target) for _, _, target in [*research, *state]] # Provider-owned data locations; selecting a component never grants permission @@ -248,6 +242,60 @@ def _check_research_destinations(entries): raise ValueError("Research destination requires an existing physical parent") +def _loopx_components(unit: Path, metadata: dict, workspace: Path) -> list[tuple[str, Path, Path]]: + """Resolve the backup owner's inventory for a fixed-path isolated replacement. + + This copies complete provider files, not a Todo projection or a rewritten + journal. The ordinary registry, canonical snapshot and session owners still + validate the restored values before a worker can obtain current admission. + """ + state = metadata["loopx_state"] + inventory = state["source_inventory"] + result = [] + for key, target in (("runtime_root", Path(_LOOPX_RUNTIME)), + ("configuration_source_registry", Path(_REGISTRY))): + entries = [entry for entry in inventory if entry["key"] == key] + if len(entries) != 1 or entries[0]["source_path"] != str(target): + raise ValueError("Full LoopX restore requires its original isolated runtime and registry paths") + entry = entries[0] + component = entry["component_id"] + if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,63}", component): + raise ValueError("LoopX state component ID is invalid") + source = unit / "components" / component + if key == "configuration_source_registry": + if state["registry_component_id"] != component: + raise ValueError("LoopX registry inventory differs from its sealed registry participant") + source = source / state["registry_member_path"] + if source.resolve(strict=True) != source or not source.is_relative_to(unit / "components"): + raise ValueError("LoopX state member is not a physical sealed participant") + if (key == "runtime_root" and not source.is_dir()) or (key != "runtime_root" and not source.is_file()): + raise ValueError("LoopX state participant kind differs") + result.append((component, source, target)) + if any(entry["key"] not in {"runtime_root", "configuration_source_registry"} + and not Path(entry["source_path"]).is_relative_to(workspace) for entry in inventory): + raise ValueError("LoopX state inventory contains an unsupported external route") + raw = result[1][1].read_bytes() + if hashlib.sha256(raw).hexdigest() != state["registry_sha256"]: + raise ValueError("LoopX source registry digest differs") + registry = json.loads(raw) + goals = registry.get("goals", []) + if (registry.get("common_runtime_root") != _LOOPX_RUNTIME or len(goals) != 1 + or goals[0].get("id") != state["goal_id"] or goals[0].get("repo") != str(workspace) + or state["agent_id"] not in goals[0].get("coordination", {}).get("registered_agents", [])): + raise ValueError("Full LoopX restore requires the captured single-Goal registry and Agent") + return result + + +def _check_loopx_destinations(entries): + for _, source, target in entries: + if target.resolve() != target or not target.parent.is_dir() or target.is_symlink(): + raise ValueError("LoopX restore destination is redirected or unavailable") + # The ordinary installer creates the empty runtime directory. Never + # replace a configured receiver, including an earlier partial restore. + if target.exists() and (not source.is_dir() or not target.is_dir() or any(target.iterdir())): + raise ValueError("LoopX receiver already has state; preserve the original attempt") + + def _phase_read_timeout(declaration: dict, limit: float) -> float: deadline = declaration["deadline_epoch"] if type(deadline) is not int or not 0 < deadline < 2 ** 53: @@ -259,7 +307,7 @@ def _phase_read_timeout(declaration: dict, limit: float) -> float: def restore_workspace(declaration: dict, workspace: Path): - """Materialize one isolated receiver workspace; never import its registry.""" + """Copy task/research files only; this transport does not configure a Goal.""" unit = Path(declaration["root"]) effect_runtime_result("runtime.checkpoint_artifacts", { "action": "verify", "root": str(unit), "expected_sha256": declaration["sha256"]}, @@ -304,6 +352,40 @@ def restore_workspace(declaration: dict, workspace: Path): _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS) +def restore_worker(declaration: dict, workspace: Path): + """Replace one stopped isolated worker with its complete retained state.""" + unit = Path(declaration["root"]) + effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "verify", "root": str(unit), "expected_sha256": declaration["sha256"]}, + timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS), large_local_snapshot=True) + manifest = json.loads((unit / "checkpoint.json").read_bytes()) + state = _loopx_components(unit, manifest["point_metadata"], workspace) + _check_loopx_destinations(state) + restore_workspace(declaration, workspace) + for _, captured, target in state: + if captured.is_dir(): + if target.exists(): + target.rmdir() # Only the verified empty installer directory. + shutil.copytree(captured, target) + else: + with target.open("xb") as output, captured.open("rb") as input_file: + shutil.copyfileobj(input_file, output) + shutil.copystat(captured, target) + if state: + copied = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", + "components": [{"id": key, "source_path": str(target)} for key, _, target in state]}, + timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS), large_local_snapshot=True) + for key, captured, _ in state: + member = captured.relative_to(unit / "components" / key).as_posix() + expected = manifest["artifact_witness"][key] + if member != ".": + expected = [{**entry, "path": entry["path"][len(member):].lstrip("/")} + for entry in expected if entry["path"] == member or entry["path"].startswith(member + "/")] + if copied["source_witness"][key] != expected: + raise ValueError("Restored LoopX files differ from the sealed state; preserve this failed receiver") + _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS) + + def _prepare_native_skills(home: Path, installed: Path) -> None: """Leave skill installation to the ordinary worker after native fork. @@ -329,7 +411,7 @@ def _prepare_native_skills(home: Path, installed: Path) -> None: def admit_warm_session(env: dict[str, str], declaration: dict) -> dict: """Recheck the full native participant within its original remaining phase. - Mature histories need the same bounded read allowance at initial preparation + Mature histories need the same bounded read allowance at initial continuation and fresh-context revalidation. This retains source/cursor checks; it neither trusts a cached history digest nor grants another execution window. """ @@ -338,6 +420,7 @@ def admit_warm_session(env: dict[str, str], declaration: dict) -> dict: timeout = _phase_read_timeout(declaration, 90) result = admit_native_checkpoint_session(env, Path(declaration["root"]), declaration["sha256"], Path(declaration["config"]), sandbox=env["LOOPX_CODEX_SANDBOX"], timeout=timeout, + restore_loopx_state=True, **({"destination_config": Path(declaration["destination_config"])} if declaration.get("destination_config") is not None else {}), **({"provider_base_url": declaration["provider_base_url"]} @@ -346,150 +429,20 @@ def admit_warm_session(env: dict[str, str], declaration: dict) -> dict: return result -def _preparation_todo(cli, goal: str, agent: str) -> str: - """Replay the exact accepted create payload; history grants no readiness.""" - operation = "checkpoint-preparation" - found = cli("todo", "receipt", "--goal-id", goal, "--operation-id", operation) - text = _PREPARATION_TASK - if found.get("status") == "found": - receipts = found.get("receipts") - original = receipts[0] if isinstance(receipts, list) and len(receipts) == 1 else {} - todo = original.get("todo") or {} - # Canonical presentation may decorate text with a priority prefix; - # title retains the create intent. The create owner checks its digest. - text = todo.get("title", todo.get("text")) - if (original.get("schema_version") != "loopx_coordination_todo_create_receipt_v0" - or original.get("goal_id") != goal or original.get("operation_id") != operation - or not isinstance(text, str) - or (text != _PREPARATION_TASK and hashlib.sha256(text.encode()).hexdigest() - != _LEGACY_PREPARATION_TEXT_SHA256)): - raise ValueError("Warm preparation operation names another request; inspect its original receipt") - elif found.get("status") != "missing": - raise ValueError("Warm preparation create receipt is unavailable; preserve its operation identity") - # The canonical owner still compares the complete request digest, including - # owner/capabilities/class. Do not derive a second Todo or accept a receipt - # as permission to execute a completed/transferred preparation. - return cli("todo", "add", "--goal-id", goal, "--role", "agent", "--claimed-by", agent, - "--operation-id", operation, "--required-capability", "shell", - "--task-class", "advancement_task", "--text", text)["todo_id"] - - def prepare_context(env: dict[str, str], declaration: dict) -> dict: - """Join actual session + research owners on one freshly bootstrapped Goal.""" - from loopx.hosts.research_checkpoint import adopt_todo_context - from benchmark.runtime.worker import loopx_command - import subprocess - - unit, digest = Path(declaration["root"]), declaration["sha256"] + """Verify native continuation of the complete retained Goal; seed no work.""" if any(env[key] != declaration[field] for key, field in (("LOOPX_GOAL_ID", "goal_id"), ("LOOPX_AGENT_ID", "agent_id"))): - raise ValueError("Warm context requires its declared destination lineage") - if declaration["deadline_epoch"] <= time.time(): - raise ValueError("Original warm start allowance expired during installation") + raise ValueError("Warm context requires its declared source lineage") + _phase_read_timeout(declaration, 90) native = admit_warm_session(env, declaration) _prepare_native_skills(Path(env["CODEX_HOME"]), Path(env["LOOPX_SHARED_SKILLS"])) - def cli(*args): - result = subprocess.run([*loopx_command(env), *args], env=env, - cwd=env["LOOPX_PROJECT"], text=True, capture_output=True, - timeout=_phase_read_timeout(declaration, 60), check=True) - value = json.loads(result.stdout) - if value.get("ok") is not True: - raise ValueError("Checkpoint destination operation failed: " + args[0]) - return value - goal, agent = declaration["goal_id"], declaration["agent_id"] - # Preparation is actual current work, not imported completion. The normal - # model/guard must inspect the retained evidence and settle this Todo. - todo = _preparation_todo(cli, goal, agent) - key = "checkpoint-preparation" - lease = cli("task-lease", "acquire", "--goal-id", goal, "--todo-id", todo, - "--owner", agent, "--idempotency-key", key)["lease"] - revision = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)["authority_read"]["provider_revision"] - context = adopt_todo_context(env, unit, digest, - source_todo_id=declaration["source_todo_id"], todo_id=todo, - operation_id="checkpoint-context", expected_revision=revision, - lease_key=key, lease_version=lease["version"], deadline=declaration["deadline_epoch"]) - # The first work guard owns the next execution lease. Context adoption is - # finished before this lease is released; no historical lease is imported. - cli("task-lease", "release", "--goal-id", goal, "--todo-id", todo, - "--owner", agent, "--idempotency-key", key, "--expected-version", str(lease["version"])) - readback = _read_preparation(cli, goal, agent, todo, declaration) - # Rejected readiness must not mutate cadence. Once this succeeds the worker - # still acquires fresh ordinary admission; no best-effort rollback may erase - # an owed current review after an ambiguous configuration response. - cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute") - return {"session": native, "context": context, "preparation_todo_id": todo, - "preparation_readback": readback, - "preparation_actionable": True, "model_calls": 0, + return {"loopx_state_restored": True, "session": native, + "goal_id": declaration["goal_id"], "agent_id": declaration["agent_id"], + "model_calls": 0, "execution_authority_granted": False, "research_continuation_qualified": False} - -def _read_preparation(cli, goal, agent, todo, declaration): - """Read current owner facts; a saved plan alone cannot admit continuation.""" - _phase_read_timeout(declaration, 60) - before = cli("todo", "list", "--goal-id", goal, "--todo-id", todo) - preparation = before["todo"] - if preparation.get("status") != "open" or preparation.get("claimed_by") != agent: - raise ValueError("Warm preparation Todo is not open under the destination agent") - guard = cli("quota", "should-run", "--goal-id", goal, "--agent-id", agent, - "--available-capability", "shell", "--available-capability", "filesystem_write") - channel = (guard.get("interaction_contract") or {}).get("agent_channel") or {} - selected = guard.get("selected_todo") or {} - pending_review = guard.get("autonomous_replan_obligation") or {} - if (guard.get("should_run") is not True or channel.get("delivery_allowed") is not True or - not (selected.get("todo_id") == todo or pending_review.get("required") is True)): - raise ValueError("Warm continuation has no actionable current preparation or review") - # The guard can legitimately require a review after this Todo completes. - # That review cannot make an older open/owned preparation snapshot current. - # Bracket it with the existing canonical read revision; do not invent a - # ready receipt, retry budget, or permission from imported history. - after = cli("todo", "list", "--goal-id", goal, "--todo-id", todo) - revision = (before.get("authority_read") or {}).get("provider_revision") - current = after["todo"] - if (not isinstance(revision, str) or not revision - or (after.get("authority_read") or {}).get("provider_revision") != revision - or current.get("status") != "open" or current.get("claimed_by") != agent): - raise ValueError("Warm preparation state changed during current readiness readback; inspect this receiver") - _phase_read_timeout(declaration, 60) - return {"authority_read": after["authority_read"], "preparation_todo_id": todo, - "selected_todo_id": selected.get("todo_id"), - "current_replan_obligation_id": pending_review.get("obligation_id"), - "current_replan_required": pending_review.get("required") is True} - - -def finish_preparation(env: dict[str, str]) -> dict: - """Restore the declared cadence only after the existing joint replan gate.""" - from benchmark.runtime.checkpoint_continuation import restored_capture_gate - from benchmark.runtime.worker import loopx_command - import subprocess - declaration = json.loads(Path(env["LOOPX_WARM_START"]).read_bytes()) - if any(env[key] != declaration[field] for key, field in - (("LOOPX_GOAL_ID", "goal_id"), ("LOOPX_AGENT_ID", "agent_id"))): - raise ValueError("Warm preparation destination lineage differs") - basis = {"checkpoint_sha256": declaration["sha256"], - "goal_id": declaration["goal_id"], "agent_id": declaration["agent_id"], - "normal_cadence": declaration["normal_cadence"]} - receipt = Path(env["LOOPX_WARM_START"]).with_name("warm-preparation-complete.json") - if receipt.exists(): - value = json.loads(receipt.read_bytes()) - if any(value.get(key) != expected for key, expected in basis.items()): - raise ValueError("Warm preparation receipt refers to another checkpoint") - return value - gate = restored_capture_gate(env) - if not gate.get("capture_allowed") or gate.get("checkpoint_sha256") != declaration["sha256"]: - return {"preparation_complete": False} - command = [*loopx_command(env), "configure-goal", "--goal-id", env["LOOPX_GOAL_ID"], - "--execution-replan-after-turns", str(declaration["normal_cadence"]), "--execute"] - raw = subprocess.check_output(command, env=env, cwd=env["LOOPX_PROJECT"], timeout=60, text=True) - value = json.loads(raw) - if value.get("ok") is not True: - raise ValueError("Could not restore the declared normal cadence after checkpoint preparation") - result = {**basis, "preparation_complete": True, "review_turn_id": gate["review_turn_id"]} - with receipt.open("x") as stream: - json.dump(result, stream) - return result - - def main(): import argparse parser = argparse.ArgumentParser(description=__doc__) @@ -499,7 +452,7 @@ def main(): declaration = json.loads(args.declaration.read_bytes()) env = dict(os.environ) if args.action == "workspace": - restore_workspace(declaration, Path(env["LOOPX_PROJECT"])) + restore_worker(declaration, Path(env["LOOPX_PROJECT"])) else: print(json.dumps(prepare_context(env, declaration))) diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 27f431c1c1..105d77bd0c 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -567,9 +567,6 @@ def run_once(env: dict[str, str]) -> dict: execution, timeout_seconds=(remaining - PHASE_WORK_RESERVE_SECONDS if execution.timeout_seconds is None else min(execution.timeout_seconds, remaining - PHASE_WORK_RESERVE_SECONDS)) ) - if env.get("LOOPX_WARM_START"): - from benchmark.runtime.sforge_warm_start import finish_preparation - receipt["warm_preparation"] = finish_preparation(env) prepare_codex_home( home, execution=execution, diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py index 60e0d9b1eb..7975fe6d17 100644 --- a/benchmark/tests/test_native_checkpoint.py +++ b/benchmark/tests/test_native_checkpoint.py @@ -305,7 +305,7 @@ def test_real_native_workspace_readback_rejects_foreign_code_basis(tmp_path): @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"), reason="Requires an explicit real native Host binary; only a local static provider is used") @pytest.mark.parametrize("case", ["resume", "scope_change", "profile_change", "joint", "binding_reply_loss", - "fork_reply_loss", "fork_missing", "fork_ambiguous", "warm_preparation", + "fork_reply_loss", "fork_missing", "fork_ambiguous", "model_transition", "provider_transport", "joint_provider_transport"]) def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, capsys, monkeypatch, case, request): """Real Host/CLI/Goal/session/artifact owners; deterministic synthetic work. @@ -343,7 +343,6 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps research.write_text("candidate 3 was rejected by the original synthetic check\n") source_goal, source_agent = "source-research", "source-researcher" destination_goal, destination_agent = "branch-research", "branch-researcher" - from benchmark.runtime.sforge_warm_start import _PREPARATION_TASK, _read_preparation def register(name, agent): registry = tmp_path / name / "registry.json" @@ -358,9 +357,8 @@ def register(name, agent): capsys.readouterr() assert main([*prefix, "todo", "add", "--goal-id", name, "--role", "agent", "--claimed-by", agent, "--task-class", "advancement_task", "--required-capability", "shell", - *(["--operation-id", "checkpoint-preparation"] if case == "warm_preparation" and name == destination_goal else []), - "--text", _PREPARATION_TASK if case == "warm_preparation" and name == destination_goal else - "Verify the retained candidate and preserve applicable negative evidence."]) == 0 + *([]), + "--text", "Verify the retained candidate and preserve applicable negative evidence."]) == 0 added = json.loads(capsys.readouterr().out) return registry, runtime, added["todo_id"] source_registry, source_runtime, source_todo = register(source_goal, source_agent) @@ -369,10 +367,9 @@ def register(name, agent): turn_id = "checkpoint-continuation-fixture" binding = ["--goal-id", destination_goal, "--agent-id", destination_agent, "--todo-id", destination_todo, "--turn-instance-id", turn_id] - if case != "warm_preparation": - assert main([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) == 0 - guard = json.loads(capsys.readouterr().out) - assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True + assert main([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) == 0 + guard = json.loads(capsys.readouterr().out) + assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True from loopx.hosts.research_checkpoint import canonical_context, research_context, adopt_todo_context, _encoded source_prefix = ["--format", "json", "--registry", str(source_registry), "--runtime-root", str(source_runtime)] def invoke(arguments): @@ -413,30 +410,17 @@ def invoke(arguments): def response(request, index, message): requests.append(request) - if case == "warm_preparation" and index == 1: - message["content"][0]["text"] = "The previous Host deadline has passed; the prior phase was closed." if index == 2: - current_command = [sys.executable, "-m", "loopx.cli", *prefix, "todo", "list", - "--goal-id", destination_goal, "--todo-id", destination_todo] - # Read the actual Todo but emit only its task text. Whole authority - # envelopes can hide the adjacent retained files in native output - # truncation, which is unrelated to continuation admission. - read_current = (shlex.join([sys.executable, "-c", - "import json, subprocess; print(json.loads(subprocess.check_output(" - + repr(current_command) + ", text=True))['todo']['text'])"]) + " && " - if case == "warm_preparation" else "") return {"type": "function_call", "id": "fc_retained", "call_id": "call_retained", "name": "exec_command", "arguments": json.dumps({ - "cmd": read_current + "cat candidate.txt research.txt " + retained_context_path + joint_command, + "cmd": "cat candidate.txt research.txt " + retained_context_path + joint_command, "yield_time_ms": 10000, "max_output_tokens": 6000})} if index == 3: message["content"][0]["text"] = "Retained candidate 7 verified; candidate 3 remains rejected." return message binary = os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"] - # The warm receiver uses the benchmark's execution profile inside its - # isolated worker. Other cases retain their original sandbox contract. - sandbox = ("danger-full-access" if provider_transport or case in {"warm_preparation", "model_transition"} else + sandbox = ("danger-full-access" if provider_transport or case == "model_transition" else "workspace-write" if joint else "read-only") execution = Execution(context="resume", sandbox=sandbox) with ExitStack() as receiver_transports, real_native_source(tmp_path / "native", binary, workspace, close_source=True, @@ -444,14 +428,6 @@ def response(request, index, message): env.update(LOOPX_REGISTRY=str(source_registry), LOOPX_RUNTIME_ROOT=str(source_runtime), LOOPX_GOAL_ID=source_goal, LOOPX_AGENT_ID=source_agent, MODEL_NAME="fixture", REASONING_EFFORT="high", LOOPX_CODEX_SANDBOX=execution.sandbox) - if case == "warm_preparation": - executable = tmp_path / "receiver-cli" - executable.write_text("#!/bin/sh\nexec " + shlex.quote(sys.executable) + ' -m loopx.cli "$@"\n') - executable.chmod(0o700) - skills = tmp_path / "shared-skills" - skills.mkdir() - env.update(LOOPX_CLI=str(executable), LOOPX_SHARED_SKILLS=str(skills), - PYTHONPATH=str(Path(__file__).resolve().parents[2]) + os.pathsep + env.get("PYTHONPATH", "")) source_wake = BenchmarkSessionWake(env, execution, {}) source_wake.admission.accept_result(lambda: _store_codex_cli_session(source_runtime, lineage={"goal_id": source_goal, "agent_id": source_agent}, session_scope="agent", @@ -545,7 +521,6 @@ def retained_source(): model_config_args = ["--destination-model-config", str(target_config)] if provider_transport: from tests.control_plane.codex_checkpoint_fixture import local_inference_provider - from benchmark.runtime.sforge_warm_start import admit_warm_session def receiver_response(body, index, message): calls.append("/v1/responses") @@ -559,9 +534,6 @@ def receiver_response(body, index, message): target_config.write_text(config.read_text().replace(source_url, receiver_url)) env["OPENAI_BASE_URL"] = receiver_url admission_options = {"destination_config": target_config, "provider_base_url": receiver_url} - transport_declaration = {"root": str(unit), "sha256": digest, "config": str(config), - "destination_config": str(target_config), "provider_base_url": receiver_url, - "deadline_epoch": int(time.time()) + 600} lifecycle_registry = tmp_path / "lifecycle-only" / "registry.json" assert main(["--format", "json", "--registry", str(lifecycle_registry), "--runtime-root", str(tmp_path / "lifecycle-only/runtime"), "project", "register", @@ -666,10 +638,11 @@ def lose_binding_reply(self, *args): assert len(calls) == 1 if provider_transport: with pytest.raises(ValueError, match="only model and reasoning effort"): - admit_warm_session(env, {key: value for key, value in transport_declaration.items() - if key != "provider_base_url"}) + admit_native_checkpoint_session(env, unit, digest, config, sandbox=execution.sandbox, + timeout=90, destination_config=target_config) assert not Path(env["CODEX_HOME"]).exists() and len(calls) == 1 - admitted = admit_warm_session(env, transport_declaration) + admitted = admit_native_checkpoint_session(env, unit, digest, config, + sandbox=execution.sandbox, timeout=90, **admission_options) else: admitted_cli = subprocess.run([sys.executable, "-m", "loopx.hosts.codex_checkpoint", "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), @@ -697,7 +670,8 @@ def forbidden_fork(*args, **kwargs): # Transport reply loss leaves the atomic source-bound binding as the # original receipt. An exact CLI retry must also recover it. if provider_transport: - assert admit_warm_session(env, transport_declaration) == admitted + assert admit_native_checkpoint_session(env, unit, digest, config, + sandbox=execution.sandbox, timeout=90, **admission_options) == admitted else: retry_cli = subprocess.run(admitted_cli.args, env=env, text=True, capture_output=True, timeout=40) assert retry_cli.returncode == 0, retry_cli.stderr @@ -731,50 +705,26 @@ def forbidden_fork(*args, **kwargs): # Both participants meet at the real destination continuation owner. # No source claim/lease is replayed; adoption alone still starts no work. - if case == "warm_preparation": - declaration = tmp_path / "warm-start.json" - payload = {"root": str(unit), "sha256": digest, "config": str(config), - "goal_id": destination_goal, "agent_id": destination_agent, - "source_todo_id": source_todo, "normal_cadence": 6, - "deadline_epoch": int(time.time()) + 600} - declaration.write_text(json.dumps(payload)) - prepared_cli = subprocess.run([sys.executable, "-m", "benchmark.runtime.sforge_warm_start", - "context", "--declaration", str(declaration)], env=env, cwd=workspace, - text=True, capture_output=True, timeout=90) - assert prepared_cli.returncode == 0, prepared_cli.stderr - prepared = json.loads(prepared_cli.stdout) - assert prepared["preparation_todo_id"] == destination_todo - assert prepared["preparation_actionable"] and prepared["model_calls"] == 0 - assert prepared["research_continuation_qualified"] is False - context_result = prepared["context"] - _read_preparation(lambda *args: invoke([*prefix, *args]), destination_goal, - destination_agent, destination_todo, payload) - assert json.loads(declaration.read_text()) == payload - guard = invoke([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding, - "--available-capability", "shell", "--available-capability", "filesystem_write"]) - assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] - assert len(calls) == 1 - else: - lease = invoke([*prefix, "task-lease", "acquire", "--goal-id", destination_goal, - "--todo-id", destination_todo, "--owner", destination_agent, "--idempotency-key", "branch-context"]) - current = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo]) - context_command = [sys.executable, "-m", "loopx.hosts.research_checkpoint", - "--root", str(unit), "--expected-sha256", digest, "--source-todo-id", source_todo, - "--todo-id", destination_todo, "--operation-id", "branch-context", - "--expected-revision", current["authority_read"]["provider_revision"], - "--lease-key", "branch-context", "--lease-version", str(lease["lease"]["version"])] - assert not explore_result_log_path(runtime, destination_goal).exists(), "session admission must not adopt knowledge" - adopted_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60) - assert adopted_cli.returncode == 0, adopted_cli.stderr - context_result = json.loads(adopted_cli.stdout) - # Retrying the exact CLI after a lost result reuses both original - # receipts; it must not fork, charge, edit the Todo or run a model. - replay_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60) - assert replay_cli.returncode == 0, replay_cli.stderr - replay_result = json.loads(replay_cli.stdout) - assert replay_result["continuation"]["adoption"]["status"] == "replayed" - assert replay_result["continuation"]["provider_revision"] == context_result["continuation"]["provider_revision"] - assert len(calls) == 1 + lease = invoke([*prefix, "task-lease", "acquire", "--goal-id", destination_goal, + "--todo-id", destination_todo, "--owner", destination_agent, "--idempotency-key", "branch-context"]) + current = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo]) + context_command = [sys.executable, "-m", "loopx.hosts.research_checkpoint", + "--root", str(unit), "--expected-sha256", digest, "--source-todo-id", source_todo, + "--todo-id", destination_todo, "--operation-id", "branch-context", + "--expected-revision", current["authority_read"]["provider_revision"], + "--lease-key", "branch-context", "--lease-version", str(lease["lease"]["version"])] + assert not explore_result_log_path(runtime, destination_goal).exists(), "session admission must not adopt knowledge" + adopted_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60) + assert adopted_cli.returncode == 0, adopted_cli.stderr + context_result = json.loads(adopted_cli.stdout) + # Retrying the exact CLI after a lost result reuses both original + # receipts; it must not fork, charge, edit the Todo or run a model. + replay_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60) + assert replay_cli.returncode == 0, replay_cli.stderr + replay_result = json.loads(replay_cli.stdout) + assert replay_result["continuation"]["adoption"]["status"] == "replayed" + assert replay_result["continuation"]["provider_revision"] == context_result["continuation"]["provider_revision"] + assert len(calls) == 1 retained_context_path = str(Path(context_result["artifact"]).relative_to(workspace)) assert Path(context_result["artifact"]).read_bytes() == context_raw and len(calls) == 1 adopted_todo = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo]) @@ -849,8 +799,7 @@ def forbidden_fork(*args, **kwargs): if case == "model_transition": assert requests[1]["model"] == "fixture-next" assert requests[1]["reasoning"]["effort"] == "max" - retained_message = ("previous Host deadline has passed" if case == "warm_preparation" - else "Synthetic source result.") + retained_message = ("Synthetic source result.") assert retained_message in json.dumps(requests[1]) assert destination_goal in json.dumps(requests[1]) and destination_todo in json.dumps(requests[1]) assert "retained candidate: 7" in json.dumps(requests[2]) @@ -891,10 +840,6 @@ def forbidden_fork(*args, **kwargs): for secret in (source_token, fresh["token"], source_judge.state.judge.admin_secret, branch_judge.state.judge.admin_secret): assert secret not in json.dumps(requests) - if case == "warm_preparation": - assert "previous Host deadline has passed" in json.dumps(requests[1]) - assert _PREPARATION_TASK in json.dumps(requests[2]) - assert "Earlier Host deadlines" not in json.dumps(requests[2]) assert "Retained candidate 7 verified" in result.stdout retained = native_session_checkpoint(env, admitted["session_id"], timeout=20) assert retained["metadata"]["turn_ids"][:-1] == native["metadata"]["turn_ids"] @@ -907,12 +852,11 @@ def forbidden_fork(*args, **kwargs): sandbox=(execution).sandbox, timeout=30, **admission_options) assert load_codex_cli_session(runtime, lineage={"goal_id": destination_goal, "agent_id": destination_agent}, session_scope="agent") == after_work - if case != "warm_preparation": - with pytest.raises(ValueError, match="native history or configuration moved"): - adopt_todo_context(env, unit, digest, source_todo_id=source_todo, - todo_id=destination_todo, operation_id="must-not-adopt-later-session", - expected_revision=adopted_todo["authority_read"]["provider_revision"], - lease_key="branch-context", lease_version=lease["lease"]["version"]) + with pytest.raises(ValueError, match="native history or configuration moved"): + adopt_todo_context(env, unit, digest, source_todo_id=source_todo, + todo_id=destination_todo, operation_id="must-not-adopt-later-session", + expected_revision=adopted_todo["authority_read"]["provider_revision"], + lease_key="branch-context", lease_version=lease["lease"]["version"]) assert len(calls) == 3 assert invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo])["todo"] == adopted_todo["todo"] diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index 7e56b190c4..e000dcc043 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -2,6 +2,7 @@ import asyncio from dataclasses import replace import hashlib +from pathlib import Path import json import shutil import time @@ -68,7 +69,7 @@ def test_native_skills_backup_never_overwrites_an_earlier_attempt(tmp_path): @pytest.fixture -def point(tmp_path, request): +def point(tmp_path, request, monkeypatch): workspace = tmp_path / "workspace" workspace.mkdir() (workspace / "candidate").write_text("retained candidate") @@ -84,12 +85,26 @@ def point(tmp_path, request): config.write_text('model="original"\nmodel_reasoning_effort="high"\n[features]\ngoals = false\n') context = tmp_path / "context.json" context.write_text(json.dumps({"snapshot": {"todos": [{"todo_id": "source-todo"}]}})) + runtime = tmp_path / "runtime" + runtime.mkdir() + (runtime / "retained.json").write_text('{"original":"state"}') + registry = tmp_path / "registry.json" + registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{"id": "source-goal", + "repo": str(workspace), "coordination": {"registered_agents": ["source-agent"]}}]})) + monkeypatch.setattr(warm, "_REGISTRY", str(registry)) + monkeypatch.setattr(warm, "_LOOPX_RUNTIME", str(runtime)) components = [{"id": "workspace", "source_path": str(workspace)}, - {"id": "context", "source_path": str(context)}] + {"id": "context", "source_path": str(context)}, + {"id": "runtime", "source_path": str(runtime)}, + {"id": "registry", "source_path": str(registry)}] metadata = { "native_session": {"guest_workspace": str(workspace), "guest_home": warm._CODEX_HOME, "config_sha256": hashlib.sha256(config.read_bytes()).hexdigest()}, - "loopx_state": {"goal_id": "source-goal", "agent_id": "source-agent"}, + "loopx_state": {"goal_id": "source-goal", "agent_id": "source-agent", + "registry_sha256": hashlib.sha256(registry.read_bytes()).hexdigest(), + "registry_component_id": "registry", "registry_member_path": "", + "source_inventory": [{"key": "runtime_root", "component_id": "runtime", "source_path": str(runtime)}, + {"key": "configuration_source_registry", "component_id": "registry", "source_path": str(registry)}]}, "worker_file_snapshot": {"components": components[:1]}, "research_context": {"component_id": "context", "member_path": "", "sha256": hashlib.sha256(context.read_bytes()).hexdigest()}, @@ -98,12 +113,12 @@ def point(tmp_path, request): unit = tmp_path / "point" sealed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components, "point_metadata": metadata, "expected_source_sha256": observed["source_sha256"], "destination": str(unit)}) - start = warm.WarmStart(unit, sealed["sha256"], config, "new-goal", "new-agent", "source-todo", int(time.time()) + 600) + start = warm.WarmStart(unit, sealed["sha256"], config, "source-goal", "source-agent", "source-todo", int(time.time()) + 600) return start, workspace @pytest.mark.parametrize("change", [ - {"goal_id": "source-goal"}, {"agent_id": "source-agent"}, {"source_todo_id": "missing"}, + {"goal_id": "foreign-goal"}, {"agent_id": "foreign-agent"}, {"source_todo_id": "missing"}, {"deadline_epoch": 1}, {"deadline_epoch": True}, {"normal_cadence": 0}, {"normal_cadence": True}, {"fresh_replan": "false"}, ]) @@ -201,7 +216,7 @@ async def upload_file(self, source, target): async def exec(self, command, **kwargs): commands.append(command) return SimpleNamespace(return_code=0) - assert asyncio.run(start.stage(Environment(), str(workspace))) == [] + assert asyncio.run(start.stage(Environment(), str(workspace))) == [warm._LOOPX_RUNTIME, warm._REGISTRY] assert "chmod -R" not in commands[0] effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(control / "restored-point"), "expected_sha256": start.sha256}) @@ -296,31 +311,6 @@ def __getattr__(self, name): effective_config=start.config.read_bytes(), provider_base_url="http://gateway:4002/v1")) -def test_normal_cadence_requires_accepted_current_replan_and_is_applied_once(tmp_path, monkeypatch): - from benchmark.runtime import checkpoint_continuation, worker - declaration = tmp_path / "warm.json" - declaration.write_text(json.dumps({"sha256": "a" * 64, "normal_cadence": 6, - "goal_id": "current", "agent_id": "receiver"})) - env = {"LOOPX_WARM_START": str(declaration), "LOOPX_GOAL_ID": "current", - "LOOPX_AGENT_ID": "receiver", "LOOPX_PROJECT": str(tmp_path)} - monkeypatch.setattr(worker, "loopx_command", lambda env: ["loopx"]) - gate = {"capture_allowed": False, "checkpoint_sha256": "a" * 64} - monkeypatch.setattr(checkpoint_continuation, "restored_capture_gate", lambda env: gate) - calls = [] - def configure(command, **kwargs): - calls.append(command) - return '{"ok": true}' - monkeypatch.setattr("subprocess.check_output", configure) - assert warm.finish_preparation(env) == {"preparation_complete": False} - assert not calls - gate.update(capture_allowed=True, review_turn_id="accepted-current-replan") - assert warm.finish_preparation(env)["preparation_complete"] - assert warm.finish_preparation(env)["normal_cadence"] == 6 - assert len(calls) == 1 - with pytest.raises(ValueError, match="lineage"): - warm.finish_preparation({**env, "LOOPX_AGENT_ID": "another-agent"}) - - def test_guest_helper_does_not_require_host_harbor_package(): import subprocess import sys @@ -523,7 +513,7 @@ async def upload_file(self, source, target): shutil.copyfile(source, target) async def exec(self, command, **kwargs): return SimpleNamespace(return_code=0) - assert asyncio.run(start.stage(Environment(), str(workspace))) == [str(research)] + assert asyncio.run(start.stage(Environment(), str(workspace))) == [str(research), warm._LOOPX_RUNTIME, warm._REGISTRY] declaration = json.loads((control / "warm-start.json").read_bytes()) assert declaration["research_component_ids"] == ["experiments"] @@ -676,7 +666,12 @@ def parent_workspace_point(point, tmp_path, request): elif variant == "noncanonical": metadata["worker_file_snapshot"]["components"] = [ {"id": "workspace", "source_path": str(parent) + "/"}] - components = [component, manifest["components"][1]] + registry = Path(warm._REGISTRY) + data = json.loads(registry.read_bytes()) + data["goals"][0]["repo"] = str(workspace) + registry.write_text(json.dumps(data)) + metadata["loopx_state"]["registry_sha256"] = hashlib.sha256(registry.read_bytes()).hexdigest() + components = [component, *manifest["components"][1:]] observed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) unit = tmp_path / "parent-point" sealed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components, diff --git a/docs/reference/research-checkpoint.md b/docs/reference/research-checkpoint.md index c457d0e6df..30641eda14 100644 --- a/docs/reference/research-checkpoint.md +++ b/docs/reference/research-checkpoint.md @@ -140,7 +140,35 @@ current guard offers original-Turn closeout, include `checkpoint:` in `agent_vision.path_delta.evidence_refs` in that writeback and finish its settlement. If the Turn has ended, carry the unresolved review to the next admitted replan; do not replay or rebind historical Turns. -Normal research cadence resumes only after the existing recovery gate passes. +Recapture remains closed until the existing recovery gate passes; the retained work cadence is unchanged. + +## Complete isolated benchmark replacement + +Benchmark warm replacement continues the original Goal/Agent and its complete +captured LoopX runtime, including canonical SQLite/File state, Todos, +dependencies, settlement/quota records and Vision/Explore history. It uses the +backup inventory and artifact witness, checks canonical revision/history parity, +then compares and replaces only the captured native session binding through the +existing Goal/session owners. Current guards and lease owners still decide +execution and interrupted-Turn recovery; settled effects are never replayed. + +This is distinct from the cross-Goal research import above. Warm replacement +requires a stopped source and fresh isolated receiver at the qualified fixed +paths. Existing receiver registry/state, divergent history or a changed native +binding rejects. Preserve the failed attempt; do not overwrite it. The launcher +uses the original Goal/Agent, validates the retained treatment without changing +it, and creates no preparation Todo or temporary cadence. Research review before +recapture remains required; checkpoint frequency, native opportunities, cooldown +and the cumulative phase clock do not increase. + +See [warm worker composition](../../benchmark/README.md#explicit-warm-worker-composition-qualification-in-progress). +The packaged frontend restore journey and sustained real-model benefit remain +unqualified. Omit warm inputs to run a separate cold trial; stopping the receiver +does not erase the sealed point or the original state. + +热恢复接续完整业务状态，跨 Goal 的 `adopt-context` 只导入研究证据。 +热恢复不再以新 preparation Todo 替换原任务队列，也不临时修改重规划频率。 +恢复副本不能覆盖活动实例，原结算、当前权限检查及预算/停止条件继续有效。 ## Default bank rotation diff --git a/loopx/control_plane/turn_driver/codex_agent_session.py b/loopx/control_plane/turn_driver/codex_agent_session.py index a88091595c..f96f7fbdf8 100644 --- a/loopx/control_plane/turn_driver/codex_agent_session.py +++ b/loopx/control_plane/turn_driver/codex_agent_session.py @@ -180,12 +180,14 @@ def prepare_checkpoint_refresh(self, review_basis: dict | None, attempt: Path, s def _bind_verified_fork(self, session_id: str, verified_profile_digest: str, checkpoint_artifact_sha256: str, checkpoint_artifact_root: str, - *, model_transition: dict[str, str] | None = None) -> None: + *, model_transition: dict[str, str] | None = None, + expected_binding: dict[str, Any] | None = None) -> None: """Bind a controller-verified native fork through the existing session owner. This admits the conversation binding only. It does not claim work, restore research authority or start a model. Unknown result delivery - requires readback; it cannot replace an existing binding. + requires readback. An isolated full-state replacement may compare and + replace only its independently verified captured binding. """ if self.session_id is not None: raise ValueError("checkpoint session adoption requires an absent destination binding") @@ -198,7 +200,8 @@ def _bind_verified_fork(self, session_id: str, verified_profile_digest: str, session_id=session_id, goal_ref=self.goal_ref, session_profile_digest=self.digest, - require_absent=True, + require_absent=expected_binding is None, + **({"expected_document": expected_binding} if expected_binding is not None else {}), checkpoint_artifact_sha256=checkpoint_artifact_sha256, checkpoint_artifact_root=checkpoint_artifact_root, checkpoint_model_transition=model_transition, diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index ccf93a2eea..c033ccf537 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -525,7 +525,8 @@ def request(method, params): def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sha256: str, config_path: Path, *, sandbox: str, timeout: float, destination_config: Path | None = None, - provider_base_url: str | None = None) -> dict[str, Any]: + provider_base_url: str | None = None, + restore_loopx_state: bool = False) -> dict[str, Any]: """Explicit controller admission of a verified fork into a destination Goal. Existing native fork, Goal lifetime and session owners retain their rules. @@ -546,6 +547,8 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh if not math.isfinite(timeout) or timeout <= 0: raise ValueError("checkpoint session admission timeout must be finite and positive") + if type(restore_loopx_state) is not bool: + raise ValueError("full LoopX state restoration requires an explicit boolean") if provider_base_url is not None and provider_base_url != env.get("OPENAI_BASE_URL"): raise ValueError("checkpoint provider transport differs from the current launch") deadline = time.monotonic() + timeout @@ -587,15 +590,32 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh source = metadata.get("loopx_state", {}) lineage = {"goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"]} if (not source.get("goal_id") or not source.get("agent_id") or - any(source[key] == lineage[key] for key in lineage)): - raise ValueError("checkpoint session admission requires distinct destination Goal and Agent identities") + (not all(source[key] == lineage[key] for key in lineage) if restore_loopx_state else + any(source[key] == lineage[key] for key in lineage))): + raise ValueError("full LoopX restoration requires the captured Goal and Agent identities" if restore_loopx_state else + "checkpoint session admission requires distinct destination Goal and Agent identities") registry = Path(env["LOOPX_REGISTRY"]) # Use the current governed Goal path. The lifecycle-only source-session # profile cannot run governed Turns and is rejected by the registry owner. goal = load_goal_from_registry(registry, lineage["goal_id"]) require_registered_agent_id(registry_path=registry, goal_id=lineage["goal_id"], agent_id=lineage["agent_id"]) - if select_codex_cli_session(Path(env["LOOPX_RUNTIME_ROOT"]), lineage=lineage, - session_scope="agent") is not None: + existing = select_codex_cli_session(Path(env["LOOPX_RUNTIME_ROOT"]), lineage=lineage, session_scope="agent") + expected_binding = None + if restore_loopx_state and (existing is None or existing.get("checkpoint_artifact_sha256") != expected_sha256): + from loopx.hosts.research_checkpoint import verify_restored_loopx_state + from loopx.control_plane.turn_driver.codex_sessions import _session_path + + runtime_source = verify_restored_loopx_state(env, unit, metadata, + deadline=time.time() + max(.1, deadline - time.monotonic())) + path = _session_path(Path(env["LOOPX_RUNTIME_ROOT"]), lineage, "agent") + sealed = runtime_source / path.relative_to(Path(env["LOOPX_RUNTIME_ROOT"])) + if sealed.resolve(strict=True) != sealed: + raise ValueError("captured LoopX session binding is redirected") + expected_binding = json.loads(sealed.read_bytes()) + if (existing != expected_binding or existing.get("session_id") != metadata["native_session"]["session_id"] + or existing.get("session_profile_digest") != source_profile): + raise ValueError("restored LoopX session differs from its sealed source binding") + if existing is not None and expected_binding is None: # The original session binding is also the admission receipt. Only a # source-bound committed admission can recover a lost controller reply; # an unrelated binding or a partial native fork is never replaced. @@ -639,7 +659,10 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh **({"destination_config": destination_config} if destination_config is not None else {}), **({"provider_base_url": provider_base_url} if provider_base_url is not None else {})) receipt: dict[str, Any] = {} - wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt=receipt) + wake = CodexAgentSession(env, context="fresh" if expected_binding is not None else "resume", + sandbox=sandbox, receipt=receipt) + if expected_binding is not None and expected_binding.get("goal_ref") != wake.goal_ref: + raise ValueError("restored Goal lifetime differs from its captured session") # Hold the existing registry owner's fence through the session write. This # does not mint a lease, task or work admission from historical state. with project_registry_transaction(registry, operation="checkpoint_session_binding", agent_id=lineage["agent_id"]): @@ -651,6 +674,7 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh # settings above. Bind the explicitly verified destination profile; # ordinary session resumes still require an exact profile match. wake._bind_verified_fork(result["session_id"], target_profile, expected_sha256, str(unit), + **({"expected_binding": expected_binding} if expected_binding is not None else {}), **({"model_transition": transition} if transition is not None else {})) return {**result, **({"model_transition": transition} if transition is not None else {}), "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"], diff --git a/loopx/hosts/research_checkpoint.py b/loopx/hosts/research_checkpoint.py index 5531694bbd..c4a9fdde88 100644 --- a/loopx/hosts/research_checkpoint.py +++ b/loopx/hosts/research_checkpoint.py @@ -1,9 +1,9 @@ """Research checkpoint context IO through existing canonical Todo owners. -Captured records are historical input, never a replacement authority store. -Only the existing explicit continuation owner can adopt a note on an already -claimed destination Todo. Verified Explore knowledge may seed a fresh graph; -historical execution observations, effects and vision authority are not revived. +Research import adopts historical input on a claimed destination Todo; verified +Explore knowledge may seed a fresh graph. Explicit isolated full-state restore +instead verifies the complete retained business state without rewriting it. +Neither path grants a work admission or settles historical execution effects. """ from __future__ import annotations @@ -123,6 +123,43 @@ def _encoded(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() +def verify_restored_loopx_state(env: dict[str, str], unit: Path, metadata: dict, *, deadline: float) -> Path: + """Read back complete business state through its existing canonical owners. + + The caller has verified the sealed artifact and materialized a quiescent, + isolated replacement at its original paths. Equality preserves receipts, + Todo identities/dependencies, acceptance, Explore and Vision history. It + does not settle an interrupted Turn or grant a lease/model invocation. + """ + state = metadata["loopx_state"] + if any(env[key] != state[field] for key, field in + (("LOOPX_GOAL_ID", "goal_id"), ("LOOPX_AGENT_ID", "agent_id"))): + raise ValueError("full LoopX restore requires the captured Goal and Agent") + registry = Path(env["LOOPX_REGISTRY"]) + if registry.resolve(strict=True) != registry or hashlib.sha256(registry.read_bytes()).hexdigest() != state["registry_sha256"]: + raise ValueError("restored LoopX registry differs from its sealed source") + roots = [entry for entry in state["source_inventory"] if entry["key"] == "runtime_root"] + if len(roots) != 1 or roots[0]["source_path"] != env["LOOPX_RUNTIME_ROOT"]: + raise ValueError("restored LoopX runtime differs from its captured route") + context = metadata["research_context"] + path = unit / "components" / context["component_id"] / context["member_path"] + raw = path.read_bytes() + if hashlib.sha256(raw).hexdigest() != context["sha256"]: + raise ValueError("restored LoopX canonical context digest differs") + expected = json.loads(raw) + actual = research_context(env, deadline=deadline) + # The canonical cursor binds the complete business revision. Derived Todo + # resume/readiness observations may legitimately change as time elapses; + # they must be evaluated now, rather than frozen to their capture time. + identity = ("store_identity", "provider_revision", "cursor", "todo_ids") + if (actual["goal_id"] != expected["goal_id"] + or any(actual["snapshot"].get(key) != expected["snapshot"].get(key) for key in identity) + or actual["research_evidence"] != expected["research_evidence"] + or actual["source_files"] != expected["source_files"]): + raise ValueError("restored LoopX canonical state or research history differs from its sealed source") + return unit / "components" / roots[0]["component_id"] + + def _handoff(env, payload): from loopx.agent_registry import registered_agent_ids_from_registry request = {"runtime_root": env["LOOPX_RUNTIME_ROOT"], "goal_id": env["LOOPX_GOAL_ID"], diff --git a/tests/control_plane/test_full_loopx_checkpoint_restore.py b/tests/control_plane/test_full_loopx_checkpoint_restore.py new file mode 100644 index 0000000000..1a8e66dd57 --- /dev/null +++ b/tests/control_plane/test_full_loopx_checkpoint_restore.py @@ -0,0 +1,231 @@ +"""A warm replacement retains work, not just research prose in a new Goal.""" +from contextlib import ExitStack +from dataclasses import replace +import hashlib +import json +import os +from pathlib import Path +import shutil +import sys +import time + +import pytest + +from benchmark.runtime import sforge_warm_start as warm +from canonical_authority_fixture import isolate_sqlite_runtime +from loopx.cli import main +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_agent_session import CodexAgentSession +from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session, load_codex_cli_session +from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session, native_session_checkpoint +from loopx.hosts.research_checkpoint import _loopx_state_components, research_context, verify_restored_loopx_state + + +@pytest.fixture +def retained_point(tmp_path, monkeypatch, capsys, request): + isolate_sqlite_runtime(tmp_path, monkeypatch) + workspace, runtime, registry = tmp_path / "workspace", tmp_path / "runtime", tmp_path / "control/registry.json" + workspace.mkdir() + registry.parent.mkdir() + monkeypatch.chdir(workspace) + + def cli(*args): + code = main(["--format", "json", "--registry", str(registry), "--runtime-root", str(runtime), *args]) + result = json.loads(capsys.readouterr().out) + assert code == 0, result + return result + + variant = getattr(request, "param", ("sqlite", False)) + provider, real_host = variant if isinstance(variant, tuple) else ("sqlite", variant) + if provider == "file": + defaults = tmp_path / "storage.json" + defaults.write_text(json.dumps({"schema_version": "loopx_goal_storage_defaults_v1", "new_goal_provider": "file", + "canonical_creation": True, "new_goal_handoff_mode": "hard_lease"})) + plan = cli("machine-config", "preview", "--namespace", "goal_storage", "--config-json", str(defaults)) + cli("machine-config", "apply", "--namespace", "goal_storage", "--config-json", str(defaults), + "--expected-plan-revision", plan["plan_revision"], "--execute") + cli("bootstrap", "--project", str(workspace), "--goal-id", "research", "--objective", "Improve verified outcomes.", + "--adapter-kind", "read_only_project_map_v0", "--adapter-status", "connected-read-only", + "--write-scope", "**", "--no-global-sync") + cli("configure-goal", "--goal-id", "research", "--registered-agent", "researcher", + "--agent-work-mode", "researcher=active", "--execution-replan-after-turns", "6", "--execute") + todo = cli("todo", "add", "--goal-id", "research", "--role", "agent", "--claimed-by", "researcher", + "--task-class", "advancement_task", "--priority", "P0", + "--text", "Continue complete experiments until acceptance or a real stop.")["todo_id"] + cli("todo", "add", "--goal-id", "research", "--role", "agent", "--claimed-by", "researcher", + "--task-class", "advancement_task", "--priority", "P2", "--text", "Qualify the final outcome independently.") + code = workspace / "candidate.txt" + code.write_text("retained candidate") + stack = ExitStack() + request.addfinalizer(stack.close) + if real_host: + binary = os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN") + if not binary: + pytest.skip("Requires explicit real native Host; inference is a deterministic local fixture") + from tests.control_plane.codex_checkpoint_fixture import real_native_source + native_env, source_session, _, calls = stack.enter_context(real_native_source( + tmp_path / "native", binary, workspace, close_source=True)) + home = Path(native_env["CODEX_HOME"]) + else: + home = tmp_path / "native-home" + home.mkdir() + (home / "config.toml").write_text('model="fixture"\nmodel_reasoning_effort="high"\n[features]\ngoals=false\n') + native_env, source_session, calls = {"CODEX_BIN": sys.executable, "CODEX_HOME": str(home)}, "source-session", [] + env = {**os.environ, **native_env, "LOOPX_PROJECT": str(workspace), "LOOPX_RUNTIME_ROOT": str(runtime), + "LOOPX_REGISTRY": str(registry), "LOOPX_GOAL_ID": "research", "LOOPX_AGENT_ID": "researcher", + "MODEL_NAME": "fixture", "REASONING_EFFORT": "high", "LOOPX_CODEX_SANDBOX": "read-only"} + wake = CodexAgentSession(env, context="fresh", sandbox="read-only", receipt={}) + wake.admission.accept_result(lambda: _store_codex_cli_session(runtime, lineage=wake.lineage, + session_scope="agent", session_id=source_session, goal_ref=wake.goal_ref, session_profile_digest=wake.digest)) + from loopx.capabilities.explore.result_log import append_explore_result_event, build_explore_node_event, explore_result_log_path + append_explore_result_event(explore_result_log_path(runtime, "research"), build_explore_node_event( + goal_id="research", node_id="rejected", title="Retained counterexample", status="dead_end")) + # Keep a real current lease in the point. Restoration does not mint or + # extend it; the ordinary task owner must inspect its current validity. + lease = cli("task-lease", "acquire", "--goal-id", "research", "--todo-id", todo, + "--owner", "researcher", "--idempotency-key", "original-work")["lease"] + actor = ["--goal-id", "research", "--agent-id", "researcher", "--todo-id", todo, + "--turn-instance-id", "source-work"] + cli("quota", "should-run", *actor, "--runtime-profile", "generic_cli", "--context-delivery", "off", + "--available-capability", "shell") + vision = tmp_path / "vision.json" + vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed", + "vision_patch": {"acceptance_summary": "Continue validated complete experiments until the original acceptance."}, + "path_delta": {"outcome": "continue", "prior_assumption": "Retain counterexamples while improving candidates.", + "observed_reality": "The first complete experiment retained a validated candidate.", + "retained": ["Complete experiments and original acceptance"], "evidence_refs": ["validation:fixture"]}})) + cli("refresh-state", *actor, "--classification", "validated_experiment", "--delivery-outcome", "outcome_progress", + "--delivery-batch-scale", "multi_surface", "--agent-vision-json", str(vision), + "--suppress-external-sinks", "--no-global-sync") + cli("quota", "spend-slot", *actor, "--source", "heartbeat", "--execute") + settled = read_heartbeat_settlement(runtime, goal_id="research", agent_id="researcher", todo_id=todo, + turn_instance_id="source-work", registry_path=registry) + assert settled.progress["state"] == "settled" + metadata = {"worker_file_snapshot": {"components": [{"id": "workspace", "source_path": str(workspace)}]}, + "session": {"session_id": source_session, "session_profile_digest": wake.digest}} + if real_host: + native = native_session_checkpoint(env, source_session, timeout=30) + components, metadata["native_session"] = native["components"], native["metadata"] + else: + components = [] + metadata["native_session"] = {"session_id": source_session, "guest_workspace": str(workspace), + "guest_home": str(home), "config_sha256": hashlib.sha256((home / "config.toml").read_bytes()).hexdigest()} + components = [*components, {"id": "workspace", "source_path": str(workspace)}] + point = tmp_path / "point" + components, metadata["loopx_state"] = _loopx_state_components(env, components, point) + context = research_context(env, deadline=time.time() + 60) + context_path = tmp_path / "context.json" + context_path.write_text(json.dumps(context)) + components.append({"id": "context", "source_path": str(context_path)}) + metadata["research_context"] = {"component_id": "context", "member_path": "", + "sha256": hashlib.sha256(context_path.read_bytes()).hexdigest()} + probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) + sealed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components, + "point_metadata": metadata, "expected_source_sha256": probe["source_sha256"], "destination": str(point)}) + config = tmp_path / "source-config.toml" + config.write_bytes((home / "config.toml").read_bytes()) + monkeypatch.setattr(warm, "_REGISTRY", str(registry)) + monkeypatch.setattr(warm, "_LOOPX_RUNTIME", str(runtime)) + monkeypatch.setattr(warm, "_CODEX_HOME", str(home)) + start = warm.WarmStart(point, sealed["sha256"], config, "research", "researcher", todo, + int(time.time()) + 600) + declaration = {"root": str(point), "sha256": sealed["sha256"], "deadline_epoch": start.deadline_epoch} + return locals() + + +def _empty_receiver(case): + for name in ("runtime", "registry", "home"): + path = case[name] + shutil.move(path, path.with_name(path.name + ".source")) + case["runtime"].mkdir() # Ordinary installer creates an empty directory. + + +@pytest.mark.parametrize("retained_point", [("file", False), ("sqlite", False), ("sqlite", True)], indirect=True) +def test_full_restore_preserves_canonical_work_and_requires_exact_binding_cas(retained_point, monkeypatch): + c = retained_point + assert c["start"].verify(str(c["workspace"])) + _empty_receiver(c) + warm.restore_worker(c["declaration"], c["workspace"]) + verify_restored_loopx_state(c["env"], c["point"], c["metadata"], deadline=time.time() + 60) + before = research_context(c["env"], deadline=time.time() + 60) + assert before == c["context"] + assert before["snapshot"]["source_authority"] == c["provider"] + "_v0" + if not c["real_host"]: + def fork(*args, **kwargs): + c["home"].mkdir() + (c["home"] / "config.toml").write_bytes(c["config"].read_bytes()) + return {"session_id": "verified-fork", "model_calls": 0} + monkeypatch.setattr("loopx.hosts.codex_checkpoint.fork_native_checkpoint", fork) + admitted = admit_native_checkpoint_session(c["env"], c["point"], c["sealed"]["sha256"], c["config"], + sandbox="read-only", timeout=60, restore_loopx_state=True) + assert admitted["session_binding_admitted"] and admitted["model_calls"] == 0 + binding = load_codex_cli_session(c["runtime"], lineage=c["wake"].lineage, session_scope="agent") + assert binding["session_id"] != c["source_session"] and binding.get("goal_ref") == c["wake"].goal_ref + assert research_context(c["env"], deadline=time.time() + 60) == before + lease = c["cli"]("task-lease", "inspect", "--goal-id", "research", "--todo-id", c["todo"])["lease"] + assert lease["version"] == c["lease"]["version"] and lease["expires_at"] == c["lease"]["expires_at"] + todos = c["cli"]("todo", "list", "--goal-id", "research")["todos"] + assert len(todos) == 2 and c["todo"] in {row["todo_id"] for row in todos} + assert not any(row.get("title", "").startswith("Read the full retained checkpoint") for row in todos) + assert c["code"].read_text() == "retained candidate" + settled = read_heartbeat_settlement(c["runtime"], goal_id="research", agent_id="researcher", todo_id=c["todo"], + turn_instance_id="source-work", registry_path=c["registry"]) + assert settled.identity.value == c["settled"].identity.value and settled.progress == c["settled"].progress + guard = c["cli"]("quota", "should-run", "--goal-id", "research", "--agent-id", "researcher", + "--turn-instance-id", "continued-work", "--runtime-profile", "generic_cli", "--context-delivery", "off", + "--available-capability", "shell") + assert guard["selected_todo"]["todo_id"] == c["todo"] + assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert len(c["calls"]) == (1 if c["real_host"] else 0) + + +def test_full_restore_cannot_replace_a_binding_changed_during_native_fork(retained_point, monkeypatch): + c = retained_point + _empty_receiver(c) + warm.restore_worker(c["declaration"], c["workspace"]) + def fork(*args, **kwargs): + c["home"].mkdir() + (c["home"] / "config.toml").write_bytes(c["config"].read_bytes()) + _store_codex_cli_session(c["runtime"], lineage=c["wake"].lineage, session_scope="agent", + session_id="concurrent-session", goal_ref=c["wake"].goal_ref, session_profile_digest=c["wake"].digest) + return {"session_id": "verified-fork", "model_calls": 0} + monkeypatch.setattr("loopx.hosts.codex_checkpoint.fork_native_checkpoint", fork) + with pytest.raises(ValueError, match="binding changed"): + admit_native_checkpoint_session(c["env"], c["point"], c["sealed"]["sha256"], c["config"], + sandbox="read-only", timeout=60, restore_loopx_state=True) + binding = load_codex_cli_session(c["runtime"], lineage=c["wake"].lineage, session_scope="agent") + assert binding["session_id"] == "concurrent-session" + + +def test_restored_work_keeps_current_user_gate_and_does_not_change_cadence(retained_point): + c = retained_point + _empty_receiver(c) + warm.restore_worker(c["declaration"], c["workspace"]) + c["cli"]("todo", "add", "--goal-id", "research", "--role", "user", "--task-class", "user_gate", + "--global-gate", "--text", "Authorize the next operation.") + guard = c["cli"]("quota", "should-run", "--goal-id", "research", "--agent-id", "researcher", + "--turn-instance-id", "current-gate", "--runtime-profile", "generic_cli", "--context-delivery", "off") + assert guard["should_run"] is False + profile = c["cli"]("configure-goal", "--goal-id", "research") + assert (profile.get("after") or profile["before"])["execution_profile"]["replan_after_effective_turns"] == 6 + + +@pytest.mark.parametrize("collision", ["registry", "runtime"]) +def test_full_restore_never_overwrites_a_live_or_partial_receiver(retained_point, collision): + c = retained_point + if collision == "runtime": + c["registry"].unlink() + else: + shutil.rmtree(c["runtime"]) + with pytest.raises(ValueError, match="already has state"): + warm.restore_worker(c["declaration"], c["workspace"]) + assert c["code"].read_text() == "retained candidate" + assert not c["workspace"].with_name("workspace.before-checkpoint").exists() + + +@pytest.mark.parametrize("change", [{"goal_id": "fresh"}, {"agent_id": "fresh"}]) +def test_full_state_cannot_be_substituted_for_research_import(retained_point, change): + c = retained_point + with pytest.raises(ValueError): + replace(c["start"], **change).verify(str(c["workspace"])) diff --git a/tests/control_plane/test_warm_continuation_readback.py b/tests/control_plane/test_warm_continuation_readback.py deleted file mode 100644 index 9df671d948..0000000000 --- a/tests/control_plane/test_warm_continuation_readback.py +++ /dev/null @@ -1,203 +0,0 @@ -"""Continuation readiness uses live Goal owners, not old plan or clock prose.""" -from __future__ import annotations - -import time - -import pytest - -from benchmark.runtime.sforge_warm_start import _PREPARATION_TASK, _read_preparation, restore_workspace -from loopx.control_plane.effect_runtime import effect_runtime_result -from test_replan_successor_guard_reentry import AGENT, GOAL, _fixture - - -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_current_preparation_readback_is_idempotent_and_respects_user_gate(tmp_path, monkeypatch, provider): - call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False) - project = tmp_path / "project" - candidate = project / "candidate.txt" - candidate.write_text("retained candidate; old Host deadline belongs to the prior phase") - component = {"id": "workspace", "source_path": str(project)} - probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": [component]}) - point = tmp_path / "captured-point" - sealed = effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "capture", "destination": str(point), "components": [component], - "expected_source_sha256": probe["source_sha256"], "point_metadata": { - "native_session": {"guest_workspace": str(project)}, - "worker_file_snapshot": {"components": [component]}, - }, - }) - (project / "image-only-file").write_text("not part of the retained work") - restore_workspace({"root": str(point), "sha256": sealed["sha256"], - "deadline_epoch": int(time.time()) + 600}, project) - assert candidate.read_text().startswith("retained candidate") - assert not (project / "image-only-file").exists() - created = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, - "--text", _PREPARATION_TASK, "--task-class", "advancement_task", "--required-capability", "shell", - "--operation-id", "checkpoint-preparation") - todo = created["todo_id"] - deadline = int(time.time()) + 600 - declaration = {"deadline_epoch": deadline} - state = project / "ACTIVE_GOAL_STATE.md" - state_before = state.read_bytes() - readback = _read_preparation(call, GOAL, AGENT, todo, declaration) - repeated = _read_preparation(call, GOAL, AGENT, todo, declaration) - assert readback == repeated - assert readback["preparation_todo_id"] == todo - assert readback["authority_read"]["provider_revision"] - assert readback["selected_todo_id"] == todo or readback["current_replan_required"] - assert state.read_bytes() == state_before - assert declaration["deadline_epoch"] == deadline - listed = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"] - assert listed["status"] == "open" and listed["claimed_by"] == AGENT - assert listed["title"] == _PREPARATION_TASK - call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate", - "--global-gate", "--text", "Authorize the separate destination operation.") - with pytest.raises(ValueError, match="no actionable"): - _read_preparation(call, GOAL, AGENT, todo, declaration) - - -def test_expired_preparation_does_not_read_or_renew_a_phase(): - declaration = {"deadline_epoch": 1} - calls = [] - with pytest.raises(ValueError, match="expired"): - _read_preparation(lambda *args: calls.append(args), "goal", "agent", "todo", declaration) - assert calls == [] and declaration == {"deadline_epoch": 1} - - -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("boundary", ["before_guard", "after_guard"]) -@pytest.mark.parametrize("change", ["complete", "transfer"]) -def test_changed_preparation_cannot_borrow_an_unrelated_current_review( - tmp_path, monkeypatch, provider, boundary, change, -): - call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False) - todo = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, - "--text", _PREPARATION_TASK, "--task-class", "advancement_task")["todo_id"] - changed = False - - def interleaved(*args): - nonlocal changed - result = call(*args) - trigger = ("todo", "list") if boundary == "before_guard" else ("quota", "should-run") - if not changed and args[:2] == trigger: - changed = True - if change == "complete": - call("todo", "complete", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", todo) - else: - call("todo", "update", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", todo, - "--claimed-by", "different-agent") - return result - - with pytest.raises(ValueError, match="preparation.*changed|not open|no actionable current preparation"): - _read_preparation(interleaved, GOAL, AGENT, todo, {"deadline_epoch": int(time.time()) + 600}) - assert changed - final = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"] - assert final["status"] == "done" if change == "complete" else final["claimed_by"] == "different-agent" - - -@pytest.mark.parametrize("todo", [ - {"status": "done", "claimed_by": "agent"}, - {"status": "open", "claimed_by": "another-agent"}, -]) -def test_closed_or_foreign_preparation_cannot_use_a_saved_ready_plan(todo): - calls = [] - - def read(*args): - calls.append(args) - return {"todo": todo} - - with pytest.raises(ValueError, match="not open"): - _read_preparation(read, "goal", "agent", "todo", {"deadline_epoch": int(time.time()) + 60}) - assert len(calls) == 1 and calls[0][:2] == ("todo", "list") - - -def test_expiry_during_readback_cannot_report_continuation_ready(tmp_path, monkeypatch): - from benchmark.runtime import sforge_warm_start as warm - - call, _, _ = _fixture(tmp_path, monkeypatch, "file", future_monitor=False) - todo = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, - "--text", _PREPARATION_TASK, "--task-class", "advancement_task")["todo_id"] - clock = [time.time()] - deadline = int(clock[0]) + 600 - monkeypatch.setattr(warm, "time", type("Clock", (), {"time": staticmethod(lambda: clock[0])})) - - def expires_after_actual_guard(*args): - result = call(*args) - if args[:2] == ("quota", "should-run"): - clock[0] = deadline - return result - - with pytest.raises(ValueError, match="expired"): - _read_preparation(expires_after_actual_guard, GOAL, AGENT, todo, {"deadline_epoch": deadline}) - - -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("legacy", [False, True]) -def test_preparation_create_reuses_original_payload_without_a_second_todo(tmp_path, monkeypatch, provider, legacy): - from benchmark.runtime.sforge_warm_start import _preparation_todo - - call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False) - text = _PREPARATION_TASK - if legacy: - text += ( - " This receiver was explicitly admitted to continue the retained work. Earlier Host deadlines " - "and Host-budget closeout messages belong to the source phase; follow the current Goal and " - "guard rather than using those historical messages to close this continuation. The Host " - "enforces the admitted allowance externally. Original task time constraints, submission " - "cooldowns, user gates and stop authority remain binding. No additional time or authority " - "is granted by this instruction." - ) - original = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, - "--text", text, "--task-class", "advancement_task", "--required-capability", "shell", - "--operation-id", "checkpoint-preparation")["todo_id"] - before = call("todo", "list", "--goal-id", GOAL) - assert _preparation_todo(call, GOAL, AGENT) == original - assert _preparation_todo(call, GOAL, AGENT) == original - after = call("todo", "list", "--goal-id", GOAL) - assert after["authority_read"] == before["authority_read"] - assert after["todos"] == before["todos"] - - -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_preparation_operation_cannot_be_repurposed(tmp_path, monkeypatch, provider): - from benchmark.runtime.sforge_warm_start import _preparation_todo - - call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False) - call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, - "--text", "Different work", "--operation-id", "checkpoint-preparation") - before = call("todo", "list", "--goal-id", GOAL) - with pytest.raises(ValueError, match="another request"): - _preparation_todo(call, GOAL, AGENT) - assert call("todo", "list", "--goal-id", GOAL)["authority_read"] == before["authority_read"] - - -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_rejected_context_preparation_does_not_change_cadence(tmp_path, monkeypatch, provider): - import json - import os - import sys - from benchmark.runtime import sforge_warm_start as warm, worker - from loopx.hosts import research_checkpoint - - call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False, handoff_mode="hard_lease") - registry = tmp_path / "registry.json" - call("configure-goal", "--goal-id", GOAL, "--execution-replan-after-turns", "6", "--execute") - call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate", - "--global-gate", "--text", "Wait for the user's authorization") - before = registry.read_bytes() - home, skills = tmp_path / "home", tmp_path / "skills" - home.mkdir() - skills.mkdir() - env = {**os.environ, "LOOPX_GOAL_ID": GOAL, "LOOPX_AGENT_ID": AGENT, - "CODEX_HOME": str(home), "LOOPX_SHARED_SKILLS": str(skills), - "LOOPX_PROJECT": str(tmp_path / "project")} - monkeypatch.setattr(warm, "admit_warm_session", lambda *_: {}) - monkeypatch.setattr(research_checkpoint, "adopt_todo_context", lambda *_, **__: {}) - monkeypatch.setattr(worker, "loopx_command", lambda _: [sys.executable, "-m", "loopx.cli", - "--registry", str(registry), "--runtime-root", str(tmp_path / "runtime"), "--format", "json"]) - with pytest.raises(ValueError, match="no actionable"): - warm.prepare_context(env, {"root": str(tmp_path / "point"), "sha256": "a" * 64, - "goal_id": GOAL, "agent_id": AGENT, "source_todo_id": "source", "normal_cadence": 6, - "deadline_epoch": int(time.time()) + 600}) - assert registry.read_bytes() == before - assert json.loads(registry.read_bytes()) == json.loads(before)