From 51c3856b0ec37d98f3d3c86c8817ded69c4e9785 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:25:22 +0800 Subject: [PATCH] fix(checkpoint): bound complete context adoption reads by caller deadline Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 8 ++- benchmark/runtime/sforge_warm_start.py | 2 +- loopx/hosts/research_checkpoint.py | 51 ++++++++++++++----- .../test_research_context_adoption.py | 43 ++++++++++++++++ 4 files changed, 90 insertions(+), 14 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index b9ee075c0a..cbd1a7a1e0 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -966,7 +966,13 @@ native registration; subsequent participant checks still verify the retained 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. These allowances grant no extra execution time and +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 diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index d8ff7f23ee..b72e50633f 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -298,7 +298,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/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_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)