From 8c30ce4027b46419eefc323cf9c17df9e4952ca8 Mon Sep 17 00:00:00 2001 From: song Date: Sun, 1 Nov 2026 02:52:06 +0800 Subject: [PATCH] fix(checkpoint): preserve typed capture failure diagnostics Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- benchmark/README.md | 7 + benchmark/edgebench/checkpoint_supervisor.py | 7 + .../tests/test_checkpoint_diagnostics.py | 188 ++++++++++++++++++ .../runtime/checkpoint_artifacts.ts | 4 +- .../test_checkpoint_artifact_transport.py | 10 +- 5 files changed, 212 insertions(+), 4 deletions(-) create mode 100644 benchmark/tests/test_checkpoint_diagnostics.py diff --git a/benchmark/README.md b/benchmark/README.md index bb61cc6b29..057873441b 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -726,6 +726,13 @@ later files with an older native point. Missing paths, archive links, traversal Declare only the intended private code/state stores: a whole home can contain credentials, and neither archive filtering nor a freeze is privacy certification. +Supervisor failure receipts retain the exception-class `error_kind`. Failures +from the managed Effect runtime additionally expose `runtime_failure.error_kind` +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. + This explicit command produces an **inert joint file snapshot**, still `research_point_qualified=false`. It does not snapshot process memory, freeze external writers or activate canonical research state. Ordinary native restore diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py index 7477f9820f..15d27d94f0 100644 --- a/benchmark/edgebench/checkpoint_supervisor.py +++ b/benchmark/edgebench/checkpoint_supervisor.py @@ -11,6 +11,8 @@ import time from pathlib import Path +from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError + from .checkpoint_controller import _controller from .checkpoint_worker import _archive, capture_worker, qualify_export_storage @@ -169,6 +171,11 @@ def _run(self, container): # Optional capture failure resumes the original source; # credentials and raw provider errors never enter it. response["error_kind"] = type(error).__name__ + if isinstance(error, EffectRuntimeRemoteError): + response["runtime_failure"] = { + "error_kind": error.error_kind, + "diagnostic_code": error.diagnostic_code, + } diagnostic = self.destination / (nonce + ".error.txt") diagnostic.write_text(str(error).replace(self.token, "[redacted]")[:2000]) diagnostic.chmod(0o600) diff --git a/benchmark/tests/test_checkpoint_diagnostics.py b/benchmark/tests/test_checkpoint_diagnostics.py new file mode 100644 index 0000000000..e70df2773b --- /dev/null +++ b/benchmark/tests/test_checkpoint_diagnostics.py @@ -0,0 +1,188 @@ +"""Real Effect failures through worker cleanup and supervisor reply transport. + +Docker IO is synthetic; this does not qualify a real Docker or native provider. +""" +from __future__ import annotations + +import hashlib +import io +import json +import stat +import tarfile +import tempfile +import time +from types import SimpleNamespace + +import docker +import pytest + +from benchmark.edgebench import checkpoint_supervisor as supervisor_module +from benchmark.edgebench import checkpoint_worker as worker_module +from loopx.control_plane.effect_runtime import effect_runtime_result, restart_effect_runtime + + +@pytest.fixture(scope="module", autouse=True) +def isolated_runtime(tmp_path_factory): + temporary = tmp_path_factory.mktemp("checkpoint-diagnostic-runtime") + with pytest.MonkeyPatch.context() as patch: + patch.setenv("LOOPX_USAGE_PING", "0") + for variable in ("TMPDIR", "TEMP", "TMP"): + patch.setenv(variable, str(temporary)) + patch.setattr(tempfile, "tempdir", str(temporary)) + yield + restart_effect_runtime(wait_for_exit=True) + + +@pytest.mark.parametrize("fault,exception_kind,runtime_kind,code", [ + ("expired", "EffectRuntimeRejected", "request_rejected", "checkpoint_capture_window_closed"), + ("missing", "EffectRuntimePermanentIOError", "io_permanent", "io_not_found"), + ("directory", "EffectRuntimePermanentIOError", "io_permanent", "io_is_directory"), + ("unexpected", "EffectRuntimeInternalError", "internal_failure", "unexpected_handler_error"), + ("untyped", "ValueError", None, None), +]) +def test_failed_capture_replies_once_after_releasing_source( + tmp_path, monkeypatch, fault, exception_kind, runtime_kind, code, +): + events = [] + replies = [] + nonce = "a" * 32 + token, admin = "synthetic-worker-secret", "synthetic-admin-secret" + candidate = tmp_path / "candidate" + candidate.write_bytes(b"original source bytes\n") + source = [{"id": "code", "source_path": str(candidate)}] + observed = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "probe", "components": source}, retry_safe=False) + ready = tmp_path / "ready.json" + ready.write_text(json.dumps({ + "capture_id": nonce, "components": [{"id": "code", "source_path": "/candidate"}], + "point_metadata": {}, "source_sha256": observed["source_sha256"], + "source_witness": observed["source_witness"], "deadline_epoch": time.time() + 60, + })) + bank = tmp_path / "bank" + bank.mkdir() + controller = tmp_path / "controller.json" + controller.write_text(json.dumps({ + "url": "http://127.0.0.1:9", "admin_secret": admin, "bank_root": str(bank), + })) + controller.chmod(0o600) + + class Worker: + id = "b" * 64 + + def __init__(self): + self.attrs = {"State": {"Running": True, "Paused": False}, "Image": "synthetic-image"} + + def reload(self): + pass + + def pause(self): + events.append("pause") + self.attrs["State"]["Paused"] = True + + def unpause(self): + events.append("unpause") + self.attrs["State"]["Paused"] = False + + def put_archive(self, directory, raw): + assert self.attrs["State"] == {"Running": True, "Paused": False} + assert directory == "/logs/agent/checkpoint-controller" + events.append("reply_transfer") + with tarfile.open(fileobj=io.BytesIO(raw)) as archive: + assert archive.getnames() == [nonce + ".pending"] + replies.append(json.load(archive.extractfile(nonce + ".pending"))) + return True + + def exec_run(self, command): + assert command == ["/bin/mv", directory + "/" + nonce + ".pending", + directory + "/" + nonce + ".json"] + assert self.attrs["State"] == {"Running": True, "Paused": False} + events.append("reply_publish") + return SimpleNamespace(exit_code=0) + + directory = "/logs/agent/checkpoint-controller" + worker = Worker() + + def get_container(identity): + assert identity == worker.id + return worker + + client = SimpleNamespace(api=SimpleNamespace(timeout=60), + containers=SimpleNamespace(get=get_container), close=lambda: events.append("client_close")) + monkeypatch.setattr(docker, "from_env", lambda **kwargs: client) + monkeypatch.setattr(worker_module, "_worker_lock", lambda *args: tmp_path / "worker.lock") + monkeypatch.setattr(supervisor_module, "_archive", lambda *args, **kwargs: ready) + + def held_archive(*args, **kwargs): + assert worker.attrs["State"]["Paused"] + return ready + + def source_archive(*args, **kwargs): + assert worker.attrs["State"]["Paused"] + return candidate + + monkeypatch.setattr(worker_module, "_archive", held_archive) + monkeypatch.setattr(worker_module, "_archive_component", source_archive) + + def native_capture(*args, **kwargs): + pytest.fail("A failed source probe must not reach native capture") + + monkeypatch.setattr(worker_module, "capture", native_capture) + malformed_root = tmp_path / "synthetic-private-path" + malformed_root.mkdir() + if fault == "directory": + (malformed_root / "checkpoint.json").mkdir() + else: + (malformed_root / "checkpoint.json").write_text("invalid JSON " + token + " " + admin) + + def failed_probe(method, request, **kwargs): + assert method == "runtime.checkpoint_artifacts" and request["action"] == "probe" + assert kwargs["retry_safe"] is False + assert worker.attrs["State"]["Paused"] + events.append("probe") + if fault == "untyped": + error = ValueError("provider body " + token) + # Attributes alone must not make an arbitrary provider exception trusted. + error.error_kind, error.diagnostic_code = admin, token + raise error + if fault == "expired": + request = {**request, "deadline_epoch_ms": 0} + elif fault == "missing": + request = {"action": "probe", "components": [ + {"id": "code", "source_path": str(tmp_path / "missing")}]} + else: + request = {"action": "read_basis", "root": str(malformed_root), "expected_sha256": "0" * 64} + return effect_runtime_result(method, request, **kwargs) + + monkeypatch.setattr(worker_module, "effect_runtime_result", failed_probe) + points = tmp_path / "points" + supervisor = supervisor_module.CheckpointSupervisor(controller, points, retention_enabled=False) + supervisor.bind({"SFORGE_TOKEN": token}) + points.mkdir(mode=0o700) + supervisor.container = worker + # Read the same handoff a second time: production's seen set must prevent a + # second attempt, then stop at its ordinary polling wait without sleeping. + monkeypatch.setattr(supervisor.stop, "wait", lambda *args: supervisor.stop.set()) + supervisor._run(worker) + supervisor.close() + + assert supervisor.failure is None + assert worker.attrs["State"] == {"Running": True, "Paused": False} + assert events == ["pause", "probe", "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() + assert not list(points.glob(".worker-capture-*")) + receipt = json.loads((points / (nonce + ".json")).read_bytes()) + assert replies == [receipt] + expected = {"request_sha256": hashlib.sha256(ready.read_bytes()).hexdigest(), + "ok": False, "error_kind": exception_kind} + if runtime_kind is not None: + expected["runtime_failure"] = {"error_kind": runtime_kind, "diagnostic_code": code} + assert receipt == expected + for private in (token, admin, str(tmp_path), "provider body", "invalid JSON"): + assert private not in json.dumps(receipt) + diagnostic = points / (nonce + ".error.txt") + assert stat.S_IMODE(diagnostic.stat().st_mode) == 0o600 + assert token not in diagnostic.read_text() + if fault == "unexpected": + assert diagnostic.read_text() == "Effect runtime handler failed unexpectedly" diff --git a/loopx/control_plane/runtime/checkpoint_artifacts.ts b/loopx/control_plane/runtime/checkpoint_artifacts.ts index 345ad044f6..66a6d02067 100644 --- a/loopx/control_plane/runtime/checkpoint_artifacts.ts +++ b/loopx/control_plane/runtime/checkpoint_artifacts.ts @@ -405,7 +405,9 @@ export async function checkpointArtifacts(value: unknown): Promise { const deadline = request.deadline_epoch_ms; if (deadline !== undefined && (typeof deadline !== "number" || !Number.isFinite(deadline))) throw new Error("checkpoint deadline must be finite"); function withinDeadline(): void { - if (typeof deadline === "number" && Date.now() >= deadline) throw new Error("checkpoint capture window has closed"); + if (typeof deadline === "number" && Date.now() >= deadline) { + throw new EffectRuntimeRequestError("checkpoint capture window has closed", "checkpoint_capture_window_closed"); + } } withinDeadline(); if (request.action === "probe") { diff --git a/tests/control_plane/test_checkpoint_artifact_transport.py b/tests/control_plane/test_checkpoint_artifact_transport.py index 10c214d714..6bb1f4a3e3 100644 --- a/tests/control_plane/test_checkpoint_artifact_transport.py +++ b/tests/control_plane/test_checkpoint_artifact_transport.py @@ -9,6 +9,7 @@ from loopx.control_plane.effect_runtime import ( MAX_REQUEST_BYTES, MAX_RESPONSE_BYTES, + EffectRuntimeRejected, EffectRuntimeRemoteError, EffectRuntimeResponseAmbiguous, effect_runtime_result, @@ -98,8 +99,8 @@ def test_large_rpc_allowance_cannot_extend_artifact_capture_deadline(tmp_path): components = [{"id": "code", "source_path": str(source)}] observed = effect_runtime_result("runtime.checkpoint_artifacts", { "action": "probe", "components": components}, large_local_snapshot=True) - # The same valid request succeeds with a future artifact deadline. Do not - # infer the rejection cause from internal error prose hidden by the bridge. + # The same valid request succeeds with a future artifact deadline. The + # known expired-window refusal must survive the real typed RPC boundary. live = effect_runtime_result("runtime.checkpoint_artifacts", { "action": "capture", "components": components, "destination": str(tmp_path / "live"), "point_metadata": {}, @@ -108,10 +109,13 @@ def test_large_rpc_allowance_cannot_extend_artifact_capture_deadline(tmp_path): }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True) assert live["source_stability_verified"] unit = tmp_path / "expired" - with pytest.raises(EffectRuntimeRemoteError): + with pytest.raises(EffectRuntimeRejected) as rejected: effect_runtime_result("runtime.checkpoint_artifacts", { "action": "capture", "components": components, "destination": str(unit), "point_metadata": {}, "expected_source_sha256": observed["source_sha256"], "deadline_epoch_ms": 0, }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True) + assert rejected.value.error_kind == "request_rejected" + assert rejected.value.diagnostic_code == "checkpoint_capture_window_closed" + assert str(rejected.value) == "checkpoint capture window has closed" assert not unit.exists()