diff --git a/benchmark/README.md b/benchmark/README.md index 3ed4d4e2d7..15fc76cd7c 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -751,12 +751,12 @@ files remain historical: current authority uses a fresh external registry/runtim and the old Goal cannot be admitted through it. Only the model response stream and grading/backend fixtures are synthetic; the native CLI, canonical SQLite authority, worker, supervisor and capsule transport are real. -The fresh-context variant now keeps the controller and its bank outside every -guest mount. It performs and settles actual candidate work, reaches a genuinely +The retained-history recapture variant keeps the controller and its bank outside +every guest mount. It performs and settles actual candidate work, reaches a genuinely new due replan, and captures research, new native history, code, incumbent, opportunity and the original phase through the real controller barrier. Another isolated receiver restores that newly sealed point and performs actual work. -`fresh.log` and `grandreceiver.log` retain those assertions. The synthetic service +`recapture.log` and `grandreceiver.log` retain those assertions. The synthetic service uses host networking only for this controller-backed variant; its public worker registration enters the guest, while the controller configuration and bank do not. @@ -963,7 +963,23 @@ Warm verification hashes the complete checkpoint with a read window of up to 60 seconds, capped by the original execution deadline. It rejects a read that finishes after that deadline. The launcher reuses this verified metadata before native registration; subsequent participant checks still verify the retained -bytes. This read window grants no extra execution time. +bytes. Workspace restoration uses the same bounded full-artifact read allowance. +Native conversation admission and fresh-replan revalidation allow up to 90 seconds +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. +These allowances grant no extra execution time and +do not establish sustained performance for growing checkpoints. + +Concurrent receivers importing the same native point serialize bank publication +with a bounded per-point lock. Each receiver still obtains its own native run and +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 @@ -972,49 +988,39 @@ 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. -An independently authorized context ablation can add -`--restore-fresh-after-replan`. After the initial accepted, settled destination -replan, the next work wake starts one new native conversation; later wakes resume -that conversation. It retains the original -checkpoint digest, source conversation and completed current decision in the -existing session binding; the first input includes that decision as evidence. -Research files, current Todo/Explore state, code, incumbent, cumulative deadline, -consumed submissions and cooldown retain their owners. This deliberately drops -earlier native messages from the new conversation input; archived source bytes -are retained. This is a separate treatment from ordinary retained-history -continuation. Omission preserves the default. Disable it on a new attempt; -do not change a frozen trial. It grants neither new work nor extra evaluation. -Unknown launches recover only from the original captured native output; missing -or ambiguous identities stop that route instead of creating another conversation. -The joined qualification described above exercises original-output recovery, -actual two-candidate work and settlement, then a new due review with full -controller-backed capture and useful work in another isolated receiver. Scripted -model and grader responses qualify this declared mechanical chain, not research -benefit or sustained growing-point reliability. +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, +retained research evidence, code and incumbent; no post-checkpoint decision or +old native message prefix is supplied. Preparation 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 之后的决定或旧原生消息前缀。准备成本仍计入原预算，后续捕获 +仍须通过当前复盘和结算校验。省略该选项保持历史接续。 +必须用新 attempt 改变处理方式，不能重置预算、机会或冷却。 + +Warm continuation preserves the admitted native conversation across the current +replan and subsequent work. The post-replan context-reset option +`--restore-fresh-after-replan` has been removed; supplying it is a CLI error +before trial creation or native registration. The joined qualification retains full controller-backed +recapture and actual work in another isolated receiver, using the complete +retained native history. Controller configuration and bank stay outside guest +mounts. Scripted model and grader responses qualify the mechanical chain; +research benefit and sustained growing-point reliability remain unproven. This is a benchmark operator transport; the generic packaged restore UI remains an open acceptance boundary. -资格脚本的 fresh 分支还要求：当前真实工作结算后产生新的 due replan，经 controller -完整封存研究、新原生历史、代码、incumbent、机会与原 phase，再由另一隔离 receiver -实际接续。controller 配置/bank 不进入 guest 挂载。可向上述资格命令同时增加 -`--reuse-point /private/sealed-point --captured-config /private/captured-config.toml` -复用合成源点，逐项核验原字节和配置摘要；原 source 测试不计通过，所有 receiver -仍真实执行。显式上下文重置使用空原生前缀时，仍须绑定原点/已完成复盘/新会话， -并观察新完成 Turn；空前缀不等于缺声明，更不授予捕获或执行权。科研效用、持续容量 -及真实新 cell 的版本/准入验收仍独立保留。 - -独立获授权的上下文消融可增加 `--restore-fresh-after-replan`：当前目标的 replan -首次受理并结算后，下一工作唤醒才新建一次原生会话，后续唤醒恢复该会话。 -既有会话绑定保留原 checkpoint -摘要、源会话及本次已完成决策，首次输入把决策作为证据交付；研究文件、当前 -Todo/Explore、代码、incumbent、累计时限、已用评测和冷却仍由原 owner 管理。 -新会话输入不包含此前原生消息，源归档字节仍保留；这是独立实验因素。 -省略即保持默认，关闭须使用新 attempt， -不热改冻结运行，也不增加执行或评测权限。未知启动只能读取原始原生输出来恢复， -缺失或歧义身份会阻塞该路径，不能另建会话。上述联合资格覆盖原输出恢复、实际 -双候选工作与结算，再经新的 due review 完整捕获，并由另一隔离接收端继续工作。 -脚本模型与 grader 只验证上述机械链路，不证明研究收益或 checkpoint 持续增大时 -的可靠性。该入口属于 benchmark 操作者传输，通用打包 restore UI 仍须另验收。 +暖恢复在当前 replan 及后续工作中保留已准入的原生会话。复盘后清空上下文的选项 +已删除；旧参数会在创建 trial 或原生注册前作为未知参数拒绝。联合资格仍覆盖 +controller 完整再捕获及另一隔离接收端的实际工作，并使用完整保留的原生历史。 +controller 配置和 bank 不进入 guest 挂载。脚本模型与 grader 只验证机械链路， +研究收益及 checkpoint 持续增大时的可靠性仍未证明。该入口属于 benchmark +操作者传输，通用打包 restore UI 仍须另验收。 Research files outside the task workspace need explicit capture **and** explicit receiver selection. Add each required directory or file to the source checkpoint diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py index 17c1c60745..8e42612281 100644 --- a/benchmark/edgebench/checkpoint_controller.py +++ b/benchmark/edgebench/checkpoint_controller.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse +from contextlib import contextmanager import fcntl import hashlib import json @@ -16,6 +17,7 @@ import stat import tempfile import time +from time import monotonic, sleep from pathlib import Path from urllib.error import HTTPError from urllib.parse import urlsplit @@ -238,6 +240,36 @@ def capture(controller_file: Path, *, token: str, capture_id: str, destination: return {**result, "native_reference": reference} +@contextmanager +def _bank_import_lock(target: Path, timeout: float): + """Serialize publication of one provider bank member, never run admission. + + Keep the lock inode: unlinking it would let a third importer bypass a waiter. + Staging stays outside the lock; partial destinations still fail verification. + """ + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError("Native bank import timeout must be finite and positive") + lock = target.with_name(".loopx-import-" + target.name + ".lock") + descriptor = os.open(lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077 or info.st_nlink != 1: + raise ValueError("Native bank import lock must be an owner-only regular file") + deadline = monotonic() + timeout + while True: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + remaining = deadline - monotonic() + if remaining <= 0: + raise TimeoutError("Native bank import is busy; inspect the original attempt before retry") + sleep(min(.02, remaining)) + yield + finally: + os.close(descriptor) + + def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: str, timeout: float = 30, phase_deadline_file: Path | None = None, recover: bool = False, phase_extension_seconds: int = 0) -> dict: @@ -307,22 +339,23 @@ def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: }, retry_safe=False) imported = stage / "unit/components/native_judge" _bound_bank(imported, reference) - if target.exists(): - _bound_bank(_physical(target), reference) # No overwrite, even on retry. - else: - # POSIX rename can replace an empty destination. Reserve its name - # exclusively; interruption leaves an unusable point, never grants. - target.mkdir(mode=0o700) - try: - imported.rename(target) - except BaseException: - target.rmdir() - raise - descriptor = os.open(bank, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) + with _bank_import_lock(target, timeout): + if target.exists(): + _bound_bank(_physical(target), reference) # No overwrite, even on retry. + else: + # POSIX rename can replace an empty destination. Reserve its name + # exclusively; interruption leaves an unusable point, never grants. + target.mkdir(mode=0o700) + try: + imported.rename(target) + except BaseException: + target.rmdir() + raise + descriptor = os.open(bank, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) body = {"checkpoint_id": reference["checkpoint_id"], "manifest_sha256": reference["manifest_sha256"], "run_id": run_id} phase_epoch = None diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 80b8f710c1..71f3fe0a64 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -136,8 +136,8 @@ def main(argv=None): parser.add_argument("--restore-source-todo") 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-after-replan", action="store_true", - help="Explicit warm treatment: reset native context once after the initial completed current replan; retain research and checkpoint lineage") + parser.add_argument("--restore-fresh-replan", action="store_true", + help="Start a fresh conversation before checkpoint preparation and the first current replan") 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, @@ -187,8 +187,8 @@ def main(argv=None): parser.error("A non-negative phase extension requires an explicit warm checkpoint continuation") if args.restore_research_component and not args.restore_checkpoint: parser.error("Research component restoration requires an explicit warm checkpoint continuation") - if args.restore_fresh_after_replan and not args.restore_checkpoint: - parser.error("Fresh context after replan requires an explicit warm checkpoint continuation") + if args.restore_fresh_replan and not args.restore_checkpoint: + parser.error("Fresh context requires an explicit warm checkpoint continuation") if args.restore_checkpoint and (not handoff or args.eval_interval != 0 or args.replan_after_todos is not None): parser.error("Warm continuation requires full native checkpoint capture, no periodic evaluation and effective-Turn cadence") # One directory is one attempt: never reuse native registration or overwrite @@ -261,7 +261,7 @@ def main(argv=None): candidate = WarmStart(args.restore_checkpoint, args.restore_sha256, args.restore_config, args.restore_goal, args.restore_agent, args.restore_source_todo, int(time.time()) + args.timeout, args.replan_after_turns or 6, - tuple(args.restore_research_component), args.restore_fresh_after_replan) + tuple(args.restore_research_component), args.restore_fresh_replan) metadata = candidate.verify(task.cwd) snapshot = metadata["worker_file_snapshot"] capsule = args.restore_checkpoint / "components" / snapshot["native_capsule_component_id"] @@ -340,7 +340,7 @@ def main(argv=None): "status": "starting", "score_countable": False, **({"restore_phase_extension_seconds": args.restore_phase_extension_seconds} if args.restore_phase_extension_seconds else {}), - **({"restore_fresh_after_replan": True} if args.restore_fresh_after_replan else {}), + **({"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} diff --git a/benchmark/runtime/checkpoint_continuation.py b/benchmark/runtime/checkpoint_continuation.py index b8661eec7e..37df0873d5 100644 --- a/benchmark/runtime/checkpoint_continuation.py +++ b/benchmark/runtime/checkpoint_continuation.py @@ -19,9 +19,13 @@ def retained_native_history(binding, metadata): refresh = binding.get("checkpoint_context_refresh") if refresh is None: return metadata["native_session"] + review = refresh.get("review_basis") + pending = refresh.get("pre_replan_basis") + if ((review is None) == (pending is None) or + not (review or pending).get("decision_sha256")): + raise ValueError("checkpoint context refresh lacks one qualified boundary") if (refresh.get("checkpoint_sha256") != binding.get("checkpoint_artifact_sha256") - or refresh.get("session_id") != binding.get("session_id") - or not refresh.get("review_basis", {}).get("decision_sha256")): + or refresh.get("session_id") != binding.get("session_id")): raise ValueError("checkpoint context refresh provenance differs") return {"turn_ids": [], "visible_history_sha256": hashlib.sha256(b"[]").hexdigest()} diff --git a/benchmark/runtime/checkpoint_session_refresh.py b/benchmark/runtime/checkpoint_session_refresh.py index 0ca86a6080..1a077ce377 100644 --- a/benchmark/runtime/checkpoint_session_refresh.py +++ b/benchmark/runtime/checkpoint_session_refresh.py @@ -1,4 +1,4 @@ -"""Explicit warm-receiver native-context treatment after completed replans. +"""Explicit warm-receiver native-context treatments at verified boundaries. The existing typed checkpoint owner qualifies the current decision. Native session IO remains at CodexAgentSession; this adapter composes those owners @@ -10,45 +10,82 @@ import json from pathlib import Path -from .checkpoint_continuation import restored_capture_gate, restored_checkpoint_review, retained_native_history +from .checkpoint_continuation import retained_native_history def prepare_session_refresh(env, session, wake): declaration = json.loads(Path(env["LOOPX_WARM_START"]).read_bytes()) - if type(declaration.get("fresh_after_replan", False)) is not bool: - raise ValueError("fresh_after_replan requires an explicit boolean treatment") - if not declaration.get("fresh_after_replan", False): + if "fresh_after_replan" in declaration: + raise ValueError("fresh_after_replan is no longer supported; use a pre-replan checkpoint") + if type(declaration.get("fresh_replan", False)) is not bool: + raise ValueError("fresh_replan requires an explicit boolean treatment") + if not declaration.get("fresh_replan", False): return "" if env.get("LOOPX_ITERATION_CONTEXT") != "resume": - raise ValueError("post-replan fresh context requires a resumed warm receiver") + raise ValueError("checkpoint fresh context requires a resumed warm receiver") if any(env[key] != declaration[field] for key, field in (("LOOPX_GOAL_ID", "goal_id"), ("LOOPX_AGENT_ID", "agent_id"))): - raise ValueError("post-replan fresh context destination lineage differs") + raise ValueError("checkpoint fresh context destination lineage differs") if not session.binding or session.binding.get("checkpoint_artifact_sha256") != declaration["sha256"]: - raise ValueError("post-replan fresh context requires its exact checkpoint binding") + raise ValueError("checkpoint fresh context requires its exact checkpoint binding") current = session.binding.get("checkpoint_context_refresh") if current is not None: + if bool(current.get("pre_replan_basis")) != bool(declaration.get("fresh_replan")): + raise ValueError("checkpoint context treatment differs from committed session") retained_native_history(session.binding, {}) return "" # This receiver's initial reset is already committed. - gate = restored_checkpoint_review(env) - if not gate.get("semantic_review_accepted"): - return "" - review = gate["review_basis"] - # The initial reset requires completed actual native work at the source. - # The cheap semantic-only read above is never a capture permission. - confirmed = restored_capture_gate(env, include_review_basis=True) - if confirmed.get("review_basis") != review: - raise ValueError("current completed replan changed before native context refresh") - key = hashlib.sha256(review["identity"]["effect_id"].encode()).hexdigest() - attempt = Path(env["LOOPX_WARM_START"]).parent / "context-refresh" / (key + ".json") - session.prepare_checkpoint_refresh(review, attempt, wake / "stdout.jsonl") + return _prepare_fresh_replan(env, session, wake, declaration) + + +def _prepare_fresh_replan(env, session, wake, declaration): + """Compose existing sealed-point selection and native admission before model work. + + The TypeScript artifact owner decides whether this is a pending replan point; + native admission rejects an already-advanced source fork. No completed review + is manufactured or fed to the fresh decision maker. + """ + from loopx.control_plane.effect_runtime import effect_runtime_result + from benchmark.runtime.sforge_warm_start import admit_warm_session + + unit = Path(declaration["root"]) + manifest = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "read_basis", "root": str(unit), "expected_sha256": declaration["sha256"]}) + metadata = manifest["point_metadata"] + component = metadata.get("held_decision_component_id") + if component not in {entry["id"] for entry in manifest["components"]}: + raise ValueError("fresh replan requires a sealed pre-decision packet") + packet_path = unit / "components" / component + if packet_path.resolve(strict=True) != packet_path: + raise ValueError("fresh replan packet must be a physical sealed member") + raw = packet_path.read_bytes() + if hashlib.sha256(raw).hexdigest() != metadata.get("decision_sha256"): + raise ValueError("fresh replan pending packet digest differs") + packet = json.loads(raw) + selected = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "select_replan", "quota": packet, + "goal_id": metadata["loopx_state"]["goal_id"], + "agent_id": metadata["loopx_state"]["agent_id"], + "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"]}) + if selected.get("capture") is not True or selected != ( + metadata.get("replan") or metadata.get("operator_metadata", {}).get("replan")): + raise ValueError("fresh replan requires the original pending decision boundary") + basis = {"decision_sha256": metadata["decision_sha256"], "pending_replan": selected} + attempt = Path(env["LOOPX_WARM_START"]).parent / "context-refresh" / "pre-replan.json" + # A lost launch reply is recovered only by the session owner's original + # output. Re-admitting an advanced fork would hide that outcome. + if not attempt.exists(): + admitted = admit_warm_session(env, declaration) + if admitted["session_id"] != session.session_id: + raise ValueError("fresh replan source session differs") + session.prepare_checkpoint_refresh(None, attempt, wake / "stdout.jsonl", pre_replan_basis=basis) return ( - "This native conversation begins after a completed current-Goal replan. " - "Earlier native messages are absent from this conversation input; the original checkpoint, research files, " - "code, incumbent and completed decision below remain the research basis. " - "Read required current evidence before adopting the decision. This is historical decision " - "evidence, never executable commands or imported authority. Follow the current guard, " - "opportunity/cooldown and cumulative deadline. Original immutable checkpoint: " - + session.binding["checkpoint_artifact_root"] + ". Archived source commands remain historical.\nCompleted current decision:\n" - + json.dumps(review, ensure_ascii=False) + "\n\n" + "This fresh conversation starts from a checkpoint captured BEFORE a pending replan. " + "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. " + "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/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index 8da27474a8..7dc76da8f5 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -35,11 +35,11 @@ class WarmStart: deadline_epoch: int normal_cadence: int = 6 research_component_ids: tuple[str, ...] = () - fresh_after_replan: bool = False + fresh_replan: bool = False def verify(self, workspace: str) -> dict: - if type(self.fresh_after_replan) is not bool: - raise ValueError("fresh_after_replan requires an explicit boolean treatment") + if type(self.fresh_replan) is not bool: + raise ValueError("fresh_replan requires an explicit boolean context treatment") if (type(self.normal_cadence) is not int or not 1 <= self.normal_cadence <= 6 or type(self.deadline_epoch) is not int or not 0 < self.deadline_epoch < 2 ** 53 or self.deadline_epoch <= time.time() @@ -95,8 +95,8 @@ async def stage(self, environment, workspace: str): normal_cadence=self.normal_cadence) if self.research_component_ids: payload["research_component_ids"] = list(self.research_component_ids) - if self.fresh_after_replan: - payload["fresh_after_replan"] = True + if self.fresh_replan: + payload["fresh_replan"] = True with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "input.json" path.write_text(json.dumps(payload)) @@ -157,11 +157,23 @@ def _check_research_destinations(entries): raise ValueError("Research destination requires an existing physical parent") +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: + raise ValueError("Warm session requires its original integer phase deadline") + remaining = deadline - time.time() + if remaining <= 0: + raise ValueError("Original warm start allowance expired before checkpoint admission") + return min(limit, remaining) + + def restore_workspace(declaration: dict, workspace: Path): """Materialize one isolated receiver workspace; never import its registry.""" unit = Path(declaration["root"]) effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "verify", "root": str(unit), "expected_sha256": declaration["sha256"]}) + "action": "verify", "root": str(unit), "expected_sha256": declaration["sha256"]}, + timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS)) + _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS) manifest = json.loads((unit / "checkpoint.json").read_bytes()) metadata = manifest["point_metadata"] research = _research_components(unit, metadata, workspace, declaration.get("research_component_ids", ())) @@ -198,9 +210,11 @@ def restore_workspace(declaration: dict, workspace: Path): shutil.copystat(captured, target) if research: copied = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", - "components": [{"id": key, "source_path": str(target)} for key, _, target in research]}) + "components": [{"id": key, "source_path": str(target)} for key, _, target in research]}, + timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS)) if any(copied["source_witness"][key] != manifest["artifact_witness"][key] for key, _, _ in research): raise ValueError("Restored research files differ from the sealed artifacts; preserve this failed receiver") + _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS) def _prepare_native_skills(home: Path, installed: Path) -> None: @@ -225,10 +239,24 @@ def _prepare_native_skills(home: Path, installed: Path) -> None: skills.symlink_to(installed.resolve(), target_is_directory=True) +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 + and fresh-context revalidation. This retains source/cursor checks; it neither + trusts a cached history digest nor grants another execution window. + """ + from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session + + 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) + _phase_read_timeout(declaration, 90) + return result + + def prepare_context(env: dict[str, str], declaration: dict) -> dict: """Join actual session + research owners on one freshly bootstrapped Goal.""" - from benchmark.runtime.codex import Execution - from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session from loopx.hosts.research_checkpoint import adopt_todo_context from benchmark.runtime.worker import loopx_command import subprocess @@ -239,9 +267,7 @@ def prepare_context(env: dict[str, str], declaration: dict) -> dict: raise ValueError("Warm context requires its declared destination lineage") if declaration["deadline_epoch"] <= time.time(): raise ValueError("Original warm start allowance expired during installation") - execution = Execution(context="resume", sandbox=env["LOOPX_CODEX_SANDBOX"]) - native = admit_native_checkpoint_session(env, unit, digest, Path(declaration["config"]), - sandbox=execution.sandbox, timeout=30) + 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, @@ -267,7 +293,7 @@ def cli(*args): 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"]) + 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, diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py index 209e2a498d..38c1f5a011 100644 --- a/benchmark/tests/qualify_checkpoint_joint.py +++ b/benchmark/tests/qualify_checkpoint_joint.py @@ -181,7 +181,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None): git_object = point / "components/task_workspace/.git/objects/ab/retained-object" assert git_object.read_text() == "Readonly source evidence" and git_object.stat().st_mode & 0o777 == 0o444 config = _stage_source_configuration(container, metadata["native_session"], output / "native-config") - for name in ("ordinary", "stride", "fresh"): + for name in ("ordinary", "stride", "recapture"): stride = name != "ordinary" results = output / name results.mkdir() @@ -190,7 +190,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None): extra_env = {} extra_volumes = {} capture_supervisor = None - if name == "fresh": + if name == "recapture": operator = output / "receiver-controller" operator.mkdir(mode=0o700) phase = operator / "phase-deadline" @@ -223,9 +223,9 @@ def guest(command, volumes, env, log, *, capture_supervisor=None): {"LOOPX_JOINT_RECEIVER_POINT": "/point", "LOOPX_JOINT_RECEIVER_CONFIG": "/captured-config.toml", "LOOPX_RUNTIME_ROOT": "/results/transport-runtime", **({"LOOPX_JOINT_RECEIVER_STRIDE": "1"} if stride else {}), - **({"LOOPX_JOINT_RECEIVER_FRESH": "1"} if name == "fresh" else {}), **extra_env}, + **extra_env}, output / f"{name}.log", capture_supervisor=capture_supervisor) - if name == "fresh": + if name == "recapture": newer = [p for p in (publication / "published").iterdir() if p.is_dir() and (p / "checkpoint.json").exists()] assert len(newer) == 1 recaptured = newer[0] @@ -269,7 +269,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None): source / "tests/control_plane/codex_checkpoint_fixture.py"]}, "image_id": image, "point_sha256": packet["sha256"], "binary_sha256": {k: hashlib.sha256(p.read_bytes()).hexdigest() for k, p in binaries.items()}, - "ordinary_and_stride": True, "post_replan_fresh_context": True, "core_goal_without_benchmark": True, + "ordinary_and_stride": True, "core_goal_without_benchmark": True, "source_created_in_this_run": not bool(args.reuse_point), "source_test_passed_in_this_run": not bool(args.reuse_point), "automatic_full_recapture_sha256": next_packet["sha256"], "another_receiver_actual_work": True, diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py index e1635c7154..4dcccb933a 100644 --- a/benchmark/tests/test_checkpoint_controller.py +++ b/benchmark/tests/test_checkpoint_controller.py @@ -741,3 +741,87 @@ def observed_post(controller, route, body, timeout): app.state.judge._reaper_thread.join(timeout=1) client.images.remove(name + ".judge.fixture:fixture") client.close() + + +def test_concurrent_same_point_imports_preserve_both_fresh_native_runs(controllers, tmp_path, monkeypatch): + from concurrent.futures import ThreadPoolExecutor + from pathlib import Path + + source, destination, first, second, files, grades = controllers + token = register(source, first, max_agent_submissions=3, submission_cooldown=0) + assert submit(first, token, 7).status_code == 200 + unit = tmp_path / 'shared-point' + point = transport.capture(files[0], token=token, capture_id='b' * 32, destination=unit) + bank = destination.state.judge.config.log_dir / 'checkpoints' + target = bank / ('b' * 32) + reserved, release, staged_second = threading.Event(), threading.Event(), threading.Event() + original_mkdir = Path.mkdir + original_effect = transport.effect_runtime_result + counts = {'reserve': 0, 'stage': 0} + mutex = threading.Lock() + + def mkdir(path, *args, **kwargs): + if path == target: + with mutex: + counts['reserve'] += 1 + first_import = counts['reserve'] == 1 + if first_import: + reserved.set() + assert release.wait(15), 'Test did not release first publisher' + return original_mkdir(path, *args, **kwargs) + + def effect(method, params, **kwargs): + result = original_effect(method, params, **kwargs) + if params.get('action') == 'materialize': + with mutex: + counts['stage'] += 1 + if counts['stage'] == 2: + staged_second.set() + return result + + monkeypatch.setattr(Path, 'mkdir', mkdir) + monkeypatch.setattr(transport, 'effect_runtime_result', effect) + def restore(run): + return transport.restore(files[1], root=unit, expected_sha256=point['sha256'], run_id=run) + with ThreadPoolExecutor(max_workers=2) as pool: + one = pool.submit(restore, 'branch-one') + try: + assert reserved.wait(15) + two = pool.submit(restore, 'branch-two') + assert staged_second.wait(15) + # Both have staged verified bytes. A competing publisher must wait, + # rather than create the destination while the first is reserving it. + time.sleep(.2) + assert not two.done() + finally: + release.set() + a, b = one.result(timeout=15), two.result(timeout=15) + assert a['token'] != b['token'] != token + assert counts['reserve'] == 1 + for run in ('branch-one', 'branch-two'): + history = destination.state.judge.get_run_history(run, 'fixture') + assert history['best_score'] == 7 and history['agent_submissions'] == 1 + assert len(grades) == 1 + assert (target / 'incumbent.tar.gz').read_bytes() == b'7' + + +def test_bank_import_contention_is_bounded_and_does_not_erase_the_lock(tmp_path): + target = tmp_path / ('e' * 32) + with transport._bank_import_lock(target, 1): + with pytest.raises(TimeoutError, match='Native bank import is busy'): + with transport._bank_import_lock(target, .03): + pytest.fail('Competing publication entered the lock') + assert not target.exists() + with transport._bank_import_lock(target, 1): + assert (tmp_path / ('.loopx-import-' + target.name + '.lock')).is_file() + + +def test_bank_import_rejects_a_redirected_lock(tmp_path): + target = tmp_path / ('f' * 32) + foreign = tmp_path / 'foreign' + foreign.write_text('unchanged') + (tmp_path / ('.loopx-import-' + target.name + '.lock')).symlink_to(foreign) + with pytest.raises(OSError): + with transport._bank_import_lock(target, 1): + pytest.fail('Symlink lock accepted') + assert foreign.read_text() == 'unchanged' and not target.exists() diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index 7adefa2edb..1910877ec0 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/benchmark/tests/test_checkpoint_joint_receiver.py @@ -49,8 +49,6 @@ def test_captured_point_continues_with_fresh_authority( # Each variant uses its own fresh guest because the native fork retains # fixed captured workspace/home paths. The default variant stays unchanged. stride_continuation = os.environ.get("LOOPX_JOINT_RECEIVER_STRIDE") == "1" - fresh_context = os.environ.get("LOOPX_JOINT_RECEIVER_FRESH") == "1" - assert not fresh_context or stride_continuation, "Fresh-context qualification requires the completed current review" unit = Path(os.environ["LOOPX_JOINT_RECEIVER_POINT"]) manifest = json.loads((unit / "checkpoint.json").read_bytes()) digest, metadata = manifest["sha256"], manifest["point_metadata"] @@ -304,7 +302,7 @@ def lose_binding_reply(self, *args): if external: # Operator credentials and the native bank stay outside the guest. # Only this destination's public worker registration enters it. - assert fresh_context + assert stride_continuation url, fresh, grades = external["url"], external["activation"], [] else: _, branch, _, url, controller_files, grades = request.getfixturevalue("controllers") @@ -485,9 +483,6 @@ def do_POST(self): def run_native_work(current_guard, logs, purpose): logs.mkdir() wake = BenchmarkSessionWake(env, execution, {}) - if fresh_context and env.get("LOOPX_WARM_START"): - from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh - purpose = prepare_session_refresh(env, wake, logs) + purpose command = native_command(env, execution, "execute", logs, wake) result = subprocess.run( [ @@ -506,13 +501,6 @@ def run_native_work(current_guard, logs, purpose): ) assert result.returncode == 0, result.stderr (logs / "stdout.jsonl").write_text(result.stdout) - if fresh_context and wake.checkpoint_refresh is not None: - # Lose the first observation reply after actual native work. The - # next Host recovers its original output, without another launch. - from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh - wake = BenchmarkSessionWake(env, execution, {}) - prepare_session_refresh(env, wake, logs.with_name("not-launched")) - assert wake.session_id != admitted["session_id"] wake.observe(logs / "stdout.jsonl") return wake @@ -787,11 +775,10 @@ def run_native_work(current_guard, logs, purpose): ) reopened = restored_capture_gate(env) assert reopened["capture_allowed"], reopened - if fresh_context: - from benchmark.runtime.checkpoint_continuation import restored_checkpoint_review - review_only = restored_checkpoint_review(env) - assert review_only["semantic_review_accepted"] and "capture_allowed" not in review_only - assert review_only["review_basis"]["identity"]["turn_instance_id"] == review_turn + from benchmark.runtime.checkpoint_continuation import restored_checkpoint_review + review_only = restored_checkpoint_review(env) + assert review_only["semantic_review_accepted"] and "capture_allowed" not in review_only + assert review_only["review_basis"]["identity"]["turn_instance_id"] == review_turn assert not reopened["instructions"] # Do not repeat the historical packet after consumption. assert not restored_capture_gate(env, obligation["obligation_id"])["capture_allowed"] configured = cli( @@ -880,11 +867,6 @@ def run_native_work(current_guard, logs, purpose): provider_phase.update( count=0, command=shlex.quote(sys.executable) + " grow.py" ) - if fresh_context: - warm = tmp_path / "warm-start.json" - warm.write_text(json.dumps({"sha256": digest, "goal_id": goal, "agent_id": agent, - "fresh_after_replan": True})) - env.update(LOOPX_WARM_START=str(warm), LOOPX_ITERATION_CONTEXT="resume") larger_wake = run_native_work( next_guard, tmp_path / "larger-work", @@ -909,19 +891,9 @@ def run_native_work(current_guard, logs, purpose): grown = native_session_checkpoint(env, active_session_id, timeout=30)[ "metadata" ] - if fresh_context: - assert active_session_id != admitted["session_id"] and len(grown["turn_ids"]) == 1 - assert "Completed current decision" in json.dumps(requests[2]) - assert "Complete candidates 9 and 10" in json.dumps(requests[2]) - assert larger_wake.binding["checkpoint_artifact_sha256"] == digest - from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh - replay = BenchmarkSessionWake(env, execution, {}) - assert prepare_session_refresh(env, replay, tmp_path / "unlaunched-retry") == "" - assert replay.session_id == active_session_id - assert not (tmp_path / "unlaunched-retry").exists() - else: - assert grown["turn_ids"][:-2] == native["turn_ids"] - assert len(grown["turn_ids"]) == len(native["turn_ids"]) + 2 + assert active_session_id == admitted["session_id"] + assert grown["turn_ids"][:-2] == native["turn_ids"] + assert len(grown["turn_ids"]) == len(native["turn_ids"]) + 2 work_vision = tmp_path / "larger-vision.json" work_vision.write_text( json.dumps( @@ -1022,7 +994,7 @@ def run_native_work(current_guard, logs, purpose): assert cm["held_decision_component_id"] and cm["replan"]["obligation_id"] != obligation["obligation_id"] assert cm["native_session"]["turn_ids"] == grown["turn_ids"] assert (recaptured / "components/task_workspace/candidate.txt").read_text() == "10" - assert cm["session"]["checkpoint_context_refresh"]["checkpoint_sha256"] == digest + assert cm["session"]["session_id"] == admitted["session_id"] nested = recaptured / "components" / cm["worker_file_snapshot"]["native_capsule_component_id"] native_point = json.loads((nested / "checkpoint.json").read_bytes())["point_metadata"] assert (nested / "components/native_judge/incumbent.tar.gz").read_bytes() == b"7" @@ -1059,11 +1031,6 @@ def run_native_work(current_guard, logs, purpose): assert completed["checkpoint"]["captured"], completed captured = json.loads((manual / "checkpoint.json").read_text()) assert captured["point_metadata"]["native_session"]["turn_ids"] == grown["turn_ids"] - if fresh_context: - provenance = captured["point_metadata"]["session"]["checkpoint_context_refresh"] - assert provenance["checkpoint_sha256"] == digest - assert provenance["source_session_id"] == admitted["session_id"] - assert provenance["session_id"] == active_session_id assert (manual / "components/code").read_text() == "10" verified_point = effect_runtime_result("runtime.checkpoint_artifacts", { "action": "verify", "root": str(manual), "expected_sha256": captured["sha256"]}) diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index 207467326b..0d26ef3607 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -98,7 +98,7 @@ def point(tmp_path): @pytest.mark.parametrize("change", [ {"goal_id": "source-goal"}, {"agent_id": "source-agent"}, {"source_todo_id": "missing"}, {"deadline_epoch": 1}, {"deadline_epoch": True}, {"normal_cadence": 0}, {"normal_cadence": True}, - {"fresh_after_replan": "false"}, + {"fresh_replan": "false"}, ]) def test_invalid_participant_rejects_before_workspace_or_authority_changes(point, change): start, workspace = point @@ -164,7 +164,7 @@ def test_restore_replaces_image_files_preserves_modes_and_refuses_second_attempt start, workspace = point assert start.verify(str(workspace)) (workspace / "stale-image-file").write_text("not captured") - declaration = {"root": str(start.root), "sha256": start.sha256} + declaration = {"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch} warm.restore_workspace(declaration, workspace) assert not (workspace / "stale-image-file").exists() assert (workspace / "candidate").read_text() == "retained candidate" @@ -242,7 +242,7 @@ def test_overlay_workspace_backup_survives_cross_device_rename(point, monkeypatc def cross_device(*args, **kwargs): raise OSError(errno.EXDEV, "overlay directory rename") monkeypatch.setattr("os.rename", cross_device) - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256}, workspace) + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch}, workspace) backup = workspace.with_name("workspace.before-checkpoint") assert (backup / "image-only").read_text() == "original image" assert not (workspace / "image-only").exists() @@ -323,7 +323,7 @@ def research_point(point, tmp_path, monkeypatch, request): def test_explicit_research_files_reproduce_the_retained_experiment(research_point): start, workspace, research = research_point assert start.verify(str(workspace)) - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": ["experiments"]}, workspace) scope = {} exec((research / "candidate.py").read_text(), scope) @@ -333,13 +333,13 @@ def test_explicit_research_files_reproduce_the_retained_experiment(research_poin effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(start.root), "expected_sha256": start.sha256}) with pytest.raises(ValueError, match="already exists"): - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": ["experiments"]}, workspace) def test_unselected_research_files_remain_inert(research_point): start, workspace, research = research_point - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256}, workspace) + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch}, workspace) assert not research.exists() @@ -350,7 +350,7 @@ def test_invalid_research_selection_rejects_before_workspace_changes(research_po with pytest.raises(ValueError): replace(start, research_component_ids=selected).verify(str(workspace)) with pytest.raises(ValueError): - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": selected}, workspace) assert (workspace / "candidate").read_text() == "retained candidate" assert not workspace.with_name("workspace.before-checkpoint").exists() and not research.exists() @@ -375,7 +375,7 @@ def test_research_destination_conflicts_preserve_receiver(research_point, collis warm._check_research_destinations([("experiments", start.root, parent / "new-file")]) return with pytest.raises(ValueError, match="already exists"): - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": ["experiments"]}, workspace) assert not workspace.with_name("workspace.before-checkpoint").exists() if collision == "file": @@ -404,7 +404,7 @@ def corrupt(source, destination, *args, **kwargs): return result monkeypatch.setattr(shutil, "copytree", corrupt) with pytest.raises(ValueError, match="differ from the sealed"): - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": ["experiments"]}, workspace) assert workspace.with_name("workspace.before-checkpoint").is_dir() @@ -426,8 +426,12 @@ async def exec(self, command, **kwargs): assert declaration["research_component_ids"] == ["experiments"] -@pytest.mark.parametrize("options", [["--restore-research-component", "experiments"], ["--restore-fresh-after-replan"]]) -def test_research_option_without_warm_entry_rejects_before_registration(tmp_path, options): +@pytest.mark.parametrize("options, error", [ + (["--restore-research-component", "experiments"], "Research component restoration requires"), + (["--restore-fresh-after-replan"], "unrecognized arguments: --restore-fresh-after-replan"), + (["--restore-fresh-replan"], "Fresh context requires an explicit warm checkpoint continuation"), +]) +def test_restore_options_reject_before_registration(tmp_path, capsys, options, error): pytest.importorskip("harbor") from benchmark.edgebench import run as entry with pytest.raises(SystemExit) as rejected: @@ -436,12 +440,13 @@ def test_research_option_without_warm_entry_rejects_before_registration(tmp_path "--effort", "high", "--judge-url", "http://localhost:1", *options]) assert rejected.value.code == 2 and not list(tmp_path.iterdir()) + assert error in capsys.readouterr().err @pytest.mark.parametrize("research_point", ["file"], indirect=True) def test_standalone_research_file_keeps_bytes_and_mode(research_point): start, workspace, research = research_point - warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch, "research_component_ids": ["experiments"]}, workspace) assert research.read_text() == "retained observations" assert research.stat().st_mode & 0o777 == 0o600 @@ -453,3 +458,94 @@ def test_workspace_backup_cannot_be_a_research_destination(research_point): metadata["worker_file_snapshot"]["components"][-1]["source_path"] = str(workspace.with_name("workspace.before-checkpoint")) with pytest.raises(ValueError, match="separate data"): warm._research_components(start.root, metadata, workspace, ("experiments",)) + + +@pytest.mark.parametrize('remaining, expected', [(600, 90), (12, 12)]) +def test_native_admission_full_history_allowance_uses_original_phase(monkeypatch, remaining, expected): + from loopx.hosts import codex_checkpoint + now = 1000 + monkeypatch.setattr(warm, 'time', SimpleNamespace(time=lambda: now)) + calls = [] + def admit(*args, **kwargs): + calls.append((args, kwargs)) + return {'session_id': 'verified-native-fork'} + monkeypatch.setattr(codex_checkpoint, 'admit_native_checkpoint_session', admit) + declaration = {'root': '/point', 'sha256': 'a' * 64, 'config': '/config.toml', + 'deadline_epoch': now + remaining} + env = {'LOOPX_CODEX_SANDBOX': 'danger-full-access'} + assert warm.admit_warm_session(env, declaration)['session_id'] == 'verified-native-fork' + assert calls[0][1]['timeout'] == expected + assert declaration['deadline_epoch'] == now + remaining + + +@pytest.mark.parametrize('deadline', [True, None, '1001', 1000, 999, 2 ** 53]) +def test_native_admission_rejects_expired_or_invalid_phase_before_io(monkeypatch, deadline): + from loopx.hosts import codex_checkpoint + monkeypatch.setattr(warm, 'time', SimpleNamespace(time=lambda: 1000)) + monkeypatch.setattr(codex_checkpoint, 'admit_native_checkpoint_session', + lambda *a, **k: pytest.fail('Native IO after invalid deadline')) + with pytest.raises(ValueError): + warm.admit_warm_session({}, {'deadline_epoch': deadline}) + + +def test_native_admission_cannot_return_success_after_original_phase(monkeypatch): + from loopx.hosts import codex_checkpoint + clock = [1000] + monkeypatch.setattr(warm, 'time', SimpleNamespace(time=lambda: clock[0])) + def admit(*args, **kwargs): + clock[0] = 1002 + return {'session_id': 'too-late'} + monkeypatch.setattr(codex_checkpoint, 'admit_native_checkpoint_session', admit) + with pytest.raises(ValueError, match='allowance expired'): + warm.admit_warm_session({'LOOPX_CODEX_SANDBOX': 'danger-full-access'}, + {'root': '/point', 'sha256': 'a' * 64, 'config': '/config.toml', 'deadline_epoch': 1001}) + + +@pytest.mark.parametrize("remaining, expected", [(600, 60), (7, 7)]) +def test_workspace_verification_uses_bounded_full_point_read(point, monkeypatch, remaining, expected): + start, workspace = point + monkeypatch.setattr(warm, "time", SimpleNamespace(time=lambda: 1000)) + original = warm.effect_runtime_result + calls = [] + def verify(method, params, **kwargs): + calls.append(kwargs["timeout"]) + return original(method, params, **kwargs) + monkeypatch.setattr(warm, "effect_runtime_result", verify) + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + "deadline_epoch": 1000 + remaining}, workspace) + assert calls == [expected] + assert (workspace / "candidate").read_text() == "retained candidate" + + +def test_expired_workspace_verification_preserves_image_before_mutation(point, monkeypatch): + start, workspace = point + clock = [1000] + monkeypatch.setattr(warm, "time", SimpleNamespace(time=lambda: clock[0])) + original = warm.effect_runtime_result + def verify(method, params, **kwargs): + result = original(method, params, **kwargs) + clock[0] = 1002 + return result + monkeypatch.setattr(warm, "effect_runtime_result", verify) + with pytest.raises(ValueError, match="allowance expired"): + warm.restore_workspace({"root": str(start.root), "sha256": start.sha256, + "deadline_epoch": 1001}, workspace) + assert not workspace.with_name("workspace.before-checkpoint").exists() + assert (workspace / "candidate").read_text() == "retained candidate" + + +@pytest.mark.parametrize("legacy", [True, False, "false"]) +def test_removed_post_replan_declaration_rejects_before_native_effects(tmp_path, legacy): + from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh + declaration = tmp_path / "warm.json" + declaration.write_text(json.dumps({"fresh_after_replan": legacy, "fresh_replan": True})) + # No usable session: rejection must precede all native/session access. + with pytest.raises(ValueError, match="no longer supported"): + prepare_session_refresh({"LOOPX_WARM_START": str(declaration)}, None, tmp_path) + + +def test_omitted_fresh_replan_keeps_default_native_history(tmp_path): + from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh + declaration = tmp_path / "warm.json" + declaration.write_text("{}") + assert prepare_session_refresh({"LOOPX_WARM_START": str(declaration)}, None, tmp_path) == "" diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md index 87a97e688a..89c1b820fc 100644 --- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md +++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md @@ -1257,27 +1257,23 @@ a deterministic local provider; absent/multiple outcomes and foreign destination attempts reject. The controller participant recovery above does not yet qualify automatic joint session/research/evaluator/Host activation recovery. -The explicit warm-worker `--restore-fresh-after-replan` treatment uses the same -current semantic-review owner before replacing native context. It binds the -original checkpoint and predecessor conversation to the completed destination -decision, delivers that decision in the new conversation's first input, and -preserves the current research/code/opportunity owners. Ordinary replacement -still cannot inherit provenance. The provider records the original output -location and source basis before dispatch; an unobserved identity cannot permit -a second launch. Later capture must observe completed native work in that new -conversation and seal its refresh lineage. Dropping the old native prefix is -declared information loss, never exact retained-history continuation. Synthetic -receiver/capture qualification and actual model evidence use or research benefit -remain separate; the treatment authorizes no new study cells or hot updates. +By default, warm-worker continuation keeps the admitted native conversation across +the current semantic replan and later work. An explicit `--restore-fresh-replan` +treatment instead starts a fresh conversation before preparation and the first +current replan from a verified pending-decision checkpoint. The original objective, +retained research evidence, code and incumbent remain the decision basis; no +post-checkpoint decision is inherited. Preparation uses the original deadline. The post-replan context-reset option +has been removed. Existing refresh provenance remains readable for previously +captured artifacts; ordinary replacement still cannot inherit provenance. +This removal does not qualify scientific benefit, a new study cell or hot updates. The joined qualification must distinguish a manual partial artifact from a -new automatic full point after useful destination work. The fresh-context +new automatic full point after useful destination work. The retained-history receiver uses its synthetic profile's explicitly selected cadence to reach that new periodic review and the actual controller/native opportunity barrier to seal code/research/native/phase state, then sends that point to another isolated receiver for real -work. Source-prefix verification uses the explicitly bound reset baseline rather -than demanding native messages that this treatment discarded. The completed -cursor and both full history/source observations remain mandatory. Reusing a +work. Source-prefix verification preserves the complete admitted native history. The +completed cursor and both full history/source observations remain mandatory. Reusing a verified synthetic source point skips only source generation, never counts a prior source test as passed, and still exercises every receiver. This closes a mechanical composition seam when qualification passes; sustained large-point diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md index 557d82e728..d0acb2d9ed 100644 --- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md +++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md @@ -978,19 +978,18 @@ pending 包与原生历史，不在每个工作 wake 重算大型代码/incumben 捕获 `0444` 参与方此前失败，修复后通过；Git object store 是这类成员的常见来源。 仅在 root 容器中验证不足以覆盖此 Host 边界。源权限和字节保持不变。 -显式暖 worker 的 `--restore-fresh-after-replan` 处理在替换原生上下文前复用同一 -当前语义复盘 owner，将原 checkpoint、前驱会话及本目标已完成决策绑定，并在新 -会话首次输入交付该决策，保持当前研究、代码和评测机会的 owner。普通替换仍不能 -继承来源。provider 在派发前记录原输出位置和源依据；未确认身份不能允许第二次 -启动。后续捕获必须观察新会话已完成的原生工作，并封存刷新链路。丢弃旧原生前缀 -是明确的信息损失，不是完整保留历史的接续。合成 receiver/捕获资格与真实模型 -证据使用、研究收益分开验收；此处理不授权新实验 cell 或热更新。 - -联合资格须把手动 partial 产物与真实工作之后的新自动完整点分开。fresh-context +默认暖 worker 接续在当前语义 replan 和后续工作中保留已准入的原生会话。 +显式 `--restore-fresh-replan` 则从已验证的待决策 checkpoint 出发，在准备和首次 +当前 replan 之前启动 fresh 会话。原目标、保留的研究证据、代码和 incumbent +仍为决策依据，不继承 checkpoint 之后的决定；准备仍使用原截止时间。复盘后清空 +上下文的选项已删除。历史产物中的刷新来源仍可读取；普通替换仍不能继承来源。 +此删除不证明科研收益，也不授权新实验 cell 或热更新。 + +联合资格须把手动 partial 产物与真实工作之后的新自动完整点分开。保留历史的 receiver 在合成 profile 显式选定的 cadence 下，实际工作结算后到达新的 periodic review，通过真实 controller/native opportunity barrier 封存当前研究、代码、 -原生历史和 phase，再由另一隔离 receiver 执行新工作。显式重置使用其有来源绑定的空原生前缀， -不能仍要求已丢弃的旧消息，也不能免除已完成 cursor、前后完整历史与来源观测。 +原生历史和 phase，再由另一隔离 receiver 执行新工作。源前缀验证保留完整已准入 +原生历史，已完成 cursor、前后完整历史与来源观测仍须核验。 复用已核验合成源点只省去重新造源，不把此前 source 失败改为通过；所有 receiver 仍须执行。通过后只关闭这项机械组合缺口，持续大点容量、安装前端和科研证据采用 仍分开验收。 diff --git a/loopx/control_plane/turn_driver/codex_agent_session.py b/loopx/control_plane/turn_driver/codex_agent_session.py index 88d433ee7e..aa7feb8ddb 100644 --- a/loopx/control_plane/turn_driver/codex_agent_session.py +++ b/loopx/control_plane/turn_driver/codex_agent_session.py @@ -130,8 +130,9 @@ def observe(self, path: Path) -> None: )) self.receipt["session"]["session_id"] = observed - def prepare_checkpoint_refresh(self, review_basis: dict, attempt: Path, stdout: Path) -> None: - """Start once after a qualified current review; recover observed results. + def prepare_checkpoint_refresh(self, review_basis: dict | None, attempt: Path, stdout: Path, + *, pre_replan_basis: dict | None = None) -> None: + """Start once from a qualified checkpoint boundary; recover observed results. An uncertain launch without a native identity remains unresolved. Never dispatch another fresh conversation merely because its reply was lost. @@ -144,8 +145,12 @@ def prepare_checkpoint_refresh(self, review_basis: dict, attempt: Path, stdout: raise ValueError("checkpoint context refresh requires physical absolute attempt/output paths") if not self.binding or not self.binding.get("checkpoint_artifact_root"): raise ValueError("checkpoint context refresh requires a source-bound conversation") + if (review_basis is None) == (pre_replan_basis is None): + raise ValueError("checkpoint refresh requires exactly one qualified boundary basis") refresh = {"checkpoint_sha256": self.binding["checkpoint_artifact_sha256"], - "source_session_id": self.binding["session_id"], "review_basis": review_basis} + "source_session_id": self.binding["session_id"], + **({"review_basis": review_basis} if review_basis is not None else + {"pre_replan_basis": pre_replan_basis})} basis = {"binding": self.binding, "refresh": refresh} with exclusive_file_lock(attempt): if attempt.exists(): diff --git a/loopx/hosts/research_checkpoint.py b/loopx/hosts/research_checkpoint.py index de7f1f6ea5..ed5583c4f9 100644 --- a/loopx/hosts/research_checkpoint.py +++ b/loopx/hosts/research_checkpoint.py @@ -10,6 +10,7 @@ import argparse import hashlib import json +import math import os import time from pathlib import Path, PurePosixPath @@ -134,7 +135,7 @@ def _handoff(env, payload): -def _context_session(env, metadata): +def _context_session(env, metadata, timeout=30): """Read the admitted native branch before publishing captured Todo context.""" from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession from loopx.hosts.codex_checkpoint import native_session_checkpoint @@ -144,7 +145,7 @@ def _context_session(env, metadata): if (not wake.session_id or wake.session_id == native["session_id"] or wake.digest != metadata.get("session", {}).get("session_profile_digest")): raise ValueError("context adoption requires the admitted destination session profile") - observed = native_session_checkpoint(env, wake.session_id, timeout=30, + observed = native_session_checkpoint(env, wake.session_id, timeout=timeout, interrupted_guard=native.get("interrupted_guard"))["metadata"] cursor_key = "interrupted_turn_id" if native.get("interrupted_guard") is not None else "completed_turn_id" for key in ("visible_history_sha256", "turn_ids", cursor_key, "host_version", "config_sha256"): @@ -155,7 +156,7 @@ def _context_session(env, metadata): def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_todo_id: str, todo_id: str, operation_id: str, expected_revision: str, - lease_key: str, lease_version: int) -> dict: + lease_key: str, lease_version: int, deadline: float | None = None) -> dict: """Adopt retained context using the destination's current claim and lease. No Todo is created, completed or transferred, and no historical authority, @@ -163,14 +164,36 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t without operational observation grants. Prepare/adopt each use stable original operation identities. A failed/uncertain call must be read back through that same owner; this adapter never guesses a new identity or starts model work. + Complete artifact/native reads allow up to 90 seconds each, bounded by the + caller's original deadline when supplied. This IO allowance grants no lease + extension and does not alter the typed continuation owner. """ from loopx.control_plane.turn_driver.codex_sessions import select_codex_cli_session from loopx.control_plane.todos.provider_projection import ( settle_canonical_todo_projection, projection_delivery_requires_ack, ) - effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "verify", "root": str(unit), "expected_sha256": digest}) + if deadline is not None and (isinstance(deadline, bool) or not isinstance(deadline, (int, float)) + or not math.isfinite(deadline)): + raise ValueError("context adoption requires a finite original deadline") + + def read_timeout(): + remaining = 90 if deadline is None else deadline - time.time() + if remaining <= 0: + raise TimeoutError("original context adoption deadline expired; inspect the original operation") + return min(90, remaining) + + def verify_artifact(): + effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "verify", "root": str(unit), "expected_sha256": digest}, timeout=read_timeout()) + read_timeout() # A late successful read cannot authorize the next write. + + def read_session(): + session = _context_session(env, metadata, read_timeout()) + read_timeout() + return session + + verify_artifact() metadata = json.loads((unit / "checkpoint.json").read_bytes())["point_metadata"] if not all(isinstance(metadata.get(key), dict) for key in ("loopx_state", "native_session", "research_context")): raise ValueError("checkpoint lacks its canonical Todo or native session context participant") @@ -209,7 +232,7 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t knowledge = rebase_explore_knowledge_events(evidence.get("explore_events", []), source_goal_id=source["goal_id"], destination_goal_id=env["LOOPX_GOAL_ID"], source_ref="checkpoint:" + digest) - session_id = _context_session(env, metadata) + session_id = read_session() bound = select_codex_cli_session(Path(env["LOOPX_RUNTIME_ROOT"]), lineage={ "goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"]}, session_scope="agent") if bound is None or bound["session_id"] != session_id: @@ -227,7 +250,7 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t if (inspect["claimed_by"] != env["LOOPX_AGENT_ID"] or inspect.get("execution_authority", {}).get("allowed") is not True): raise ValueError("destination context revision, owner or execution authority changed") - destination = canonical_context(env, deadline=time.time() + 30)["snapshot"] + destination = canonical_context(env, deadline=time.time() + min(30, read_timeout()))["snapshot"] current = next((row for row in destination["todos"] if row["todo_id"] == todo_id), None) if destination["provider_revision"] != inspect["provider_revision"] or current is None: raise ValueError("destination canonical context moved before preparation") @@ -240,6 +263,7 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t # current note. Changed operation/context/proof must still fail there. # Keep the complete source read in a workspace-relative private artifact. # Never overwrite a moved artifact on an operation retry. + read_timeout() relative = Path(".local/checkpoint-context") / (digest + ".json") target = workspace / relative for directory in (workspace / ".local", target.parent): @@ -259,8 +283,7 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t raise ValueError("destination context artifact already differs") if target.read_bytes() != raw: raise ValueError("destination context artifact readback differs") - effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "verify", "root": str(unit), "expected_sha256": digest}) + verify_artifact() # Keep old context-only request bytes stable for original receipt recovery # across an adapter upgrade. New history is disclosed only when captured. history_guidance = (" Read its research_evidence for prior branches, negative results and vision/stride decisions." @@ -271,22 +294,24 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t "source-revision:" + declared["provider_revision"]], "files_touched": [{"path": relative.as_posix(), "action": "read", "summary": "Complete captured canonical Todo population, including retained notes and rejected approaches."}]} + read_timeout() prepared = _handoff(env, {**common, "action": "prepare", "session_id": metadata["native_session"]["session_id"], "operation_id": operation_id + "-prepare", "expected_provider_revision": expected_revision, "context": note}) if select_codex_cli_session(Path(env["LOOPX_RUNTIME_ROOT"]), lineage={ "goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"]}, session_scope="agent") != bound: raise ValueError("destination session changed; inspect the original prepared context") - if _context_session(env, metadata) != session_id: + if read_session() != session_id: raise ValueError("destination native branch changed; inspect the original prepared context") # Preserve the original prepare commit revision. The outer readback may # already include adoption if its response was lost; using that newer head # would change the adopt request hash and defeat original-receipt recovery. + read_timeout() adopted = _handoff(env, {**common, "action": "adopt", "artifacts": [relative.as_posix()], "operation_id": operation_id + "-adopt", "expected_provider_revision": prepared["result"]["provider_revision"]}) - effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "verify", "root": str(unit), "expected_sha256": digest}) + verify_artifact() if target.read_bytes() != raw: raise ValueError("adopted context artifact changed; inspect the original continuation receipt") + read_timeout() result = settle_canonical_todo_projection(adopted, registry_path=Path(env["LOOPX_REGISTRY"]), runtime_root=Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"]) if result.get("ok") is not True or not projection_delivery_requires_ack(result["projection_delivery"]): @@ -300,9 +325,11 @@ def adopt_todo_context(env: dict[str, str], unit: Path, digest: str, *, source_t if (current_owner["claimed_by"] != env["LOOPX_AGENT_ID"] or current_owner.get("execution_authority", {}).get("allowed") is not True): raise ValueError("destination knowledge adoption lost current execution authority") + read_timeout() knowledge_result = append_explore_result_events( explore_result_log_path(Path(env["LOOPX_RUNTIME_ROOT"]), env["LOOPX_GOAL_ID"]), knowledge, expected_goal_id=env["LOOPX_GOAL_ID"], require_ordered_prefix=True) + read_timeout() return {"context_adopted": True, "session_id": bound["session_id"], "todo_id": todo_id, "artifact": str(target), "context_sha256": declared["sha256"], **({"explore_knowledge": knowledge_result} if knowledge_result is not None else {}), diff --git a/tests/control_plane/test_codex_checkpoint_context_refresh.py b/tests/control_plane/test_codex_checkpoint_context_refresh.py index 23ff9c3598..238bee6921 100644 --- a/tests/control_plane/test_codex_checkpoint_context_refresh.py +++ b/tests/control_plane/test_codex_checkpoint_context_refresh.py @@ -120,3 +120,21 @@ def test_redirected_original_output_cannot_recover_an_identity(tmp_path): with pytest.raises(ValueError, match="physical absolute"): subject.prepare_checkpoint_refresh(review, attempt, original) assert load_codex_cli_session(tmp_path, lineage=subject.lineage, session_scope="agent")["session_id"] == "original" + + +def test_pre_replan_refresh_has_no_completed_decision_and_preserves_recovery(tmp_path): + from benchmark.runtime.checkpoint_continuation import retained_native_history + subject = session(tmp_path) + basis = {"decision_sha256": "d" * 64, "pending_replan": {"capture": True}} + original, retry = tmp_path / "original.jsonl", tmp_path / "retry.jsonl" + subject.prepare_checkpoint_refresh(None, tmp_path / "attempt.json", original, pre_replan_basis=basis) + observed(original, "fresh-before-review") + subject.prepare_checkpoint_refresh(None, tmp_path / "attempt.json", retry, pre_replan_basis=basis) + binding = load_codex_cli_session(tmp_path, lineage=subject.lineage, session_scope="agent") + assert binding["session_id"] == "fresh-before-review" and not retry.exists() + refresh = binding["checkpoint_context_refresh"] + assert "review_basis" not in refresh and refresh["pre_replan_basis"] == basis + assert retained_native_history(binding, {})["turn_ids"] == [] + with pytest.raises(ValueError, match="exactly one"): + session(tmp_path / "invalid").prepare_checkpoint_refresh(basis, tmp_path / "both.json", retry, + pre_replan_basis=basis) diff --git a/tests/control_plane/test_research_context_adoption.py b/tests/control_plane/test_research_context_adoption.py index a3dfffcaa3..1e55ac59a6 100644 --- a/tests/control_plane/test_research_context_adoption.py +++ b/tests/control_plane/test_research_context_adoption.py @@ -119,6 +119,37 @@ def run(*args): adopt_todo_context(env,unit,digest,**args) assert list(outside.iterdir())==[] (workspace/'.local').unlink() + # A complete slow read may finish after the caller's deadline. Neither + # initial verification nor the last pre-prepare verification may authorize + # a continuation write then; immutable staging remains safe to replay. + if lost_response is None and with_knowledge: + from types import SimpleNamespace + from loopx.hosts import research_checkpoint + original_rpc = research_checkpoint.effect_runtime_result + original_handoff = research_checkpoint._handoff + for expire_at in (1, 2): + clock, reads, writes = [100.0], [], [] + def slow_read(method, payload, **kwargs): + if method == "runtime.checkpoint_artifacts" and payload["action"] == "verify": + reads.append(kwargs["timeout"]) + assert kwargs["timeout"] == 40 + result = original_rpc(method, payload, **kwargs) + if len(reads) == expire_at: + clock[0] = 141.0 + return result + return original_rpc(method, payload, **kwargs) + def track_handoff(env, payload): + if payload["action"] != "inspect": + writes.append(payload["action"]) + return original_handoff(env, payload) + with monkeypatch.context() as bounded: + bounded.setattr(research_checkpoint, "time", SimpleNamespace(time=lambda: clock[0])) + bounded.setattr(research_checkpoint, "effect_runtime_result", slow_read) + bounded.setattr(research_checkpoint, "_handoff", track_handoff) + with pytest.raises(TimeoutError, match="original context adoption deadline"): + adopt_todo_context(env, unit, digest, **args, deadline=140) + assert len(reads) == expire_at and not writes + assert run('todo','list','--goal-id','branch','--todo-id',destination_todo)['authority_read']['provider_revision'] == args['expected_revision'] source_run('todo','update','--goal-id','source','--todo-id',source_todo,'--agent-id','source-actor', '--note','Candidate 999 is later evidence excluded from the captured point.', '--update-operation-id','later-source-evidence','--update-expected-provider-revision',source_context['snapshot']['provider_revision'], @@ -378,3 +409,15 @@ def racing_read(*args, **kwargs): monkeypatch.setattr(research_checkpoint, "canonical_context", racing_read) with pytest.raises(ValueError, match="moved during the context read"): research_context(env, deadline=time.time() + 30) + + +@pytest.mark.parametrize("deadline", [float("nan"), float("inf"), True, "later", 0, -1]) +def test_adoption_rejects_invalid_or_expired_deadline_before_io(tmp_path, monkeypatch, deadline): + from loopx.hosts import research_checkpoint + def forbidden(*args, **kwargs): + raise AssertionError("expired/invalid admission must not read artifact or authority") + monkeypatch.setattr(research_checkpoint, "effect_runtime_result", forbidden) + with pytest.raises((ValueError, TimeoutError), match="deadline"): + adopt_todo_context({}, tmp_path / "absent", "digest", source_todo_id="source", + todo_id="destination", operation_id="original", expected_revision="revision", + lease_key="lease", lease_version=1, deadline=deadline)