Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions benchmark/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -674,6 +674,16 @@ the source continues from the original packet and deadline. Optional capture
failure records a missed point, never a later replacement or a new model/eval budget.
A barrier that cannot be released remains an explicit controller recovery problem.

The explicitly enabled full controller capture hooks now share a 90-second window for complete
native/research preparation, stability checks and live-source capture (previously
30 seconds). Both hooks cap it by the original trial phase; the controller keeps
that same absolute deadline. After releasing the source, it may spend up to 30
more seconds sealing its immutable exports, also within the original phase. This
allowance changes neither the trial budget nor cooldown, completeness or retry
rules. A missed point remains missed; direct controller commands retain their
caller-supplied timeout, and other capture declarations keep their 30-second
window. Capture remains off unless explicitly enabled.

A real native CLI/worker plus isolated Docker/native-HTTP qualification covers the
mid-wake path, complete canonical context, conserved evaluator opportunity and
original source continuation. It uses a deterministic local model and synthetic
Expand Down Expand Up @@ -897,6 +907,12 @@ worker 保留原 guard,controller 暂停全部容器写入者,核对仍有
文件 witness,再封存原生评测机会、incumbent 及剩余预算。源变化、过期或失去
交接会拒绝;可恢复的捕获失败不取消原研究,也不得改用稍后的状态。解除暂停后,
源会话继续使用原 packet 和 deadline。
显式开启的两处完整 controller 捕获 hook 共用 90 秒窗口(原为 30 秒),覆盖完整原生/研究上下文
准备、一致性复核与 live-source 捕获,并受原试验 phase 截止时间约束。controller
沿用同一个绝对截止时间;解除暂停后,只能在原 phase 内再用至多 30 秒封存静态副本。
这不增加试验预算、冷却或机会,不改变完整性及首次尝试规则;丢失的点仍不能补记,
直接 controller 命令仍受调用者给定的 timeout 约束,其他捕获声明保持 30 秒窗口。
未显式开启时仍不捕获。
自动 supervisor 的准备截止时间仍约束暂停、文件导出及原生捕获。成功解除暂停后,
controller 只对自己持有的静态导出副本继续校验、封存,使用已捕获的原 Host phase
截止时间;该截止时间同时传给类型化产物 owner,过期不能发布。源可以继续工作,
Expand Down
4 changes: 3 additions & 1 deletion benchmark/runtime/checkpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from loopx.hosts.research_checkpoint import (
_participant_id, native_binding_participant, research_participants, verify_research_participants,
)
from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS


def normalize_checkpoint_request_path(value: str) -> str:
Expand Down Expand Up @@ -246,7 +247,8 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
started = time.monotonic()
try:
phase_deadline = env.get("LOOPX_PHASE_DEADLINE_EPOCH")
deadline = time.time() + 30
window = CHECKPOINT_CAPTURE_SECONDS if supplied_request and supplied_request[0].get("controller_handoff") else 30
deadline = time.time() + window
if phase_deadline is not None:
original_deadline = float(phase_deadline)
if not math.isfinite(original_deadline):
Expand Down
5 changes: 5 additions & 0 deletions benchmark/runtime/checkpoint_handoff.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,11 @@
import uuid


# One shared budget for explicitly enabled full controller capture, including complete
# native/research preparation and the controller's live-source export. Mature
# histories can exhaust 30 seconds before the writer barrier even starts. The
# original phase still caps this clock; handoff/retries cannot renew it.
CHECKPOINT_CAPTURE_SECONDS = 90
CONTROLLER_SEAL_SECONDS = 30


Expand Down
4 changes: 3 additions & 1 deletion benchmark/runtime/checkpoint_wake.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

from loopx.control_plane.effect_runtime import effect_runtime_result
from .checkpoint import _mark_replan_attempt, _participant_id, held_quota_decision, read_checkpoint_request
from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS


class CheckpointWake:
Expand Down Expand Up @@ -103,7 +104,8 @@ def _capture(self, held, deadline):
"decision_sha256": held["sha256"],
"capture_order": "after_native_interruption",
"writer_barrier": "owned_native_process_group_only"}
window = min(time.time() + 30, time.time() + deadline - time.monotonic())
capture_seconds = CHECKPOINT_CAPTURE_SECONDS if self.request.get("controller_handoff") else 30
window = min(time.time() + capture_seconds, time.time() + deadline - time.monotonic())
if window <= time.time():
raise TimeoutError("original wake budget expired before capture")
components = list(self.request["components"])
Expand Down
77 changes: 77 additions & 0 deletions benchmark/tests/test_checkpoint_handoff.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,83 @@ def test_controller_completion_requires_finite_clocks(capture, phase):
controller_completion_deadline(capture, phase)


@pytest.mark.parametrize("hook", ["settled", "interrupted"])
@pytest.mark.parametrize("phase_seconds", [None, 40, 20])
def test_capture_preparation_uses_one_window_without_renewing_phase(tmp_path, monkeypatch, hook, phase_seconds):
"""A mature witness may take >30s; that grants no new phase or retry clock."""
from contextlib import contextmanager
from types import SimpleNamespace
from benchmark.runtime import checkpoint, checkpoint_wake

clock = [100.0]
timer = SimpleNamespace(time=lambda: clock[0], monotonic=lambda: clock[0])
monkeypatch.setattr(checkpoint, "time", timer)
monkeypatch.setattr(checkpoint_wake, "time", timer)
code = tmp_path / "code"
code.write_text("retained source")
env = {"LOOPX_CHECKPOINT_REQUEST": "explicit declaration", "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"),
"LOOPX_CLI": "loopx", "LOOPX_PROJECT": str(tmp_path),
"LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), "LOOPX_REGISTRY": str(tmp_path / "registry"),
"LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent"}
if phase_seconds is not None:
env["LOOPX_PHASE_DEADLINE_EPOCH"] = str(100 + phase_seconds)
request = {"destination": str(tmp_path / "checkpoints" / "point"),
"components": [{"id": "code", "source_path": str(code)}], "point_metadata": {},
"controller_handoff": True, "loopx_state": True, "native_session": True, "evaluation_opportunity": True}
receipt = {"ok": True, "stage": "execute", "mode": "heartbeat", "turn_id": "turn",
"session": {"session_id": "native-session"}}
binding = SimpleNamespace(session_id="native-session", goal_ref="source", digest="profile")
monkeypatch.setattr(checkpoint, "research_participants", lambda env, components, *a: (components, {}))
monkeypatch.setattr(checkpoint, "verify_research_participants", lambda *a: None)
monkeypatch.setattr(checkpoint, "native_binding_participant", lambda env, components, *a: (components, binding))
monkeypatch.setattr("benchmark.runtime.sessions.BenchmarkSessionWake", lambda *a, **k: binding)
monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_history", lambda *a, **k: {})
@contextmanager
def reader(*a, **k):
yield object()
monkeypatch.setattr("loopx.hosts.codex_checkpoint.native_checkpoint_reader", reader)
monkeypatch.setattr("loopx.hosts.codex_checkpoint.native_session_checkpoint",
lambda *a, **k: {"components": [], "metadata": {"store_members": []}})
deadlines = []
def artifact(method, params, **kwargs):
deadline = params["deadline_epoch_ms"] / 1000
deadlines.append(deadline)
if params["action"] == "probe":
clock[0] += 35 # Complete retained witness, rather than trimming history.
if clock[0] >= deadline:
raise TimeoutError("original capture window expired")
return {"source_sha256": "witness", "source_witness": {}, "sha256": "sealed-original"}
def controller(root, components, metadata, probe, deadline, **kwargs):
deadlines.append(deadline)
if clock[0] >= deadline:
raise TimeoutError("original capture window expired")
return {"sha256": "sealed-original"}
monkeypatch.setattr("benchmark.runtime.checkpoint_handoff.await_controller_capture", controller)
monkeypatch.setattr(checkpoint, "effect_runtime_result", artifact)
monkeypatch.setattr(checkpoint_wake, "effect_runtime_result", artifact)
if hook == "settled":
observed = SimpleNamespace(progress={"state": "settled"},
identity=SimpleNamespace(value=SimpleNamespace(effect_id="original-effect")))
monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", lambda *a, **k: observed)
checkpoint.capture_after_wake(env, receipt, supplied_request=(request, b"declared"))
assert receipt["checkpoint"]["captured"] is (phase_seconds != 20)
assert receipt["ok"] is True # An optional miss cannot replay successful work.
else:
wake = checkpoint_wake.CheckpointWake(env, receipt)
wake.request, wake.raw = request, b"declared"
held = {"intent": {"obligation_id": "replan-fixture"}, "sha256": "original-guard",
"path": code, "quota": {"heartbeat_receipt": {"turn_instance_id": "turn"}}}
phase = float("inf") if phase_seconds is None else 100 + phase_seconds
if phase_seconds == 20:
with pytest.raises(TimeoutError, match="original capture window"):
wake._capture(held, phase)
else:
assert wake._capture(held, phase)["sha256"] == "sealed-original"
expected = 190 if phase_seconds is None else 100 + phase_seconds
assert deadlines and all(deadline == expected for deadline in deadlines)
assert code.read_text() == "retained source"


@pytest.mark.skipif(sys.platform != "linux", reason="Qualifies the Linux writer process group")
@pytest.mark.parametrize("use_selector,closeout,settlement_only", [
(False, False, False), (True, False, False), (True, True, False), (True, False, True),
Expand Down
Loading