fix(checkpoint): preserve typed capture failure diagnostics

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

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)

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"

loopx/control_plane/runtime/checkpoint_artifacts.ts

@@ -405,7 +405,9 @@ export async function checkpointArtifacts(value: unknown): Promise<JsonObject> {
   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") {

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()