fix(benchmark): identify missing checkpoint components

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 截止时间，也不创建或跟随外来根目录。

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,

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", {

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

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

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)

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

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

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"]