From 6929a2447080be10ba33496f67fa2cf5a7040312 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:36:18 +0800 Subject: [PATCH] fix(benchmark): identify missing checkpoint components Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 7 ++++++ benchmark/edgebench/checkpoint_supervisor.py | 6 +++-- benchmark/edgebench/checkpoint_worker.py | 25 +++++++++++++++++-- benchmark/edgebench/terminal_capture.py | 4 ++- benchmark/runtime/checkpoint_handoff.py | 8 +++++- .../tests/test_checkpoint_diagnostics.py | 10 +++++++- benchmark/tests/test_checkpoint_handoff.py | 13 ++++++++-- benchmark/tests/test_checkpoint_worker.py | 11 +++++++- benchmark/tests/test_terminal_preservation.py | 6 +++++ 9 files changed, 80 insertions(+), 10 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 8d6d26895f..fdd4141cda 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -823,6 +823,10 @@ and `runtime_failure.diagnostic_code`, without copying the error message or raw provider response into the worker reply. An expired artifact capture window is `request_rejected` / `checkpoint_capture_window_closed`. These diagnostics do not retry the capture, extend its deadline or grant execution authority. +When Docker cannot find a declared component during export, the supervisor and +terminal receipts include its declared `missing_component_id`. The worker reports +that ID when it matches the original request. The capture still fails and the +existing source-retention rules apply. This explicit command produces an **inert joint file snapshot**, still `research_point_qualified=false`. It does not snapshot process memory, freeze @@ -1149,6 +1153,9 @@ worker 保留原 guard，controller 暂停全部容器写入者，核对仍有 文件 witness，再封存原生评测机会、incumbent 及剩余预算。源变化、过期或失去 交接会拒绝；可恢复的捕获失败不取消原研究，也不得改用稍后的状态。解除暂停后， 源会话继续使用原 packet 和 deadline。 +Docker 导出找不到某个已声明组件时，supervisor 和终态回执会记录其 +`missing_component_id`。worker 在确认 ID 与原请求匹配后报告该 ID。 +捕获仍失败，来源仍按原规则保留。 原生交接先接收并读取一条有界请求，再校验捕获根目录的物理路径。目录缺失或为 符号链接时关闭该连接，原来源可纠正路径后经同一监听器继续；拒绝不捕获、不释放 旧 packet，不延长原 wake 截止时间，也不创建或跟随外来根目录。 diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py index 3bbe88134e..27f0b1a672 100644 --- a/benchmark/edgebench/checkpoint_supervisor.py +++ b/benchmark/edgebench/checkpoint_supervisor.py @@ -15,7 +15,9 @@ from benchmark.runtime.layout import _ACTIVE_PHASE_DEADLINE, _PHASE_DEADLINE from .checkpoint_controller import _controller -from .checkpoint_worker import _archive, _guest_path, _validate_request, capture_worker, qualify_export_storage +from .checkpoint_worker import ( + _archive, _guest_path, _validate_request, capture_failure_fields, capture_worker, qualify_export_storage, +) class CheckpointSupervisor: @@ -202,7 +204,7 @@ def _run(self, container): except Exception as error: # Optional capture failure resumes the original source; # credentials and raw provider errors never enter it. - response["error_kind"] = type(error).__name__ + response.update(capture_failure_fields(error)) if isinstance(error, EffectRuntimeRemoteError): response["runtime_failure"] = { "error_kind": error.error_kind, diff --git a/benchmark/edgebench/checkpoint_worker.py b/benchmark/edgebench/checkpoint_worker.py index 927f0d74d6..92d6a4ea79 100644 --- a/benchmark/edgebench/checkpoint_worker.py +++ b/benchmark/edgebench/checkpoint_worker.py @@ -26,6 +26,21 @@ from .checkpoint_controller import _controller, _physical, _remaining, _phase_budget, capture +_COMPONENT_ID = re.compile(r"[A-Za-z][A-Za-z0-9_.-]{0,63}") + + +def capture_failure_fields(error: Exception) -> dict[str, str]: + """Keep a bounded diagnostic for a declared component's failed Docker read.""" + fields = {"error_kind": type(error).__name__} + component_id = getattr(error, "checkpoint_component_id", None) + if isinstance(component_id, str) and _COMPONENT_ID.fullmatch(component_id): + from docker.errors import NotFound + + if isinstance(error, NotFound): + fields["missing_component_id"] = component_id + return fields + + def qualify_export_storage(destination: Path) -> None: """Check filename fidelity before registration, without reserving a run. @@ -72,7 +87,7 @@ def _validate_request(value: dict) -> dict: for entry in value["components"]: if (not isinstance(entry, dict) or set(entry) != {"id", "source_path"} or not isinstance(entry["id"], str) or - not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,63}", entry["id"]) or entry["id"] in ids): + not _COMPONENT_ID.fullmatch(entry["id"]) or entry["id"] in ids): raise ValueError("Worker components require unique IDs") ids.add(entry["id"]) paths.append(_guest_path(entry["source_path"])) @@ -334,7 +349,13 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str, deadline = min(deadline, time.monotonic() + prepared["deadline_epoch"] - time.time()) for entry in request["components"]: client.api.timeout = _remaining(deadline) - copied = _archive_component(container, entry["source_path"], stage / "workers" / entry["id"], deadline=deadline) + try: + copied = _archive_component( + container, entry["source_path"], stage / "workers" / entry["id"], deadline=deadline) + except docker.errors.NotFound as error: + # Keep the provider exception for direct capture callers. + error.checkpoint_component_id = entry["id"] + raise components.append({"id": entry["id"], "source_path": str(copied)}) if prepared is not None: observed = effect_runtime_result("runtime.checkpoint_artifacts", { diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 320775df5b..c8ff3a6a1d 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/benchmark/edgebench/terminal_capture.py @@ -150,7 +150,9 @@ def finish(self, execution): self.receipt.update(capture_succeeded=True, capture=point) except Exception as error: # No remote message, output, account information or token is retained. - self.receipt["error_kind"] = type(error).__name__ + from .checkpoint_worker import capture_failure_fields + + self.receipt.update(capture_failure_fields(error)) finally: # Native final-archive extraction and grading drain occur before its # cleanup callback. Stop here so orphaned work cannot run through diff --git a/benchmark/runtime/checkpoint_handoff.py b/benchmark/runtime/checkpoint_handoff.py index 19a77dc6a0..c8afb53253 100644 --- a/benchmark/runtime/checkpoint_handoff.py +++ b/benchmark/runtime/checkpoint_handoff.py @@ -77,7 +77,13 @@ def await_controller_capture(root, components, metadata, probe, deadline, *, pha if result.get("request_sha256") != hashlib.sha256(raw).hexdigest(): raise ValueError("checkpoint controller reply belongs to another boundary") if result.get("ok") is not True: - raise ValueError("controller checkpoint failed: " + str(result.get("error_kind"))) + error_kind = result.get("error_kind") + message = f"controller checkpoint failed: {error_kind}" + missing = result.get("missing_component_id") + if (error_kind == "NotFound" and isinstance(missing, str) and + any(entry["id"] == missing for entry in components)): + message += f" (missing component {missing})" + raise ValueError(message) return result["capture"] time.sleep(.05) raise TimeoutError("controller did not publish a sealed checkpoint before the completion deadline") diff --git a/benchmark/tests/test_checkpoint_diagnostics.py b/benchmark/tests/test_checkpoint_diagnostics.py index 2cd5c92bdf..b76259e08b 100644 --- a/benchmark/tests/test_checkpoint_diagnostics.py +++ b/benchmark/tests/test_checkpoint_diagnostics.py @@ -36,6 +36,7 @@ def isolated_runtime(tmp_path_factory): @pytest.mark.parametrize("fault,exception_kind,runtime_kind,code", [ ("expired", "EffectRuntimeRejected", "request_rejected", "checkpoint_capture_window_closed"), ("missing", "EffectRuntimePermanentIOError", "io_permanent", "io_not_found"), + ("archive_missing", "NotFound", None, None), ("directory", "EffectRuntimePermanentIOError", "io_permanent", "io_is_directory"), ("unexpected", "EffectRuntimeInternalError", "internal_failure", "unexpected_handler_error"), ("untyped", "ValueError", None, None), @@ -118,6 +119,9 @@ def held_archive(*args, **kwargs): def source_archive(*args, **kwargs): assert worker.attrs["State"]["Paused"] + if fault == "archive_missing": + events.append("archive_missing") + raise docker.errors.NotFound("provider body " + token + " " + str(tmp_path)) return candidate monkeypatch.setattr(worker_module, "_archive", held_archive) @@ -143,6 +147,7 @@ def failed_probe(method, request, **kwargs): error = ValueError("provider body " + token) # Attributes alone must not make an arbitrary provider exception trusted. error.error_kind, error.diagnostic_code = admin, token + error.checkpoint_component_id = "code" raise error if fault == "expired": request = {**request, "deadline_epoch_ms": 0} @@ -174,7 +179,8 @@ def failed_probe(method, request, **kwargs): assert supervisor.failure is None assert storage_probes == [points, points / nonce] assert worker.attrs["State"] == {"Running": True, "Paused": False} - assert events == ["pause", "probe", "unpause", "client_close", "reply_transfer", "reply_publish"] + failure_step = "archive_missing" if fault == "archive_missing" else "probe" + assert events == ["pause", failure_step, "unpause", "client_close", "reply_transfer", "reply_publish"] assert candidate.read_bytes() == b"original source bytes\n" assert not (points / nonce).exists() and not (bank / nonce).exists() assert (bank / ".loopx-worker-attempts" / nonce).is_dir() @@ -185,6 +191,8 @@ def failed_probe(method, request, **kwargs): "ok": False, "error_kind": exception_kind} if runtime_kind is not None: expected["runtime_failure"] = {"error_kind": runtime_kind, "diagnostic_code": code} + if fault == "archive_missing": + expected["missing_component_id"] = "code" assert receipt == expected for private in (token, admin, str(tmp_path), "provider body", "invalid JSON"): assert private not in json.dumps(receipt) diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py index e395d34bc6..de6a4cf42d 100644 --- a/benchmark/tests/test_checkpoint_handoff.py +++ b/benchmark/tests/test_checkpoint_handoff.py @@ -124,7 +124,8 @@ def test_completed_attempt_survives_artifact_index_permission_recovery(tmp_path, assert not list(index.iterdir()) and witness.read_bytes() == original -@pytest.mark.parametrize("response", ["captured", "failed", "foreign", "timeout"]) +@pytest.mark.parametrize("response", ["captured", "failed", "missing_component", + "foreign_component", "foreign", "timeout"]) def test_original_request_only_and_cleanup_after_missed_point(tmp_path, response): mailbox = tmp_path / "checkpoint-controller" ready = mailbox / "ready.json" @@ -141,6 +142,9 @@ def controller(): result = {"request_sha256": hashlib.sha256(raw).hexdigest(), "ok": response == "captured", "error_kind": "SyntheticFailure", "capture": {"research_point_qualified": False, "sha256": "retained"}} + if response in {"missing_component", "foreign_component"}: + result["error_kind"] = "NotFound" + result["missing_component_id"] = "code" if response == "missing_component" else "other" if response == "foreign": result["request_sha256"] = "another-point" pending = mailbox / "reply.tmp" @@ -154,8 +158,13 @@ def controller(): if response == "captured": assert await_controller_capture(*args)["sha256"] == "retained" else: - with pytest.raises(TimeoutError if response == "timeout" else ValueError): + with pytest.raises(TimeoutError if response == "timeout" else ValueError) as error: await_controller_capture(*args, phase_deadline=args[-1]) + if response in {"missing_component", "foreign_component"}: + expected = "controller checkpoint failed: NotFound" + if response == "missing_component": + expected += " (missing component code)" + assert str(error.value) == expected finally: thread.join(3) assert not thread.is_alive() and not ready.exists() and len(observed) == 1 diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index 2d540c1855..55bef508d5 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/benchmark/tests/test_checkpoint_worker.py @@ -114,7 +114,7 @@ def test_archive_keeps_private_directory_modes_without_special_bits(writer, tmp_ assert (copied / "child").stat().st_mode & 0o7777 == 0o700 -@pytest.mark.parametrize("fault", [None, "source_moved", "expired", "read_race", "late_native", +@pytest.mark.parametrize("fault", [None, "source_moved", "source_missing", "expired", "read_race", "late_native", "retention", "retention_default", "retention_off", "retention_pin", "retention_failure", "continued_clock"]) def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writer, tmp_path, monkeypatch, fault): from loopx.control_plane.effect_runtime import effect_runtime_result @@ -145,6 +145,8 @@ def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writ "deadline_epoch": time.time() + (-1 if fault == "expired" else 8 if fault == "late_native" else 30)} if fault == "source_moved": assert container.exec_run(["python", "-c", "from pathlib import Path; Path('/candidate').write_text('later')"]).exit_code == 0 + if fault == "source_missing": + assert container.exec_run(["rm", "/candidate"]).exit_code == 0 assert container.exec_run(["python", "-c", "from pathlib import Path; " "Path('/logs/agent/checkpoint-controller/ready.json').write_text(" + repr(json.dumps(request)) + ")"]).exit_code == 0 if fault == "read_race": @@ -232,6 +234,13 @@ def with_old_point(*args, **kwargs): assert (tmp_path / "points/retention-error.json").exists() else: assert not unit.exists() + if fault == "source_missing": + reply = container.exec_run(["cat", "/logs/agent/checkpoint-controller/" + nonce + ".json"]) + assert reply.exit_code == 0 + assert json.loads(reply.output) == response + assert response["error_kind"] == "NotFound" + assert response["missing_component_id"] == "code" + assert b"/candidate" not in reply.output if fault != "late_native": assert not (Path(json.loads(files[0].read_bytes())["bank_root"]) / nonce).exists() container.reload() diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index 3558631d1e..638f3e7e3d 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/benchmark/tests/test_terminal_preservation.py @@ -326,6 +326,12 @@ def registered(token): assert receipt["source_retained"] and receipt["source_stopped"] assert not receipt["restore_qualified"] assert receipt["capture_succeeded"] is not capture_failure + if capture_failure: + assert receipt["error_kind"] == "NotFound" + assert receipt["missing_component_id"] == "missing" + assert "/missing-participant" not in json.dumps(receipt) + else: + assert "missing_component_id" not in receipt assert len(grades) == 1 # Saving the tail did not create an evaluation opportunity. if not capture_failure: point = supervisor.destination / receipt["capture_id"]