diff --git a/docs/components/xpois.md b/docs/components/xpois.md
index c6d76732..8f132cd0 100644
--- a/docs/components/xpois.md
+++ b/docs/components/xpois.md
@@ -649,6 +649,73 @@ backend. `--max-workers`, `--worker-timeout-sec`, and
 For a single-node launch, use `-s` instead of `-m -N 2 -w slurm`, while
 retaining `-t tcp -o tcp` to select TCP transport.
 
+### Repeat a batch in persistent workers
+
+Add `--warmup-rounds` or `--measure-rounds` to the same `fit-batch` command
+to measure repeated passes over the manifest. For example, the Dragon wrapper
+can run one warmup and three measured rounds on a single node:
+
+```bash
+.venv/bin/dragon -s -t tcp -o tcp examples/xpois/dragon_batch.py \
+  --backend cupy --manifest /shared/manifests/fixed-32.yaml \
+  --output-dir /shared/results/xpois-dragon --name repeated-dragon \
+  --max-workers 4 --warmup-rounds 1 --measure-rounds 3
+```
+
+Use the same two flags with a collective MPI launch:
+
+```bash
+mpirun -n 4 --map-by slot --bind-to none -x CUDA_VISIBLE_DEVICES \
+  .venv/bin/cuphoton-openmpi-rank-exec -- \
+  .venv/bin/cuphoton xpois fit-batch --executor mpi \
+  --backend cupy --manifest /shared/manifests/fixed-32.yaml \
+  --output-dir /shared/results/xpois-mpi --name repeated-mpi \
+  --aggregation-mode mpi --warmup-rounds 1 --measure-rounds 3
+```
+
+File aggregation does not support synchronized rounds.
+Omitting both flags preserves the ordinary single-pass execution and artifact
+layout. Supplying either flag opts in; unspecified warmup and measured counts
+default to zero and one respectively.
+
+The same processes and GPU assignments remain alive across rounds. Every
+round runs the ordinary image-pair workflow, including input identity checks,
+input reads, host preparation, device transfers, numerical processing, and
+output writes. Imports, CUDA contexts, and runtime caches can stay warm; input
+arrays are not held on the device between rounds. Worker readiness includes
+GPU identity initialization, so zero warmup rounds does not mean cold CUDA.
+These are independent-image batch timings, not one image split across GPUs.
+
+Each round retains its normal outputs and audits under
+`rounds/warmup-0000/`, `rounds/measure-0000/`, and subsequent numbered
+directories. Its run ID identifies both the parent invocation and the round.
+The root `summary.json` contains a `benchmark` report with round receipts,
+phase-tagged failures, explicit timing definitions, and measured minimum,
+median, and maximum. An interrupted round can leave partial artifacts without
+a complete receipt.
+A failed warmup or measured round invalidates the aggregate. Dragon also
+requires successful worker exits and cleanup before accepting an aggregate.
+Failed and slow rounds remain in the evidence; no automatic
+outlier removal or relabeling as warmup occurs. A failed round stops further
+rounds after the normal shard processing and audits.
+
+`batch_wall_sec` starts before the coordinator releases a round and ends when
+all completion receipts arrive. It includes the normal durable item/worker
+record writes but excludes subsequent coordinator artifact audits. Readiness
+and function-level elapsed time are reported separately; Dragon also records
+process-group join and cleanup times. MPI process exit and finalization occur
+after the report, so check the launcher's status as well.
+`worker_wall_max_sec` is the longest worker-local round duration. Subtracting
+it from batch wall time gives a mixed remainder that includes scheduling,
+communication, and publication; it is not a transport-only measurement.
+External launch-to-exit wall time still requires timing the launcher.
+
+Dragon places each command queue on its consuming worker's host so idle
+workers do not occupy remote receive slots on the coordinator. Transport
+selection remains a launcher option; `--executor dragon` cannot change the
+transport of an already running Dragon session. The worker wall-time limit
+covers the entire invocation, including warmups and all measured rounds.
+
 ### Results and limits
 
 Both routes print a compact result containing the executor, run ID, run
diff --git a/src/cuphoton/core/benchmark.py b/src/cuphoton/core/benchmark.py
new file mode 100644
index 00000000..7865cdeb
--- /dev/null
+++ b/src/cuphoton/core/benchmark.py
@@ -0,0 +1,171 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Round identities and reports for repeated work in persistent executors.
+
+Executors own synchronization and workload-specific artifact validation. This
+module has no optional runtime dependencies and never discards a slow round.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass
+from statistics import median
+from typing import Any
+
+
+@dataclass(frozen=True)
+class BenchmarkRound:
+    """A warmup or measured pass over the complete input manifest."""
+
+    phase: str
+    index: int
+
+    def __post_init__(self) -> None:
+        if self.phase not in {"warmup", "measure"}:
+            raise ValueError("round phase must be warmup or measure")
+        _validate_count(self.index, "round index", minimum=0)
+
+    @property
+    def round_id(self) -> str:
+        return f"{self.phase}-{self.index:04d}"
+
+    def to_payload(self) -> dict[str, Any]:
+        return {"round_id": self.round_id, **asdict(self)}
+
+    def run_id(self, parent_run_id: str) -> str:
+        """Bind ordinary per-item artifacts to this invocation and round."""
+
+        digest = hashlib.sha256(parent_run_id.encode("utf-8")).hexdigest()[
+            :32
+        ]
+        return f"benchmark-{digest}-{self.round_id}"
+
+
+@dataclass(frozen=True)
+class BenchmarkOptions:
+    """Opt-in repetitions; None keeps an executor's ordinary run."""
+
+    warmup_rounds: int = 0
+    measure_rounds: int = 1
+
+    def __post_init__(self) -> None:
+        _validate_count(self.warmup_rounds, "warmup_rounds", minimum=0)
+        _validate_count(self.measure_rounds, "measure_rounds", minimum=1)
+
+    def rounds(self) -> tuple[BenchmarkRound, ...]:
+        return tuple(
+            BenchmarkRound(phase, index)
+            for phase, count in (
+                ("warmup", self.warmup_rounds),
+                ("measure", self.measure_rounds),
+            )
+            for index in range(count)
+        )
+
+    def to_payload(self) -> dict[str, int]:
+        return asdict(self)
+
+
+def build_benchmark_report(
+    options: BenchmarkOptions,
+    rounds: Sequence[Mapping[str, Any]],
+    *,
+    errors: Sequence[Mapping[str, Any]] = (),
+) -> dict[str, Any]:
+    """Keep raw evidence and summarize only a complete successful invocation.
+
+    An adapter must audit scientific artifacts and worker provenance before
+    marking a round successful. Durations use the coordinator's monotonic
+    clock; worker durations use each worker's local clock, never subtracted
+    timestamps.
+    """
+
+    receipts = [dict(receipt) for receipt in rounds]
+    problems = [dict(error) for error in errors]
+    expected = options.rounds()
+    if len(receipts) != len(expected):
+        problems.append(
+            {"phase": "round_audit", "message": "incomplete round sequence"}
+        )
+    for index, receipt in enumerate(receipts):
+        identity = (
+            expected[index].to_payload() if index < len(expected) else {}
+        )
+        if not identity or any(
+            type(receipt.get(key)) is not type(value)
+            or receipt.get(key) != value
+            for key, value in identity.items()
+        ):
+            problems.append(
+                {"phase": "round_audit", "message": f"invalid round {index}"}
+            )
+        if receipt.get("status") != "success":
+            problems.append(
+                {"phase": "round_audit", "message": f"failed round {index}"}
+            )
+        for field in ("batch_wall_sec", "worker_wall_max_sec"):
+            value = receipt.get(field)
+            if (
+                isinstance(value, bool)
+                or not isinstance(value, (int, float))
+                or not math.isfinite(value)
+                or value < 0
+            ):
+                problems.append(
+                    {
+                        "phase": "round_audit",
+                        "message": f"invalid {field} in round {index}",
+                    }
+                )
+    measured = (
+        [
+            receipt["batch_wall_sec"]
+            for receipt in receipts
+            if receipt.get("phase") == "measure"
+        ]
+        if not problems
+        else []
+    )
+    return {
+        "schema": "cuphoton.core.benchmark/v1",
+        "options": options.to_payload(),
+        "status": "failed" if problems else "success",
+        "rounds": receipts,
+        "errors": problems,
+        "measured_batch_wall_sec": (
+            {
+                "min": min(measured),
+                "median": median(measured),
+                "max": max(measured),
+            }
+            if measured
+            else None
+        ),
+        "timing_definitions": {
+            "batch_wall_sec": (
+                "Coordinator time from before round release through receipt "
+                "of all worker completions, including ordinary input reads, "
+                "numerical work, output writes and record publication; "
+                "excludes subsequent coordinator artifact audits."
+            ),
+            "worker_wall_max_sec": (
+                "Maximum worker-local elapsed time for the round. Its "
+                "difference from batch_wall_sec is a mixed scheduling, "
+                "communication and publication remainder, not pure transport."
+            ),
+        },
+    }
+
+
+def _validate_count(value: int, name: str, *, minimum: int) -> None:
+    if (
+        isinstance(value, bool)
+        or not isinstance(value, int)
+        or value < minimum
+    ):
+        raise ValueError(f"{name} must be an integer >= {minimum}")
diff --git a/src/cuphoton/xpois/commands.py b/src/cuphoton/xpois/commands.py
index 3aff8d23..3f3cee78 100644
--- a/src/cuphoton/xpois/commands.py
+++ b/src/cuphoton/xpois/commands.py
@@ -12,6 +12,7 @@
 
 from numpy.linalg import LinAlgError
 
+from cuphoton.core.benchmark import BenchmarkOptions
 from cuphoton.core.cli import (
     BoolInvariant,
     CommandError,
@@ -613,6 +614,28 @@ class FitBatchCommand(_SpatialSolverOptionsCommand):
     rank_timeout_sec = None
     attempt_id = None
     backend = None
+    warmup_rounds = None
+    measure_rounds = None
+
+    class WarmupRoundsArg(NonNegativeIntegerInvariant):
+        _arg = "--warmup-rounds"
+        _help = (
+            "Opt into persistent-worker benchmarking with this many warmup "
+            "passes over the manifest. Outputs are retained. "
+            "[default when benchmarking: 0]"
+        )
+        _mandatory = False
+        _default = None
+
+    class MeasureRoundsArg(PositiveIntegerInvariant):
+        _arg = "--measure-rounds"
+        _help = (
+            "Opt into persistent-worker benchmarking with this many measured "
+            "passes over the manifest. MPI requires aggregation mpi. "
+            "[default when benchmarking: 1]"
+        )
+        _mandatory = False
+        _default = None
 
     class ExecutorArg(SetInvariant):
         _arg = "--executor"
@@ -728,6 +751,11 @@ def run(self) -> None:
             "run_id": self.run_name or None,
             "options": options,
         }
+        if self.warmup_rounds is not None or self.measure_rounds is not None:
+            common["benchmark"] = BenchmarkOptions(
+                warmup_rounds=self.warmup_rounds or 0,
+                measure_rounds=self.measure_rounds or 1,
+            )
         if self.executor == "dragon":
             result = self._call(
                 _run_dragon_image_pair_batch,
@@ -778,6 +806,18 @@ def run(self) -> None:
             )
 
     def _validate_executor_options(self) -> None:
+        if (
+            self.executor == "mpi"
+            and self.aggregation_mode == "files"
+            and (
+                self.warmup_rounds is not None
+                or self.measure_rounds is not None
+            )
+        ):
+            raise CommandError(
+                "--warmup-rounds and --measure-rounds require MPI "
+                "--aggregation-mode mpi"
+            )
         if self.executor == "dragon":
             invalid = (
                 ("--aggregation-mode", self.aggregation_mode),
diff --git a/src/cuphoton/xpois/dragon.py b/src/cuphoton/xpois/dragon.py
index e0238ae3..608b828f 100644
--- a/src/cuphoton/xpois/dragon.py
+++ b/src/cuphoton/xpois/dragon.py
@@ -23,6 +23,7 @@
 from pathlib import Path
 from typing import Any
 
+from cuphoton.core.benchmark import BenchmarkOptions, build_benchmark_report
 from cuphoton.core.bulk import (
     Placement,
     WorkItem,
@@ -57,6 +58,13 @@
 _DRAGON_GPU_BACKENDS = frozenset({"cupy", "numba-cuda", "cutile"})
 _DRAGON_LAUNCH_SCHEMA = "cuphoton.xpois.dragon-launch/v1"
 _DRAGON_TEMPLATE_BUDGET_BYTES = 96 * 1024
+_WORKER_POLL_SEC = 0.5
+
+
+def _dragon_process_id() -> int:
+    from dragon.infrastructure.parameters import this_process
+
+    return int(this_process.my_puid)
 
 
 @dataclass(frozen=True)
@@ -150,6 +158,7 @@ def _dragon_launch_worker(
     descriptor_sha256: str,
     context: Mapping[str, Any],
     results_queue: Any,
+    commands: Any | None = None,
 ) -> None:
     """Load an immutable shard descriptor before importing its worker."""
 
@@ -197,15 +206,17 @@ def _dragon_launch_worker(
                 "Dragon launch descriptor shard identity differs"
             )
         target = _resolve_worker_target(descriptor["worker_target"])
-        target(
-            run_id,
-            str(run_dir),
-            descriptor["placement"],
-            descriptor["items"],
-            descriptor["options"],
-            results_queue,
-            descriptor["allow_loopback_alias"],
-        )
+        if commands is None:
+            target(
+                run_id,
+                str(run_dir),
+                descriptor["placement"],
+                descriptor["items"],
+                descriptor["options"],
+                results_queue,
+                descriptor["allow_loopback_alias"],
+            )
+            return
     except Exception as exc:
         result = {
             "schema": context["shard_schema"],
@@ -224,6 +235,18 @@ def _dragon_launch_worker(
             "record_write_errors": [],
             "error": error_payload(exc),
         }
+        if commands is not None:
+            puid = None
+            try:
+                puid = _dragon_process_id()
+            except Exception as identity_exc:
+                result["identity_error"] = error_payload(identity_exc)
+            result.update(
+                kind="ready",
+                run_id=run_id,
+                round_id=None,
+                puid=puid,
+            )
         try:
             atomic_write_json(
                 run_dir / "launch" / f"worker-{worker_id:04d}-failure.json",
@@ -237,6 +260,21 @@ def _dragon_launch_worker(
         except Exception as report_exc:
             exc.add_note(f"Dragon launch failure report failed: {report_exc}")
         raise
+    # The persistent worker reports its own READY and round failures. Keep its
+    # invocation outside this handler to avoid a second READY receipt.
+    target(
+        run_id,
+        str(run_dir),
+        descriptor["placement"],
+        descriptor["items"],
+        descriptor["options"],
+        descriptor["benchmark"],
+        commands,
+        results_queue,
+        descriptor["allow_loopback_alias"],
+        descriptor["worker_timeout_sec"],
+        descriptor["result_timeout_sec"],
+    )
 
 
 def run_dragon_image_pair_batch(
@@ -248,12 +286,15 @@ def run_dragon_image_pair_batch(
     result_timeout_sec: float,
     options: BatchFitOptions,
     worker_timeout_sec: float = 3600.0,
+    benchmark: BenchmarkOptions | None = None,
 ) -> DragonBatchResult:
     """Run deterministic XPOIS shards with one Dragon worker per GPU."""
 
     invocation_start = time.perf_counter()
     started_at = timestamp_utc()
     coordinator_timings: dict[str, float] = {}
+    if benchmark is not None and not isinstance(benchmark, BenchmarkOptions):
+        raise TypeError("benchmark must be BenchmarkOptions or None")
     if options.backend not in _DRAGON_GPU_BACKENDS:
         raise ValueError(
             "Dragon XPOIS workers require an explicit GPU backend"
@@ -289,6 +330,7 @@ def run_dragon_image_pair_batch(
         coordinator_timings=coordinator_timings,
         invocation_start=invocation_start,
         started_at=started_at,
+        benchmark=benchmark,
     )
 
 
@@ -311,6 +353,7 @@ def run_dragon_work_items(
     record_schema: str,
     shard_schema: str,
     worker_timeout_sec: float = 3600.0,
+    benchmark: BenchmarkOptions | None = None,
     coordinator_timings: Mapping[str, float] | None = None,
     invocation_start: float | None = None,
     started_at: str | None = None,
@@ -331,6 +374,13 @@ def run_dragon_work_items(
     order. Launch descriptors require a shared filesystem.
     """
 
+    if benchmark is not None:
+        if not isinstance(benchmark, BenchmarkOptions):
+            raise TypeError("benchmark must be BenchmarkOptions or None")
+        if worker_target is not _dragon_shard_worker:
+            raise ValueError(
+                "benchmark rounds currently require the XPOIS worker"
+            )
     if invocation_start is None:
         invocation_start = time.perf_counter()
     started_at = started_at or timestamp_utc()
@@ -415,12 +465,21 @@ def run_dragon_work_items(
     coordinator_timings["partition_sec"] = time.perf_counter() - phase_start
 
     phase_start = time.perf_counter()
-    join_timeout_sec = float(worker_timeout_sec + result_timeout_sec)
+    join_timeout_sec = float(
+        worker_timeout_sec + result_timeout_sec
+        if benchmark is None
+        else result_timeout_sec
+    )
     effective_run_id = run_id or new_run_id(run_prefix)
     validate_identifier(effective_run_id, field="run_id")
     run_dir = output_root.expanduser().resolve() / effective_run_id
     run_dir.mkdir(parents=True, exist_ok=False)
-    for name in ("items", "launch", "records", "workers"):
+    directories = (
+        ("items", "launch", "records", "workers")
+        if benchmark is None
+        else ("launch", "rounds", "startup")
+    )
+    for name in directories:
         (run_dir / name).mkdir()
     atomic_write_json(run_dir / "manifest.json", normalized_manifest)
     atomic_write_json(
@@ -428,30 +487,42 @@ def run_dragon_work_items(
         normalized_input_identity,
     )
     dragon_version = _distribution_version("dragonhpc")
-    atomic_write_json(
-        run_dir / "run.json",
-        {
-            "schema": run_schema,
-            "record_type": "immutable-launch",
-            "executor": "dragon",
-            "run_id": effective_run_id,
-            "started_at_utc": started_at,
-            "manifest_sha256": manifest_sha256,
-            "dragonhpc_version": dragon_version,
-            "worker_count": worker_count,
-            "allocation_node_count": allocation_node_count,
-            "distinct_host_count": distinct_host_count,
-            "worker_timeout_sec": worker_timeout_sec,
-            "join_timeout_sec": join_timeout_sec,
-            "result_timeout_sec": result_timeout_sec,
-            "placements": [placement.to_dict() for placement in selected],
-            "options": normalized_options,
-            "worker_target": worker_reference,
-        },
-    )
+    launch_payload = {
+        "schema": run_schema,
+        "record_type": "immutable-launch",
+        "executor": "dragon",
+        "run_id": effective_run_id,
+        "started_at_utc": started_at,
+        "manifest_sha256": manifest_sha256,
+        "dragonhpc_version": dragon_version,
+        "worker_count": worker_count,
+        "allocation_node_count": allocation_node_count,
+        "distinct_host_count": distinct_host_count,
+        "worker_timeout_sec": worker_timeout_sec,
+        "join_timeout_sec": join_timeout_sec,
+        "result_timeout_sec": result_timeout_sec,
+        "placements": [placement.to_dict() for placement in selected],
+        "options": normalized_options,
+        "worker_target": worker_reference,
+    }
+    if benchmark is not None:
+        launch_payload["benchmark"] = benchmark.to_payload()
+    atomic_write_json(run_dir / "run.json", launch_payload)
     coordinator_timings["run_artifact_setup_sec"] = (
         time.perf_counter() - phase_start
     )
+    if benchmark is not None:
+        return _run_dragon_rounds(
+            api=api,
+            run_dir=run_dir,
+            launch_payload=launch_payload,
+            placements=selected,
+            shards=shards,
+            options=BatchFitOptions.from_payload(normalized_options),
+            benchmark=benchmark,
+            invocation_start=invocation_start,
+            coordinator_timings=coordinator_timings,
+        )
 
     lifecycle_errors: list[dict[str, str]] = []
     exit_status: list[dict[str, int]] = []
@@ -779,6 +850,658 @@ def run_dragon_work_items(
     )
 
 
+def _run_dragon_rounds(
+    *,
+    api: _DragonAPI,
+    run_dir: Path,
+    launch_payload: Mapping[str, Any],
+    placements: Sequence[Placement],
+    shards: Sequence[Sequence[WorkItem]],
+    options: BatchFitOptions,
+    benchmark: BenchmarkOptions,
+    invocation_start: float,
+    coordinator_timings: dict[str, float],
+) -> DragonBatchResult:
+    """Keep placed workers alive while timing complete production shards."""
+
+    run_id = launch_payload["run_id"]
+    worker_count = len(placements)
+    worker_timeout = launch_payload["worker_timeout_sec"]
+    result_timeout = launch_payload["result_timeout_sec"]
+    allow_loopback = launch_payload["allocation_node_count"] == 1
+    lifecycle_errors: list[dict[str, Any]] = []
+    ready_messages: list[dict[str, Any]] = []
+    rounds: list[dict[str, Any]] = []
+    exit_status: list[dict[str, int]] = []
+    command_queues: list[Any] = []
+    results_queue: Any | None = None
+    group: Any | None = None
+    start_attempted = False
+    joined = False
+    phase = "queue_create"
+    process_setup_start = time.perf_counter()
+    try:
+        results_queue = api.Queue(maxsize=worker_count)
+        phase = "group_create"
+        group = api.ProcessGroup(
+            restart=False,
+            ignore_error_on_exit=False,
+            walltime=worker_timeout,
+        )
+        phase = "process_setup"
+        for placement, shard in zip(placements, shards):
+            descriptor_path = (
+                run_dir / "launch" / f"worker-{placement.worker_id:04d}.json"
+            )
+            atomic_write_json(
+                descriptor_path,
+                {
+                    "schema": _DRAGON_LAUNCH_SCHEMA,
+                    "run_id": run_id,
+                    "run_dir": str(run_dir),
+                    "worker_target": _worker_target_reference(
+                        _dragon_round_worker
+                    ),
+                    "placement": placement.to_dict(),
+                    "items": [item.to_dict() for item in shard],
+                    "options": options.to_payload(),
+                    "benchmark": benchmark.to_payload(),
+                    "allow_loopback_alias": allow_loopback,
+                    "worker_timeout_sec": worker_timeout,
+                    "result_timeout_sec": result_timeout,
+                },
+                overwrite=False,
+            )
+            encoded = descriptor_path.read_bytes()
+            context = {
+                "run_dir": str(run_dir),
+                "worker_id": placement.worker_id,
+                "shard_schema": "cuphoton.xpois.dragon-shard/v1",
+                "item_count": len(shard),
+                "weight_bytes": sum(item.weight_bytes for item in shard),
+                "item_ids_sha256": _item_ids_sha256(shard),
+                "descriptor_bytes": len(encoded),
+            }
+            policy = api.Policy(
+                placement=api.Policy.Placement.HOST_NAME,
+                host_name=placement.host,
+                gpu_affinity=[placement.gpu_id],
+            )
+            # Blocking receives reside with their consumer. Python TCP's
+            # coordinator transport threads must remain free to send.
+            commands = api.Queue(maxsize=1, policy=policy)
+            command_queues.append(commands)
+            template = api.ProcessTemplate(
+                target=_dragon_launch_worker,
+                args=(
+                    run_id,
+                    str(descriptor_path),
+                    hashlib.sha256(encoded).hexdigest(),
+                    context,
+                    results_queue,
+                    commands,
+                ),
+                policy=policy,
+            )
+            if len(template.argdata) > _DRAGON_TEMPLATE_BUDGET_BYTES:
+                raise ValueError(
+                    "Dragon worker launch arguments exceed 96 KiB"
+                )
+            group.add_process(nproc=1, template=template)
+        phase = "init"
+        group.init()
+        coordinator_timings["dragon_process_setup_sec"] = (
+            time.perf_counter() - process_setup_start
+        )
+        phase = "start"
+        launch_start = time.perf_counter()
+        worker_deadline = time.monotonic() + worker_timeout
+        start_attempted = True
+        try:
+            group.start()
+        finally:
+            coordinator_timings["dragon_launch_sec"] = (
+                time.perf_counter() - launch_start
+            )
+        phase = "ready"
+        ready_start = time.perf_counter()
+        try:
+            _collect_worker_messages(
+                results_queue,
+                ready_messages,
+                worker_count=worker_count,
+                run_id=run_id,
+                kind="ready",
+                round_id=None,
+                deadline=worker_deadline,
+                group=group,
+            )
+            for message in ready_messages:
+                placement = placements[message["worker_id"]]
+                if not _valid_shard_provenance(
+                    message.get("provenance"),
+                    placement=placement,
+                    allow_loopback_alias=allow_loopback,
+                    backend=options.backend,
+                ):
+                    raise ValueError("invalid Dragon READY provenance")
+        finally:
+            coordinator_timings["worker_ready_wait_sec"] = (
+                time.perf_counter() - ready_start
+            )
+            coordinator_timings["readiness_sec"] = (
+                time.perf_counter() - invocation_start
+            )
+        for spec in benchmark.rounds():
+            phase = f"round:{spec.round_id}"
+            round_dir = run_dir / "rounds" / spec.round_id
+            for name in ("items", "records", "workers"):
+                (round_dir / name).mkdir(parents=True, exist_ok=False)
+            messages: list[dict[str, Any]] = []
+            round_errors: list[dict[str, Any]] = []
+            timings: dict[str, float] = {}
+            batch_start = time.perf_counter()
+            try:
+                try:
+                    for commands in command_queues:
+                        remaining = min(
+                            result_timeout,
+                            worker_deadline - time.monotonic(),
+                        )
+                        if remaining <= 0:
+                            raise TimeoutError(
+                                "Dragon worker deadline expired"
+                            )
+                        commands.put(
+                            {"run_id": run_id, **spec.to_payload()},
+                            timeout=remaining,
+                        )
+                finally:
+                    timings["dispatch_sec"] = (
+                        time.perf_counter() - batch_start
+                    )
+                collection_start = time.perf_counter()
+                try:
+                    _collect_worker_messages(
+                        results_queue,
+                        messages,
+                        worker_count=worker_count,
+                        run_id=run_id,
+                        kind="round",
+                        round_id=spec.round_id,
+                        deadline=worker_deadline,
+                        group=group,
+                    )
+                finally:
+                    timings["collection_sec"] = (
+                        time.perf_counter() - collection_start
+                    )
+            except Exception as exc:
+                round_errors.append(error_payload(exc))
+            batch_wall = time.perf_counter() - batch_start
+            shard_results = [
+                message["result"]
+                for message in messages
+                if message.get("kind") == "round"
+                and message.get("run_id") == run_id
+                and message.get("round_id") == spec.round_id
+                and isinstance(message.get("result"), Mapping)
+            ]
+            ready_by_worker = {
+                message["worker_id"]: message["provenance"]
+                for message in ready_messages
+            }
+            if any(
+                result.get("provenance")
+                != ready_by_worker.get(result.get("worker_id"))
+                for result in shard_results
+            ):
+                round_errors.append(
+                    {
+                        "type": "WorkerIdentityChanged",
+                        "message": "worker provenance changed after READY",
+                    }
+                )
+            audit_start = time.perf_counter()
+            records, record_errors = _wait_terminal_artifacts(
+                round_dir, shards, shard_results, result_timeout
+            )
+            audit = _audit_batch_records(
+                run_id=spec.run_id(run_id),
+                shards=shards,
+                placements=placements,
+                options=options,
+                records=records,
+                record_errors=record_errors,
+                shard_results=shard_results,
+                allow_loopback_alias=allow_loopback,
+            )
+            timings["artifact_audit_sec"] = time.perf_counter() - audit_start
+            worker_wall = max(
+                (
+                    float(message["worker_wall_sec"])
+                    for message in messages
+                    if isinstance(
+                        message.get("worker_wall_sec"), (int, float)
+                    )
+                    and not isinstance(message["worker_wall_sec"], bool)
+                    and math.isfinite(message["worker_wall_sec"])
+                    and message["worker_wall_sec"] >= 0
+                ),
+                default=0.0,
+            )
+            receipt = {
+                **spec.to_payload(),
+                "status": (
+                    "success"
+                    if audit["status"] == "success" and not round_errors
+                    else "failed"
+                ),
+                "batch_wall_sec": batch_wall,
+                "worker_wall_max_sec": worker_wall,
+                "coordinator_timings_sec": timings,
+                "summary_path": f"rounds/{spec.round_id}/summary.json",
+            }
+            atomic_write_json(
+                round_dir / "summary.json",
+                {
+                    "schema": "cuphoton.xpois.dragon-round/v1",
+                    "run_id": spec.run_id(run_id),
+                    "parent_run_id": run_id,
+                    **audit,
+                    **receipt,
+                    "messages": messages,
+                    "errors": round_errors,
+                },
+            )
+            rounds.append(receipt)
+            if receipt["status"] != "success":
+                raise RuntimeError(f"Dragon round {spec.round_id} failed")
+        phase = "join"
+        join_start = time.perf_counter()
+        try:
+            group.join(timeout=result_timeout)
+            joined = True
+        finally:
+            coordinator_timings["worker_join_sec"] = (
+                time.perf_counter() - join_start
+            )
+        phase = "unexpected_result"
+        try:
+            extra = results_queue.get(timeout=0)
+        except (queue.Empty, TimeoutError):
+            pass
+        else:
+            raise ValueError(f"unexpected Dragon worker result: {extra!r}")
+    except Exception as exc:
+        lifecycle_errors.append({"phase": phase, **error_payload(exc)})
+    finally:
+        coordinator_timings.setdefault(
+            "dragon_process_setup_sec",
+            time.perf_counter() - process_setup_start,
+        )
+        cleanup_start = time.perf_counter()
+        if group is not None:
+            if start_attempted and not joined:
+                try:
+                    group.stop(patience=5.0)
+                except Exception as exc:
+                    lifecycle_errors.append(
+                        {"phase": "stop_after_failure", **error_payload(exc)}
+                    )
+            try:
+                exit_status = [
+                    {"puid": int(puid), "exit_code": int(code)}
+                    for puid, code in group.inactive_puids
+                ]
+            except Exception as exc:
+                lifecycle_errors.append(
+                    {"phase": "exit_status", **error_payload(exc)}
+                )
+            try:
+                group.close(patience=5.0)
+            except Exception as exc:
+                lifecycle_errors.append(
+                    {"phase": "close", **error_payload(exc)}
+                )
+                cleanup = getattr(group, "_close_no_decorator", None)
+                if cleanup is not None:
+                    try:
+                        cleanup(patience=5.0)
+                    except Exception as cleanup_exc:
+                        lifecycle_errors.append(
+                            {
+                                "phase": "forced_close",
+                                **error_payload(cleanup_exc),
+                            }
+                        )
+        for channel in [*command_queues, results_queue]:
+            if channel is not None:
+                try:
+                    channel.close()
+                except Exception as exc:
+                    lifecycle_errors.append(
+                        {"phase": "queue_close", **error_payload(exc)}
+                    )
+        coordinator_timings["dragon_cleanup_sec"] = (
+            time.perf_counter() - cleanup_start
+        )
+    process_audit = {
+        "ok": len(exit_status) == worker_count
+        and all(item["exit_code"] == 0 for item in exit_status),
+        "expected_count": worker_count,
+        "observed_count": len(exit_status),
+        "nonzero": [item for item in exit_status if item["exit_code"] != 0],
+    }
+    if not process_audit["ok"]:
+        lifecycle_errors.append(
+            {
+                "phase": "exit_status",
+                "type": "WorkerExitFailure",
+                "message": "missing or nonzero Dragon worker exit status",
+            }
+        )
+    report = build_benchmark_report(
+        benchmark, rounds, errors=lifecycle_errors
+    )
+    summary = {
+        **launch_payload,
+        "schema": "cuphoton.xpois.dragon-benchmark-summary/v1",
+        "record_type": "terminal",
+        "status": report["status"],
+        "completed_at_utc": timestamp_utc(),
+        "coordinator_wall_sec": time.perf_counter() - invocation_start,
+        "coordinator_wall_definition": (
+            "Function entry through worker cleanup and terminal audits; "
+            "the final summary.json atomic commit is excluded."
+        ),
+        "coordinator_timings_sec": coordinator_timings,
+        "ready_messages": ready_messages,
+        "command_queue_placement": "consumer",
+        "process_exit_status": exit_status,
+        "process_exit_audit": process_audit,
+        "lifecycle_errors": lifecycle_errors,
+        "benchmark": report,
+    }
+    summary_path = run_dir / "summary.json"
+    atomic_write_json(summary_path, summary)
+    return DragonBatchResult(
+        run_id=run_id,
+        run_dir=run_dir,
+        summary_path=summary_path,
+        status=report["status"],
+        summary=summary,
+    )
+
+
+def _collect_worker_messages(
+    results_queue: Any,
+    messages: list[dict[str, Any]],
+    *,
+    worker_count: int,
+    run_id: str,
+    kind: str,
+    round_id: str | None,
+    deadline: float,
+    group: Any,
+) -> None:
+    """Collect one identified receipt per worker, failing on any replay."""
+
+    seen: set[int] = set()
+    seen_puids: set[int] = set()
+    failed: list[int] = []
+    while len(seen) < worker_count:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError(f"Dragon {kind} deadline expired")
+        try:
+            raw = results_queue.get(timeout=min(remaining, _WORKER_POLL_SEC))
+        except queue.Empty:
+            exits = list(group.inactive_puids)
+            # A worker publishes before exiting. Recheck the queue after the
+            # exit snapshot so a concurrently delivered receipt is retained.
+            try:
+                raw = results_queue.get(timeout=0)
+            except queue.Empty:
+                missing = [
+                    entry for entry in exits if entry[0] not in seen_puids
+                ]
+                if missing:
+                    raise RuntimeError(
+                        f"Dragon worker exited before {kind} completion: "
+                        f"{missing}"
+                    ) from None
+                continue
+        message = json_mapping(raw, field=f"Dragon {kind} result")
+        messages.append(message)
+        worker_id = _strict_integer(message.get("worker_id"))
+        puid = _strict_integer(message.get("puid"))
+        if (
+            message.get("run_id") != run_id
+            or message.get("kind") != kind
+            or message.get("round_id") != round_id
+            or worker_id is None
+            or worker_id not in range(worker_count)
+            or worker_id in seen
+            or puid is None
+            or puid in seen_puids
+        ):
+            raise ValueError(f"invalid or duplicate Dragon {kind} identity")
+        seen.add(worker_id)
+        seen_puids.add(puid)
+        if message.get("status") != "success":
+            failed.append(worker_id)
+            continue
+        if kind == "round":
+            result = message.get("result")
+            duration = message.get("worker_wall_sec")
+            if (
+                isinstance(duration, bool)
+                or not isinstance(duration, (int, float))
+                or not math.isfinite(duration)
+                or duration < 0
+                or not isinstance(result, Mapping)
+                or _strict_integer(result.get("worker_id")) != worker_id
+                or result.get("status") != "success"
+            ):
+                raise ValueError("invalid Dragon round shard result")
+    if failed:
+        raise RuntimeError(f"Dragon workers {failed} {kind} failed")
+
+
+def _audit_batch_records(
+    *,
+    run_id: str,
+    shards: Sequence[Sequence[WorkItem]],
+    placements: Sequence[Placement],
+    options: BatchFitOptions,
+    records: Sequence[Mapping[str, Any]],
+    record_errors: Sequence[Mapping[str, Any]],
+    shard_results: Sequence[Mapping[str, Any]],
+    allow_loopback_alias: bool,
+) -> dict[str, Any]:
+    """Apply the same scientific artifact contract to every execution."""
+
+    errors = list(record_errors)
+    audit = audit_terminal_records(
+        [item.item_id for shard in shards for item in shard], records
+    )
+    expected = {
+        item.item_id: {
+            "worker_id": placement.worker_id,
+            "weight_bytes": item.weight_bytes,
+            "backend": options.backend,
+            "solver": options.solver,
+        }
+        for placement, shard in zip(placements, shards)
+        for item in shard
+    }
+    errors.extend(
+        _audit_terminal_record_contract(
+            run_id=run_id, expected=expected, records=records
+        )
+    )
+    shard_audit = _audit_shard_results(
+        shards,
+        shard_results,
+        records,
+        placements=placements,
+        allow_loopback_alias=allow_loopback_alias,
+        backend=options.backend,
+    )
+    item_timings, timing_errors = _aggregate_item_timings(records)
+    errors.extend(timing_errors)
+    return {
+        "status": (
+            "success"
+            if audit["ok"]
+            and not audit["failed_item_ids"]
+            and shard_audit["ok"]
+            and not errors
+            else "failed"
+        ),
+        "shard_result_audit": shard_audit,
+        "terminal_record_audit": audit,
+        "terminal_record_errors": errors,
+        "timings_sec": item_timings,
+    }
+
+
+def _dragon_round_worker(
+    run_id: str,
+    run_dir_raw: str,
+    placement_payload: Mapping[str, Any],
+    item_payloads: Sequence[Mapping[str, Any]],
+    options_payload: Mapping[str, Any],
+    benchmark_payload: Mapping[str, Any],
+    commands: Any,
+    results_queue: Any,
+    allow_loopback_alias: bool,
+    worker_timeout: float,
+    result_timeout: float,
+) -> None:
+    """Initialize once, then execute the normal shard for each release."""
+
+    run_dir = Path(run_dir_raw)
+    placement = Placement(**dict(placement_payload))
+    ready: dict[str, Any] = {
+        "kind": "ready",
+        "run_id": run_id,
+        "round_id": None,
+        "worker_id": placement.worker_id,
+        "puid": _dragon_process_id(),
+        "status": "success",
+    }
+    try:
+        items = tuple(
+            WorkItem.from_dict(payload) for payload in item_payloads
+        )
+        options = BatchFitOptions.from_payload(options_payload)
+        benchmark = BenchmarkOptions(**dict(benchmark_payload))
+        actual_host, visibility = _validate_worker_placement(
+            placement,
+            require_clean_cuda_imports=True,
+            allow_loopback_alias=allow_loopback_alias,
+        )
+        gpu_identity = dict(_collect_gpu_identity(options.backend))
+        ready["provenance"] = {
+            "worker_id": placement.worker_id,
+            "requested_host": placement.host,
+            "requested_gpu_id": placement.gpu_id,
+            "hostname": actual_host,
+            "pid": os.getpid(),
+            "cuda_visible_devices": visibility,
+            "gpu": gpu_identity,
+        }
+    except Exception as exc:
+        ready.update(status="failed", error=error_payload(exc))
+    try:
+        atomic_write_json(
+            run_dir / "startup" / f"worker-{placement.worker_id:04d}.json",
+            ready,
+        )
+    except Exception as exc:
+        ready.update(status="failed", artifact_error=error_payload(exc))
+    results_queue.put(ready, timeout=result_timeout)
+    if ready["status"] != "success":
+        raise RuntimeError("Dragon worker initialization failed")
+    for spec in benchmark.rounds():
+        round_dir = run_dir / "rounds" / spec.round_id
+        message: dict[str, Any] = {
+            "kind": "round",
+            "run_id": run_id,
+            "round_id": spec.round_id,
+            "worker_id": placement.worker_id,
+            "puid": ready["puid"],
+            "status": "success",
+        }
+        try:
+            command = commands.get(timeout=worker_timeout)
+            if command != {"run_id": run_id, **spec.to_payload()}:
+                raise ValueError("unexpected Dragon round command")
+            worker_start = time.perf_counter()
+            result = _execute_shard(
+                run_id=spec.run_id(run_id),
+                run_dir=round_dir,
+                placement=placement,
+                items=items,
+                options=options,
+                item_runner=run_image_pair_item,
+                gpu_identity_loader=lambda backend: gpu_identity,
+                require_clean_cuda_imports=False,
+                allow_loopback_alias=allow_loopback_alias,
+            )
+            message.update(
+                status=result["status"],
+                result=result,
+                worker_wall_sec=time.perf_counter() - worker_start,
+            )
+        except Exception as exc:
+            message.update(status="failed", error=error_payload(exc))
+            try:
+                atomic_write_json(
+                    round_dir
+                    / "errors"
+                    / f"worker-{placement.worker_id:04d}.json",
+                    message,
+                )
+            except Exception as artifact_exc:
+                message["artifact_error"] = error_payload(artifact_exc)
+        results_queue.put(message, timeout=result_timeout)
+        if message["status"] != "success":
+            raise RuntimeError(f"Dragon worker round {spec.round_id} failed")
+
+
+def _validate_worker_placement(
+    placement: Placement,
+    *,
+    require_clean_cuda_imports: bool,
+    allow_loopback_alias: bool,
+) -> tuple[str, str]:
+    actual_host = socket.gethostname()
+    if not _hostnames_match(
+        placement.host,
+        actual_host,
+        allow_loopback_alias=allow_loopback_alias,
+    ):
+        raise RuntimeError(
+            "Dragon worker placement mismatch: requested "
+            f"{placement.host!r}, running on {actual_host!r}"
+        )
+    visibility = _singleton_cuda_visibility(placement.gpu_id)
+    premature = sorted(
+        name
+        for name in ("cupy", "numba.cuda", "cuda.tile", "torch")
+        if name in sys.modules
+    )
+    if require_clean_cuda_imports and premature:
+        raise RuntimeError(
+            "CUDA modules were imported before Dragon worker placement: "
+            + ", ".join(premature)
+        )
+    return actual_host, visibility
+
+
 def discover_gpu_placements(
     system_type: Callable[[], Any],
     node_type: Callable[[Any], Any],
@@ -956,26 +1679,11 @@ def _execute_shard(
         raise ValueError("Dragon shard backend must be a non-empty string")
     try:
         actual_host = socket.gethostname()
-        if not _hostnames_match(
-            placement.host,
-            actual_host,
+        actual_host, visibility = _validate_worker_placement(
+            placement,
+            require_clean_cuda_imports=require_clean_cuda_imports,
             allow_loopback_alias=allow_loopback_alias,
-        ):
-            raise RuntimeError(
-                "Dragon worker placement mismatch: requested "
-                f"{placement.host!r}, running on {actual_host!r}"
-            )
-        visibility = _singleton_cuda_visibility(placement.gpu_id)
-        premature = sorted(
-            name
-            for name in ("cupy", "numba.cuda", "cuda.tile", "torch")
-            if name in sys.modules
         )
-        if require_clean_cuda_imports and premature:
-            raise RuntimeError(
-                "CUDA modules were imported before Dragon worker placement: "
-                + ", ".join(premature)
-            )
         gpu_identity = dict(gpu_identity_loader(resolved_backend))
     except Exception as exc:
         setup_exception = exc
@@ -1176,7 +1884,10 @@ def _load_terminal_records(
         try:
             if not _regular_file(path):
                 raise ValueError(f"record is not a regular file: {path.name}")
-            records.append(read_json_mapping(path))
+            record = read_json_mapping(path)
+            if record.get("item_id") != path.stem:
+                raise ValueError("record item_id differs from its filename")
+            records.append(record)
         except Exception as exc:
             error = {
                 "record_path": str(path.relative_to(run_dir)),
diff --git a/src/cuphoton/xpois/mpi.py b/src/cuphoton/xpois/mpi.py
index bd0817cb..ae490f86 100644
--- a/src/cuphoton/xpois/mpi.py
+++ b/src/cuphoton/xpois/mpi.py
@@ -24,6 +24,7 @@
 from pathlib import Path
 from typing import Any
 
+from cuphoton.core.benchmark import BenchmarkOptions, build_benchmark_report
 from cuphoton.core.bulk import (
     WorkItem,
     atomic_write_json,
@@ -125,16 +126,23 @@ def run_mpi_image_pair_batch(
     attempt_id: str | None,
     options: BatchFitOptions,
     rank_setup_timeout_sec: float = 600.0,
+    benchmark: BenchmarkOptions | None = None,
 ) -> MPIBatchResult | None:
     """Run byte-balanced XPOIS shards under an external MPI launcher."""
 
     started_at = timestamp_utc()
     start = time.perf_counter()
+    if benchmark is not None and aggregation_mode != "mpi":
+        raise ValueError("benchmark rounds require aggregation_mode='mpi'")
     api = None
     if aggregation_mode == "mpi":
         visibility = None
         startup_error = None
         try:
+            if benchmark is not None and not isinstance(
+                benchmark, BenchmarkOptions
+            ):
+                raise TypeError("benchmark must be BenchmarkOptions or None")
             visibility = _validate_rank_startup(
                 aggregation_mode,
                 rank_setup_timeout_sec,
@@ -227,6 +235,7 @@ def run_mpi_image_pair_batch(
             api.mpi4py_version,
             api.library_version,
             rank_setup_timeout_sec,
+            benchmark,
         )
         if manifest is None or shards is None or run_dir is None:
             raise RuntimeError(
@@ -282,6 +291,23 @@ def run_mpi_image_pair_batch(
         assert manifest is not None
         assert shards is not None
 
+    if benchmark is not None:
+        assert api is not None
+        return _run_mpi_benchmark(
+            api,
+            context,
+            run_dir,
+            effective_run_id,
+            manifest,
+            options,
+            shards,
+            visibility,
+            benchmark,
+            started_at,
+            start,
+            rank_setup_timeout_sec,
+        )
+
     rank_run_dir = run_dir
     if api is None:
         assert attempt_id is not None
@@ -389,6 +415,181 @@ def _validate_rank_startup(
     return visibility
 
 
+def _run_mpi_benchmark(
+    api: _MPIAPI,
+    context: _RankContext,
+    run_dir: Path,
+    run_id: str,
+    manifest: ImagePairManifest,
+    options: BatchFitOptions,
+    shards: Sequence[Sequence[WorkItem]],
+    visibility: str,
+    benchmark: BenchmarkOptions,
+    started_at: str,
+    start: float,
+    rank_setup_timeout_sec: float,
+) -> MPIBatchResult | None:
+    """Repeat the ordinary shard execution within the original MPI ranks."""
+
+    rounds: list[dict[str, Any]] = []
+    errors: list[dict[str, Any]] = []
+    startup_ready_sec = None
+    phase = "setup"
+    try:
+        setup_error = None
+        gpu = None
+        try:
+            gpu = json_mapping(
+                _gpu_identity(options.backend), field="MPI GPU identity"
+            )
+            if context.rank == 0:
+                atomic_write_json(
+                    run_dir / "benchmark-plan.json", benchmark.to_payload()
+                )
+        except Exception as exc:
+            setup_error = error_payload(exc)
+        _mpi_failure_consensus(
+            api.comm, "MPI benchmark readiness", setup_error
+        )
+        assert gpu is not None
+        startup_ready_sec = time.perf_counter() - start
+
+        for planned in benchmark.rounds():
+            phase = planned.round_id
+            round_dir = run_dir / "rounds" / planned.round_id
+            round_run_id = planned.run_id(run_id)
+            round_started_at = timestamp_utc()
+            _mpi_prepare_run(
+                api.comm,
+                context,
+                round_dir,
+                round_run_id,
+                round_started_at,
+                manifest,
+                options,
+                api,
+                rank_setup_timeout_sec,
+                shards=shards,
+                start=time.perf_counter(),
+            )
+            # Run preparation includes a readiness gather. No rank starts
+            # work until root releases this round through the broadcast.
+            round_start = time.perf_counter()
+            api.comm.bcast(planned.round_id, root=0)
+            timing = {"release_sec": time.perf_counter() - round_start}
+            worker_start = time.perf_counter()
+            rank_result = _execute_rank(
+                run_id=round_run_id,
+                run_dir=round_dir,
+                manifest_sha256=manifest.sha256,
+                context=context,
+                items=shards[context.rank],
+                options=options,
+                visibility=visibility,
+                setup_error=None,
+                item_runner=run_image_pair_item,
+                gpu_identity_loader=lambda backend: gpu,
+            )
+            timing["worker_wall_sec"] = time.perf_counter() - worker_start
+            result = _mpi_aggregate(
+                api.comm,
+                context,
+                round_dir,
+                round_run_id,
+                manifest,
+                options,
+                shards,
+                rank_result,
+                round_started_at,
+                round_start,
+                api,
+                rank_setup_timeout_sec=rank_setup_timeout_sec,
+                benchmark_timing=timing,
+            )
+            round_decision = None
+            if context.rank == 0:
+                try:
+                    assert result is not None
+                    rounds.append(
+                        {
+                            **planned.to_payload(),
+                            "status": result.status,
+                            "batch_wall_sec": timing.pop("batch_wall_sec"),
+                            "worker_wall_max_sec": timing.pop(
+                                "worker_wall_max_sec"
+                            ),
+                            "summary_path": str(
+                                result.summary_path.relative_to(run_dir)
+                            ),
+                            "coordinator_timings_sec": timing,
+                        }
+                    )
+                    round_decision = {"status": result.status, "error": None}
+                except Exception as exc:
+                    round_decision = {
+                        "status": "failed",
+                        "error": error_payload(exc),
+                    }
+            round_decision = api.comm.bcast(round_decision, root=0)
+            if not isinstance(round_decision, Mapping):
+                raise RuntimeError("invalid MPI benchmark round decision")
+            if round_decision.get("error"):
+                raise RuntimeError(str(round_decision["error"]))
+            if round_decision.get("status") != "success":
+                break
+    except Exception as exc:
+        errors.append({"phase": phase, **error_payload(exc)})
+
+    result = None
+    decision = None
+    if context.rank == 0:
+        try:
+            report = build_benchmark_report(benchmark, rounds, errors=errors)
+            summary = {
+                "schema": "cuphoton.xpois.mpi-benchmark-summary/v1",
+                "executor": "mpi",
+                "status": report["status"],
+                "run_id": run_id,
+                "manifest_sha256": manifest.sha256,
+                "aggregation_mode": "mpi",
+                "world_size": context.world_size,
+                "mpi4py_version": api.mpi4py_version,
+                "mpi_library_version": api.library_version,
+                "options": options.to_payload(),
+                "rank_setup_timeout_sec": rank_setup_timeout_sec,
+                "started_at_utc": started_at,
+                "completed_at_utc": timestamp_utc(),
+                "coordinator_wall_sec": time.perf_counter() - start,
+                "startup_ready_sec": startup_ready_sec,
+                "benchmark": report,
+                "launcher_exit_note": (
+                    "Collective completion is not launcher process-exit "
+                    "proof."
+                ),
+            }
+            summary_path = run_dir / "summary.json"
+            atomic_write_json(summary_path, summary)
+            result = MPIBatchResult(
+                run_id, run_dir, summary_path, report["status"], summary
+            )
+            decision = {"status": result.status, "error": None}
+        except Exception as exc:
+            decision = {"status": "failed", "error": error_payload(exc)}
+            _aggregate_error(run_dir, run_id, decision["error"])
+    decision = api.comm.bcast(decision, root=0)
+    if not isinstance(decision, Mapping):
+        raise RuntimeError(
+            "MPI benchmark aggregate decision was not a mapping"
+        )
+    if decision.get("error"):
+        raise RuntimeError(
+            f"cannot persist MPI benchmark aggregate: {decision['error']}"
+        )
+    if context.rank != 0 and decision.get("status") != "success":
+        raise RuntimeError(f"MPI benchmark {run_id!r} failed")
+    return result
+
+
 def _mpi_failure_consensus(
     comm: Any,
     phase: str,
@@ -524,6 +725,7 @@ def _mpi_manifest_consensus(
     mpi4py_version: str | None,
     library_version: str | None,
     rank_setup_timeout_sec: float = 600.0,
+    benchmark: BenchmarkOptions | None = None,
 ) -> None:
     local = {
         "rank": context.rank,
@@ -533,6 +735,7 @@ def _mpi_manifest_consensus(
             str(resolved_run_dir) if resolved_run_dir is not None else None
         ),
         "options": options.to_payload(),
+        "benchmark": benchmark.to_payload() if benchmark else None,
         "mpi4py_version": mpi4py_version,
         "mpi_library_version": _normalize_mpi_library_version(
             library_version
@@ -621,6 +824,11 @@ def _mpi_manifest_consensus(
                 item["options"] != records[0]["options"] for item in records
             ):
                 message = "XPOIS options differ across MPI ranks"
+            elif any(
+                item.get("benchmark") != records[0].get("benchmark")
+                for item in records
+            ):
+                message = "benchmark options differ across MPI ranks"
             elif any(
                 item.get("mpi4py_version") != records[0].get("mpi4py_version")
                 for item in records
@@ -1489,12 +1697,32 @@ def _mpi_aggregate(
     api: _MPIAPI,
     *,
     rank_setup_timeout_sec: float = 600.0,
+    benchmark_timing: dict[str, float] | None = None,
 ) -> MPIBatchResult | None:
-    gathered = comm.gather(dict(rank_result), root=0)
+    payload = dict(rank_result)
+    if benchmark_timing is not None:
+        payload = {
+            "rank_result": payload,
+            "worker_wall_sec": benchmark_timing.pop("worker_wall_sec"),
+        }
+    collection_start = time.perf_counter()
+    gathered = comm.gather(payload, root=0)
+    if benchmark_timing is not None:
+        collected_at = time.perf_counter()
+        benchmark_timing["collection_sec"] = collected_at - collection_start
+        benchmark_timing["batch_wall_sec"] = collected_at - start
     result = None
     decision = None
     if context.rank == 0:
         try:
+            if benchmark_timing is not None:
+                worker_times = [item["worker_wall_sec"] for item in gathered]
+                if not all(
+                    _finite_nonnegative(value) for value in worker_times
+                ):
+                    raise ValueError("invalid MPI benchmark worker duration")
+                benchmark_timing["worker_wall_max_sec"] = max(worker_times)
+                gathered = [item["rank_result"] for item in gathered]
             aggregation_errors = _wait_collective_artifacts(
                 run_dir,
                 shards,
@@ -1524,9 +1752,11 @@ def _mpi_aggregate(
     if not isinstance(decision, Mapping):
         raise RuntimeError("MPI aggregate decision was not a mapping")
     if decision.get("error"):
-        raise RuntimeError("cannot persist MPI aggregate")
+        raise RuntimeError(
+            f"cannot persist MPI aggregate: {decision['error']}"
+        )
     if context.rank != 0:
-        if decision.get("status") != "success":
+        if decision.get("status") != "success" and benchmark_timing is None:
             raise RuntimeError(f"MPI batch {run_id!r} failed")
         return None
     assert result is not None
@@ -1622,7 +1852,9 @@ def _finalize(
     rank_timeout_sec: float | None = None,
     rank_setup_timeout_sec: float | None = None,
 ) -> MPIBatchResult:
-    records, record_errors = _read_mappings(run_dir / "records")
+    records, record_errors = _read_mappings(
+        run_dir / "records", validate_item_filename=True
+    )
     expected_items = [item.item_id for shard in shards for item in shard]
     item_audit = audit_terminal_records(expected_items, records)
     record_errors.extend(
@@ -2191,12 +2423,17 @@ def _same_json_value(left: Any, right: Any) -> bool:
 
 def _read_mappings(
     directory: Path,
+    *,
+    validate_item_filename: bool = False,
 ) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
     records: list[dict[str, Any]] = []
     errors: list[dict[str, str]] = []
     for path in sorted(directory.glob("*.json")):
         try:
-            records.append(_read_regular_mapping(path))
+            record = _read_regular_mapping(path)
+            if validate_item_filename and path.stem != record.get("item_id"):
+                raise ValueError("record filename does not match item_id")
+            records.append(record)
         except Exception as exc:
             errors.append({"path": path.name, **error_payload(exc)})
     return records, errors
diff --git a/tests/core/test_benchmark.py b/tests/core/test_benchmark.py
new file mode 100644
index 00000000..82fbb7f8
--- /dev/null
+++ b/tests/core/test_benchmark.py
@@ -0,0 +1,126 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from cuphoton.core.benchmark import (
+    BenchmarkOptions,
+    BenchmarkRound,
+    build_benchmark_report,
+)
+from cuphoton.core.bulk import validate_identifier
+
+
+def _receipts(options):
+    return [
+        {
+            **item.to_payload(),
+            "status": "success",
+            "batch_wall_sec": duration,
+            "worker_wall_max_sec": duration / 2,
+            "summary_path": f"rounds/{item.round_id}/summary.json",
+        }
+        for item, duration in zip(options.rounds(), (100.0, 20.0, 2.0, 4.0))
+    ]
+
+
+def test_report_retains_first_round_and_excludes_only_explicit_warmup():
+    options = BenchmarkOptions(warmup_rounds=1, measure_rounds=3)
+    receipts = _receipts(options)
+    report = build_benchmark_report(options, receipts)
+    assert report["status"] == "success"
+    assert report["rounds"] == receipts
+    assert report["measured_batch_wall_sec"] == {
+        "min": 2.0,
+        "median": 4.0,
+        "max": 20.0,
+    }
+
+
+@pytest.mark.parametrize(
+    "defect",
+    [
+        "missing",
+        "duplicate",
+        "wrong_phase",
+        "bool_index",
+        "failed",
+        "bad_time",
+        "extra",
+    ],
+)
+def test_invalid_evidence_has_no_accepted_aggregate(defect):
+    options = BenchmarkOptions(warmup_rounds=1, measure_rounds=3)
+    receipts = _receipts(options)
+    if defect == "missing":
+        receipts.pop()
+    elif defect == "duplicate":
+        receipts[2] = dict(receipts[1])
+    elif defect == "wrong_phase":
+        receipts[1]["phase"] = "warmup"
+    elif defect == "bool_index":
+        receipts[1]["index"] = False
+    elif defect == "failed":
+        receipts[0]["status"] = "failed"
+    elif defect == "bad_time":
+        receipts[1]["batch_wall_sec"] = -1
+    else:
+        receipts.append(dict(receipts[-1]))
+    original = copy.deepcopy(receipts)
+    report = build_benchmark_report(options, receipts)
+    assert report["status"] == "failed"
+    assert report["measured_batch_wall_sec"] is None
+    assert report["rounds"] == original
+    assert report["errors"]
+
+
+def test_cleanup_failure_invalidates_complete_measurements():
+    options = BenchmarkOptions(measure_rounds=2)
+    report = build_benchmark_report(
+        options,
+        _receipts(options),
+        errors=[{"phase": "close", "message": "worker still running"}],
+    )
+    assert report["status"] == "failed"
+    assert report["measured_batch_wall_sec"] is None
+    assert report["errors"][0]["phase"] == "close"
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("warmup_rounds", -1),
+        ("warmup_rounds", False),
+        ("measure_rounds", 0),
+        ("measure_rounds", 1.5),
+        ("measure_rounds", True),
+    ],
+)
+def test_options_reject_invalid_counts(field, value):
+    with pytest.raises(ValueError):
+        BenchmarkOptions(**{field: value})
+
+
+@pytest.mark.parametrize(
+    "value", [float("nan"), float("inf"), True, None, "1"]
+)
+def test_report_rejects_invalid_durations(value):
+    options = BenchmarkOptions()
+    receipts = _receipts(options)
+    receipts[0]["worker_wall_max_sec"] = value
+    assert build_benchmark_report(options, receipts)["status"] == "failed"
+
+
+def test_round_run_ids_bind_artifacts_to_invocation_and_round():
+    first, second = BenchmarkOptions(measure_rounds=2).rounds()
+    assert first.run_id("parent") != second.run_id("parent")
+    assert first.run_id("parent") != first.run_id("other")
+    assert first.run_id("parent") == first.run_id("parent")
+    validate_identifier(first.run_id("a" * 128), field="run_id")
+    with pytest.raises(ValueError):
+        BenchmarkRound("unknown", 0)
diff --git a/tests/core/test_cli_contract.py b/tests/core/test_cli_contract.py
index 8a575e7c..4be2b23d 100644
--- a/tests/core/test_cli_contract.py
+++ b/tests/core/test_cli_contract.py
@@ -72,7 +72,7 @@ def test_public_command_surface_counts_are_exact() -> None:
     assert per_group == [
         ("xdr", 1, 0, 14, 0),
         ("xfit", 3, 3, 17, 1),
-        ("xpois", 7, 7, 144, 1),
+        ("xpois", 7, 7, 146, 1),
         ("xscan", 42, 42, 171, 1),
         ("xrep", 6, 6, 103, 1),
         ("xray", 33, 31, 390, 1),
@@ -81,7 +81,7 @@ def test_public_command_surface_counts_are_exact() -> None:
     assert sum(item[1] for item in per_group) == 92
     assert sum(item[1] + item[4] for item in per_group) == 97
     assert sum(item[2] for item in per_group) == 89
-    assert sum(item[3] for item in per_group) == 839
+    assert sum(item[3] for item in per_group) == 841
 
 
 def test_public_registry_order_and_component_derivations() -> None:
diff --git a/tests/xpois/test_cli.py b/tests/xpois/test_cli.py
index 6434e550..06e6ec36 100644
--- a/tests/xpois/test_cli.py
+++ b/tests/xpois/test_cli.py
@@ -87,6 +87,8 @@ def test_help_for_fit_batch_command(capsys) -> None:
     assert "--max-workers" in captured.out
     assert "--result-timeout-sec" in captured.out
     assert "--worker-timeout-sec" in captured.out
+    assert "--warmup-rounds" in captured.out
+    assert "--measure-rounds" in captured.out
     assert "--aggregation-mode" in captured.out
     assert "--rank-setup-timeout-sec" in captured.out
     assert "--rank-timeout-sec" in captured.out
@@ -479,6 +481,88 @@ def run_dragon(**kwargs):
     assert seen["max_workers"] is None
     assert seen["result_timeout_sec"] == 60.0
     assert seen["worker_timeout_sec"] == 3600.0
+    assert "benchmark" not in seen
+
+
+@pytest.mark.parametrize("executor", ["dragon", "mpi"])
+@pytest.mark.parametrize(
+    "flags,counts",
+    [
+        (["--warmup-rounds", "1", "--measure-rounds", "3"], (1, 3)),
+        (["--measure-rounds", "1"], (0, 1)),
+        (["--warmup-rounds", "0"], (0, 1)),
+    ],
+)
+def test_fit_batch_forwards_benchmark_to_existing_executor(
+    monkeypatch, tmp_path, executor, flags, counts
+):
+    from cuphoton.core.benchmark import BenchmarkOptions
+    from cuphoton.xpois import commands
+
+    seen = {}
+
+    def run(**kwargs):
+        seen.update(kwargs)
+        return None
+
+    monkeypatch.setattr(commands, f"_run_{executor}_image_pair_batch", run)
+    rc = _run_cli(
+        [
+            "fit-batch",
+            "--executor",
+            executor,
+            "--manifest",
+            str(tmp_path / "pairs.json"),
+            "--backend",
+            "cupy",
+            *flags,
+        ]
+    )
+    assert rc == 0
+    assert seen["benchmark"] == BenchmarkOptions(*counts)
+
+
+@pytest.mark.parametrize(
+    "flags",
+    [
+        ["--measure-rounds", "0"],
+        ["--warmup-rounds", "-1"],
+        [
+            "--measure-rounds",
+            "2",
+            "--aggregation-mode",
+            "files",
+            "--name",
+            "repeated",
+            "--attempt-id",
+            "attempt-1",
+        ],
+    ],
+)
+def test_fit_batch_rejects_invalid_round_options_before_executor(
+    monkeypatch, tmp_path, flags
+):
+    from cuphoton.xpois import commands
+
+    def unexpected(**kwargs):
+        pytest.fail("invalid rounds reached runtime")
+
+    monkeypatch.setattr(commands, "_run_mpi_image_pair_batch", unexpected)
+    assert (
+        _run_cli(
+            [
+                "fit-batch",
+                "--executor",
+                "mpi",
+                "--manifest",
+                str(tmp_path / "pairs.json"),
+                "--backend",
+                "cupy",
+                *flags,
+            ]
+        )
+        != 0
+    )
 
 
 def test_fit_batch_dispatches_explicit_dragon_limits(
diff --git a/tests/xpois/test_dragon.py b/tests/xpois/test_dragon.py
index c5029935..9942c6ad 100644
--- a/tests/xpois/test_dragon.py
+++ b/tests/xpois/test_dragon.py
@@ -1404,6 +1404,24 @@ def test_terminal_loader_exposes_unexpected_record_files(tmp_path) -> None:
     assert audit["unexpected_item_ids"] == ["unexpected"]
 
 
+def test_terminal_loader_rejects_swapped_record_names(tmp_path) -> None:
+    for filename, item_id in (("one", "two"), ("two", "one")):
+        atomic_write_json(
+            tmp_path / "records" / f"{filename}.json",
+            {"item_id": item_id, "status": "success"},
+        )
+
+    records, errors = _load_terminal_records(tmp_path)
+
+    assert records == []
+    assert [error["record_path"] for error in errors] == [
+        "records/one.json",
+        "records/two.json",
+    ]
+    assert all("item_id differs" in error["message"] for error in errors)
+    assert all(not error.get("retryable") for error in errors)
+
+
 def test_terminal_artifact_wait_absorbs_visibility_delay(
     monkeypatch, tmp_path
 ) -> None:
@@ -2732,7 +2750,17 @@ def validate(record):
 
 
 @pytest.mark.parametrize("failure", ["digest", "size", "symlink"])
-def test_launch_worker_retains_descriptor_failures(tmp_path, failure) -> None:
+@pytest.mark.parametrize("rounds", [False, True])
+@pytest.mark.parametrize("identity_failure", [False, True])
+def test_launch_worker_retains_descriptor_failures(
+    monkeypatch, tmp_path, failure, rounds, identity_failure
+) -> None:
+    def process_id():
+        if identity_failure:
+            raise RuntimeError("process identity unavailable")
+        return 1000
+
+    monkeypatch.setattr(dragon_module, "_dragon_process_id", process_id)
     run_dir = tmp_path / "run"
     descriptor = run_dir / "launch/worker-0000.json"
     atomic_write_json(descriptor, {"invalid": True}, overwrite=False)
@@ -2759,10 +2787,21 @@ def test_launch_worker_retains_descriptor_failures(tmp_path, failure) -> None:
                 "descriptor_bytes": descriptor_bytes,
             },
             results,
+            *([_FakeQueue()] if rounds else []),
         )
     result = results.get_nowait()
     assert result["status"] == "failed"
     assert result["worker_id"] == 0
+    if rounds:
+        assert result["kind"] == "ready"
+        assert result["run_id"] == "run"
+        assert result["round_id"] is None
+        assert result["puid"] == (None if identity_failure else 1000)
+        if identity_failure:
+            assert result["identity_error"]["message"] == (
+                "process identity unavailable"
+            )
+    assert result["error"]["type"] == "ValueError"
     receipt = json.loads(
         (run_dir / "launch/worker-0000-failure.json").read_text()
     )
@@ -2814,3 +2853,619 @@ def init(self):
         }
     ]
     assert UninitializedGroup.last_closed
+
+
+def _install_round_runtime(
+    monkeypatch, *, mutate=None, close_failure=False, template=_FakeTemplate
+):
+    """Run persistent targets in threads with CUDA and Dragon faked."""
+    import threading
+
+    state = SimpleNamespace(
+        groups=[],
+        queues=[],
+        calls=[],
+        preflights=[],
+        initialized=[],
+        local=threading.local(),
+    )
+
+    class RoundQueue(_FakeQueue):
+        def __init__(self, maxsize=0, policy=None):
+            super().__init__(maxsize=maxsize)
+            self.policy = policy
+            self.closed = False
+            state.queues.append(self)
+
+        def put(self, value, block=True, timeout=None):
+            values = [value]
+            if self.policy is None and mutate is not None:
+                values = mutate(value)
+            for message in values:
+                super().put(message, block=block, timeout=timeout)
+
+        def close(self):
+            self.closed = True
+            if close_failure and self.policy is None:
+                raise RuntimeError("synthetic result queue close failure")
+
+    class RoundGroup(_FakeGroup):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.threads = []
+            self.stopped = False
+            self.closed = False
+            self.join_timeout = None
+            state.groups.append(self)
+
+        def start(self):
+            def invoke(index, template):
+                state.local.puid = 1000 + index
+                try:
+                    template.target(*template.args)
+                except BaseException:
+                    self.exit_status.append((1000 + index, 1))
+                else:
+                    self.exit_status.append((1000 + index, 0))
+
+            for index, template in enumerate(self.templates):
+                thread = threading.Thread(
+                    target=invoke, args=(index, template)
+                )
+                thread.start()
+                self.threads.append(thread)
+
+        def join(self, timeout=None):
+            self.join_timeout = timeout
+            deadline = dragon_module.time.monotonic() + timeout
+            for thread in self.threads:
+                thread.join(max(0, deadline - dragon_module.time.monotonic()))
+            if any(thread.is_alive() for thread in self.threads):
+                raise TimeoutError("fake group join timed out")
+
+        def stop(self, patience=5.0):
+            self.stopped = True
+            for template in self.templates:
+                try:
+                    template.args[5].put_nowait(None)
+                except queue.Full:
+                    pass
+            for thread in self.threads:
+                thread.join(patience)
+            assert not any(thread.is_alive() for thread in self.threads)
+
+        def close(self, patience=5.0):
+            self.closed = True
+
+    def preflight(
+        placement, *, require_clean_cuda_imports, allow_loopback_alias
+    ):
+        del allow_loopback_alias
+        state.local.placement = placement
+        state.preflights.append(
+            (placement.worker_id, require_clean_cuda_imports)
+        )
+        return placement.host, str(placement.gpu_id)
+
+    def identity(backend):
+        placement = state.local.placement
+        state.initialized.append(placement.worker_id)
+        return _valid_shard_result((), placement, backend=backend)[
+            "provenance"
+        ]["gpu"]
+
+    def item_runner(item, output_dir, options):
+        state.calls.append(
+            (
+                state.local.placement.worker_id,
+                output_dir.parent.parent.name,
+                threading.get_ident(),
+            )
+        )
+        output_dir.mkdir(parents=True)
+        atomic_write_json(output_dir / "summary.json", {})
+        return {
+            "run_dir": str(output_dir),
+            "summary_path": str(output_dir / "summary.json"),
+            "solver": options.solver,
+            "requested_backend": options.backend,
+            "backend": options.backend,
+            "device": "fake-gpu",
+            "runtime": {},
+            "timings_sec": {"solve": 0.1},
+            "wall_sec": {"item_runner": 0.2},
+        }
+
+    monkeypatch.setattr(
+        dragon_module, "_validate_worker_placement", preflight
+    )
+    monkeypatch.setattr(dragon_module, "_collect_gpu_identity", identity)
+    monkeypatch.setattr(
+        dragon_module, "_dragon_process_id", lambda: state.local.puid
+    )
+    monkeypatch.setattr(
+        dragon_module.socket, "gethostname", lambda: "fake-node"
+    )
+    monkeypatch.setattr(dragon_module, "run_image_pair_item", item_runner)
+    monkeypatch.setattr(
+        dragon_module,
+        "_load_dragon_api",
+        lambda: _DragonAPI(
+            System=_FakeSystem,
+            Node=_FakeNode,
+            Policy=_FakePolicy,
+            ProcessGroup=RoundGroup,
+            ProcessTemplate=template,
+            Queue=RoundQueue,
+        ),
+    )
+    return state
+
+
+def _run_round_test(
+    tmp_path,
+    *,
+    warmup=1,
+    measure=2,
+    worker_count=2,
+    worker_timeout=10.0,
+    item_ids=("one", "two"),
+):
+    from cuphoton.core.benchmark import BenchmarkOptions
+
+    manifest = _write_fake_manifest(tmp_path, item_ids=item_ids)
+    return run_dragon_image_pair_batch(
+        manifest_path=manifest,
+        output_root=tmp_path / "runs",
+        run_id="repeated",
+        max_workers=worker_count,
+        result_timeout_sec=0.2,
+        worker_timeout_sec=worker_timeout,
+        options=_options(),
+        benchmark=BenchmarkOptions(
+            warmup_rounds=warmup, measure_rounds=measure
+        ),
+    )
+
+
+def test_benchmark_keeps_placed_workers_alive_and_audits_all_rounds(
+    monkeypatch, tmp_path
+):
+    state = _install_round_runtime(monkeypatch)
+    result = _run_round_test(tmp_path)
+
+    assert result.status == "success", result.summary
+    assert len(state.groups) == 1
+    assert len(state.groups[0].templates) == 2
+    assert state.groups[0].closed and not state.groups[0].stopped
+    assert state.groups[0].join_timeout == 0.2
+    assert result.summary["join_timeout_sec"] == 0.2
+    assert (
+        result.summary["schema"]
+        == "cuphoton.xpois.dragon-benchmark-summary/v1"
+    )
+    assert all(channel.closed for channel in state.queues)
+    assert sorted(state.initialized) == [0, 1]
+    assert [
+        channel.policy.host_name for channel in state.queues if channel.policy
+    ] == ["fake-node", "fake-node"]
+    expected_rounds = ["warmup-0000", "measure-0000", "measure-0001"]
+    for worker_id in range(2):
+        calls = [call for call in state.calls if call[0] == worker_id]
+        assert [call[1] for call in calls] == expected_rounds
+        assert len({call[2] for call in calls}) == 1
+        assert [
+            clean for worker, clean in state.preflights if worker == worker_id
+        ] == [True, False, False, False]
+    report = result.summary["benchmark"]
+    assert [
+        receipt["round_id"] for receipt in report["rounds"]
+    ] == expected_rounds
+    assert report["measured_batch_wall_sec"] is not None
+    assert result.summary["coordinator_timings_sec"]["readiness_sec"] > 0
+    record_run_ids = []
+    for receipt in report["rounds"]:
+        summary = json.loads(
+            (result.run_dir / receipt["summary_path"]).read_text()
+        )
+        assert summary["terminal_record_audit"]["ok"]
+        assert summary["shard_result_audit"]["ok"]
+        assert summary["parent_run_id"] == result.run_id
+        assert not (
+            result.run_dir / "rounds" / receipt["round_id"] / "receipts"
+        ).exists()
+        record = json.loads(
+            (
+                result.run_dir
+                / "rounds"
+                / receipt["round_id"]
+                / "records"
+                / "one.json"
+            ).read_text()
+        )
+        record_run_ids.append(record["run_id"])
+        assert record["run_id"] == summary["run_id"]
+    assert len(set(record_run_ids)) == 3
+    assert not (result.run_dir / "items").exists()
+
+
+def test_benchmark_preserves_large_file_backed_workload(
+    monkeypatch, tmp_path
+):
+    state = _install_round_runtime(monkeypatch)
+    item_ids = tuple(f"image-{index:04d}" for index in range(512))
+    result = _run_round_test(
+        tmp_path,
+        warmup=0,
+        measure=1,
+        worker_count=1,
+        item_ids=item_ids,
+    )
+
+    assert result.status == "success", result.summary
+    template = state.groups[0].templates[0]
+    assert template.target is dragon_module._dragon_launch_worker
+    assert len(template.args) == 6
+    assert len(repr(template.args)) < 2000
+    descriptor_path = Path(template.args[1])
+    assert descriptor_path.stat().st_size > 131_072
+    descriptor = json.loads(descriptor_path.read_text())
+    assert [item["item_id"] for item in descriptor["items"]] == list(item_ids)
+    assert descriptor["benchmark"] == {
+        "warmup_rounds": 0,
+        "measure_rounds": 1,
+    }
+    assert descriptor["worker_target"] == {
+        "module": dragon_module.__name__,
+        "qualname": "_dragon_round_worker",
+    }
+    assert len(state.calls) == 512
+    assert state.initialized == [0]
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is not None
+
+
+def test_benchmark_rejects_oversized_arguments_before_init(
+    monkeypatch, tmp_path
+):
+    class OversizedTemplate(_FakeTemplate):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.argdata = b"x" * (96 * 1024 + 1)
+
+    state = _install_round_runtime(monkeypatch, template=OversizedTemplate)
+    monkeypatch.setattr(
+        dragon_module._load_dragon_api().ProcessGroup,
+        "init",
+        lambda _: pytest.fail("oversized arguments must fail before init"),
+    )
+    result = _run_round_test(tmp_path, worker_count=1)
+
+    assert result.status == "failed"
+    assert result.summary["lifecycle_errors"][0] == {
+        "phase": "process_setup",
+        "type": "ValueError",
+        "message": "Dragon worker launch arguments exceed 96 KiB",
+    }
+    assert result.summary["benchmark"]["rounds"] == []
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+    assert not state.calls and not state.initialized
+    assert state.groups[0].closed
+    assert all(channel.closed for channel in state.queues)
+
+
+@pytest.mark.parametrize(
+    "corruption",
+    [
+        "ready-duplicate",
+        "wrong-round",
+        "missing-round",
+        "failed-round",
+        "changed-worker",
+        "invalid-duration",
+        "ready-bad-provenance",
+    ],
+)
+def test_benchmark_rejects_bad_receipts_and_preserves_partial_evidence(
+    monkeypatch, tmp_path, corruption
+):
+    def mutate(message):
+        if (
+            corruption == "ready-bad-provenance"
+            and message["kind"] == "ready"
+        ):
+            return [{**message, "provenance": {}}]
+        if corruption == "ready-duplicate" and message["kind"] == "ready":
+            return [message, message]
+        if message["kind"] == "round":
+            if corruption == "wrong-round":
+                return [{**message, "round_id": "measure-9999"}]
+            if corruption == "missing-round":
+                return []
+            if corruption == "failed-round":
+                return [{**message, "status": "failed"}]
+            if corruption == "changed-worker":
+                result = dict(message["result"])
+                result["provenance"] = {
+                    **result["provenance"],
+                    "pid": 99999999,
+                }
+                return [{**message, "result": result}]
+            if corruption == "invalid-duration":
+                return [{**message, "worker_wall_sec": True}]
+        return [message]
+
+    state = _install_round_runtime(monkeypatch, mutate=mutate)
+    result = _run_round_test(
+        tmp_path,
+        worker_count=1,
+        worker_timeout=0.5 if corruption == "missing-round" else 10.0,
+    )
+
+    assert result.status == "failed"
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+    assert result.summary["lifecycle_errors"]
+    assert state.groups[0].stopped and state.groups[0].closed
+    assert all(channel.closed for channel in state.queues)
+    assert len(result.summary["benchmark"]["rounds"]) <= 1
+    assert (result.run_dir / "summary.json").is_file()
+    assert not any(thread.is_alive() for thread in state.groups[0].threads)
+    if corruption == "ready-bad-provenance":
+        assert result.summary["lifecycle_errors"][0] == {
+            "phase": "ready",
+            "type": "ValueError",
+            "message": "invalid Dragon READY provenance",
+        }
+        assert result.summary["benchmark"]["rounds"] == []
+        assert not state.calls
+
+
+def test_benchmark_cleanup_failure_invalidates_successful_rounds(
+    monkeypatch, tmp_path
+):
+    _install_round_runtime(monkeypatch, close_failure=True)
+    result = _run_round_test(tmp_path, warmup=0, measure=1)
+
+    assert result.status == "failed"
+    assert result.summary["benchmark"]["rounds"][0]["status"] == "success"
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+    assert any(
+        error["phase"] == "queue_close"
+        for error in result.summary["lifecycle_errors"]
+    )
+
+
+def test_benchmark_notifies_startup_artifact_failure(monkeypatch, tmp_path):
+    state = _install_round_runtime(monkeypatch)
+    original_write = dragon_module.atomic_write_json
+
+    def fail_startup(path, payload, **kwargs):
+        if path.parent.name == "startup":
+            raise OSError("synthetic startup write failure")
+        original_write(path, payload, **kwargs)
+
+    monkeypatch.setattr(dragon_module, "atomic_write_json", fail_startup)
+    result = _run_round_test(tmp_path, worker_count=1)
+
+    assert result.status == "failed"
+    assert result.summary["ready_messages"][0]["status"] == "failed"
+    assert "artifact_error" in result.summary["ready_messages"][0]
+    assert not state.calls
+    assert state.groups[0].stopped
+
+
+@pytest.mark.parametrize("exit_code", [0, -9])
+def test_round_collector_detects_receiptless_exit_without_deadline_wait(
+    exit_code,
+):
+    class EmptyResults:
+        def get(self, *, timeout):
+            assert 0 <= timeout <= dragon_module._WORKER_POLL_SEC
+            raise queue.Empty
+
+    with pytest.raises(RuntimeError, match="exited before ready completion"):
+        dragon_module._collect_worker_messages(
+            EmptyResults(),
+            [],
+            worker_count=1,
+            run_id="run",
+            kind="ready",
+            round_id=None,
+            deadline=dragon_module.time.monotonic() + 30,
+            group=SimpleNamespace(inactive_puids=[(1000, exit_code)]),
+        )
+
+
+def test_round_collector_drains_receipt_delivered_during_exit_check():
+    receipt = {
+        "kind": "ready",
+        "run_id": "run",
+        "round_id": None,
+        "worker_id": 0,
+        "puid": 1000,
+        "status": "success",
+    }
+
+    class RacingResults:
+        delivered = False
+
+        def get(self, *, timeout):
+            if self.delivered:
+                assert timeout == 0
+                return receipt
+            raise queue.Empty
+
+    results = RacingResults()
+
+    class ExitedGroup:
+        @property
+        def inactive_puids(self):
+            results.delivered = True
+            return [(1000, 0)]
+
+    messages = []
+    dragon_module._collect_worker_messages(
+        results,
+        messages,
+        worker_count=1,
+        run_id="run",
+        kind="ready",
+        round_id=None,
+        deadline=dragon_module.time.monotonic() + 30,
+        group=ExitedGroup(),
+    )
+    assert messages == [receipt]
+
+
+def test_round_collector_waits_for_peers_after_reported_worker_exit():
+    failed = {
+        "kind": "ready",
+        "run_id": "run",
+        "round_id": None,
+        "worker_id": 0,
+        "puid": 1000,
+        "status": "failed",
+    }
+    healthy = {**failed, "worker_id": 1, "puid": 1001, "status": "success"}
+
+    class Results:
+        pending = iter((failed, None, None, healthy))
+
+        def get(self, *, timeout):
+            message = next(self.pending)
+            if message is None:
+                raise queue.Empty
+            return message
+
+    messages = []
+    with pytest.raises(RuntimeError, match=r"workers \[0\] ready failed"):
+        dragon_module._collect_worker_messages(
+            Results(),
+            messages,
+            worker_count=2,
+            run_id="run",
+            kind="ready",
+            round_id=None,
+            deadline=dragon_module.time.monotonic() + 30,
+            group=SimpleNamespace(inactive_puids=[(1000, 1)]),
+        )
+    assert messages == [failed, healthy]
+
+
+@pytest.mark.parametrize("phase", ["ready", "round"])
+def test_benchmark_detects_worker_exit_without_receipt(
+    monkeypatch, tmp_path, phase
+):
+    state = _install_round_runtime(monkeypatch)
+    monkeypatch.setattr(dragon_module, "_WORKER_POLL_SEC", 0.01)
+    attribute = (
+        "_collect_gpu_identity" if phase == "ready" else "run_image_pair_item"
+    )
+    original = getattr(dragon_module, attribute)
+
+    def exit_worker(*args, **kwargs):
+        if state.local.placement.worker_id == 0:
+            raise SystemExit("synthetic abrupt worker exit")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(dragon_module, attribute, exit_worker)
+    result = _run_round_test(tmp_path, warmup=0, measure=1, worker_timeout=2)
+
+    assert result.status == "failed"
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+    if phase == "ready":
+        errors = result.summary["lifecycle_errors"]
+    else:
+        summary = json.loads(
+            (result.run_dir / "rounds/measure-0000/summary.json").read_text()
+        )
+        errors = summary["errors"]
+    assert any(
+        f"exited before {phase} completion" in error["message"]
+        for error in errors
+    )
+    assert not any(thread.is_alive() for thread in state.groups[0].threads)
+
+
+def test_benchmark_finishes_healthy_shard_after_peer_failure(
+    monkeypatch, tmp_path
+):
+    import threading
+
+    state = _install_round_runtime(monkeypatch)
+    monkeypatch.setattr(dragon_module, "_WORKER_POLL_SEC", 0.01)
+    original_runner = dragon_module.run_image_pair_item
+    healthy_can_finish = threading.Event()
+    failed_receipt_read = threading.Event()
+    results_type = dragon_module._load_dragon_api().Queue
+    original_get = results_type.get
+
+    def get(channel, *args, **kwargs):
+        if channel.policy is None and failed_receipt_read.is_set():
+            healthy_can_finish.set()
+        message = original_get(channel, *args, **kwargs)
+        if (
+            channel.policy is None
+            and message.get("kind") == "round"
+            and message.get("status") == "failed"
+        ):
+            failed_receipt_read.set()
+        return message
+
+    def run_item(item, output_dir, options):
+        if state.local.placement.worker_id == 0:
+            raise RuntimeError("synthetic failed item")
+        assert healthy_can_finish.wait(2), (
+            "coordinator stopped collecting receipts"
+        )
+        return original_runner(item, output_dir, options)
+
+    monkeypatch.setattr(results_type, "get", get)
+    monkeypatch.setattr(dragon_module, "run_image_pair_item", run_item)
+    result = _run_round_test(tmp_path, warmup=0, measure=2, worker_timeout=3)
+
+    assert result.status == "failed"
+    assert len(result.summary["benchmark"]["rounds"]) == 1
+    summary = json.loads(
+        (result.run_dir / "rounds/measure-0000/summary.json").read_text()
+    )
+    assert sorted(
+        message["worker_id"] for message in summary["messages"]
+    ) == [0, 1]
+    assert summary["shard_result_audit"]["missing_worker_ids"] == []
+    assert summary["terminal_record_audit"]["missing_item_ids"] == []
+    records = list(
+        (result.run_dir / "rounds/measure-0000/records").glob("*.json")
+    )
+    assert sorted(
+        json.loads(path.read_text())["status"] for path in records
+    ) == [
+        "failed",
+        "success",
+    ]
+    assert not any(thread.is_alive() for thread in state.groups[0].threads)
+
+
+@pytest.mark.parametrize("exit_status", [[], [(1000, -9)]])
+def test_benchmark_exit_audit_invalidates_successful_round(
+    monkeypatch, tmp_path, exit_status
+):
+    state = _install_round_runtime(monkeypatch)
+    group_type = dragon_module._load_dragon_api().ProcessGroup
+    original_join = group_type.join
+
+    def join(group, timeout=None):
+        original_join(group, timeout=timeout)
+        group.exit_status = exit_status
+
+    monkeypatch.setattr(group_type, "join", join)
+    result = _run_round_test(tmp_path, warmup=0, measure=1, worker_count=1)
+
+    assert result.status == "failed"
+    assert result.summary["benchmark"]["rounds"][0]["status"] == "success"
+    assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+    assert result.summary["process_exit_audit"]["ok"] is False
+    assert any(
+        error["phase"] == "exit_status"
+        for error in result.summary["lifecycle_errors"]
+    )
+    assert state.groups[0].join_timeout == 0.2
diff --git a/tests/xpois/test_mpi.py b/tests/xpois/test_mpi.py
index 8fe1b522..910f9cfe 100644
--- a/tests/xpois/test_mpi.py
+++ b/tests/xpois/test_mpi.py
@@ -10,12 +10,13 @@
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import replace
 from pathlib import Path
-from threading import Barrier, Event
+from threading import Barrier, Event, local
 from types import SimpleNamespace
 
 import numpy as np
 import pytest
 
+from cuphoton.core.benchmark import BenchmarkOptions
 from cuphoton.core.bulk import WorkItem
 from cuphoton.xpois import mpi
 from cuphoton.xpois.batch import BatchFitOptions
@@ -252,7 +253,7 @@ def _stage_file_rank(
                 "status": "failed",
                 "error": {"type": "OSError", "message": "cannot write"},
             },
-            "cannot persist MPI aggregate",
+            "cannot persist MPI aggregate: .*OSError.*cannot write",
         ),
     ],
 )
@@ -1025,6 +1026,10 @@ def test_file_preflight_peer_failure_does_not_wait_for_missing_rank(
     failure = {"type": "ValueError", "message": "peer preflight failed"}
     published = Barrier(2)
     original_publish = mpi._publish_file_preflight
+    original_consensus = mpi._file_preflight_consensus
+    original_wait = mpi._shared_filesystem_wait
+    state = local()
+    consensus_checked = Event()
 
     def publish(*args, **kwargs) -> None:
         original_publish(*args, **kwargs)
@@ -1032,6 +1037,24 @@ def publish(*args, **kwargs) -> None:
 
     monkeypatch.setattr(mpi, "_publish_file_preflight", publish)
 
+    def consensus(*args, **kwargs):
+        consensus_checked.set()
+        state.checking_consensus = True
+        try:
+            return original_consensus(*args, **kwargs)
+        finally:
+            state.checking_consensus = False
+
+    def wait(deadline, delay):
+        # The peer may wait for the terminal marker; consensus must not
+        # wait for the absent third rank after both records are published.
+        if getattr(state, "checking_consensus", False):
+            pytest.fail("preflight consensus waited after a peer failure")
+        return original_wait(deadline, delay)
+
+    monkeypatch.setattr(mpi, "_file_preflight_consensus", consensus)
+    monkeypatch.setattr(mpi, "_shared_filesystem_wait", wait)
+
     def prepare(rank: int) -> str:
         try:
             mpi._file_prepare_run(
@@ -1051,11 +1074,10 @@ def prepare(rank: int) -> str:
             return str(exc)
         return "unexpected success"
 
-    start = time.perf_counter()
     with ThreadPoolExecutor(max_workers=2) as pool:
         outcomes = list(pool.map(prepare, (0, 1)))
 
-    assert time.perf_counter() - start < 1.0
+    assert consensus_checked.is_set()
     assert all("peer preflight failed" in outcome for outcome in outcomes)
     marker = mpi.read_json_mapping(
         mpi._attempt_path(run_dir, run_id, attempt_id)
@@ -2909,6 +2931,300 @@ def load_api():
     assert run_record["mpi4py_version"] == "4.test"
 
 
+@pytest.mark.parametrize("readiness_failure", [False, True])
+def test_benchmark_reuses_rank_and_gpu_with_audited_round_artifacts(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, readiness_failure: bool
+) -> None:
+    _mpi_environment(monkeypatch)
+    _remove_cuda_modules(monkeypatch)
+    manifest = _write_manifest(tmp_path)
+    comm = _SingletonComm()
+    loads = []
+    identities = []
+    executions = []
+    clock = [0.0]
+    real_wait = mpi._wait_collective_artifacts
+    real_prepare = mpi._mpi_prepare_run
+
+    def load_api():
+        loads.append(True)
+        return mpi._MPIAPI(
+            SimpleNamespace(COMM_TYPE_SHARED=1), comm, "4.test", "Test MPI"
+        )
+
+    def gpu_identity(backend):
+        identities.append(backend)
+        clock[0] += 20.0
+        if readiness_failure:
+            raise RuntimeError("GPU initialization failed")
+        return _gpu()
+
+    def execute(item, output, options):
+        executions.append(output)
+        clock[0] += 2.0
+        return _item_runner(item, output, options)
+
+    def audit(*args, **kwargs):
+        clock[0] += 100.0
+        return real_wait(*args, **kwargs)
+
+    def prepare(*args, **kwargs):
+        if args[2].parent.name == "rounds":
+            clock[0] += 5.0
+        return real_prepare(*args, **kwargs)
+
+    monkeypatch.setattr(mpi, "_load_mpi_api", load_api)
+    monkeypatch.setattr(mpi, "_gpu_identity", gpu_identity)
+    monkeypatch.setattr(mpi, "run_image_pair_item", execute)
+    monkeypatch.setattr(mpi, "_wait_collective_artifacts", audit)
+    monkeypatch.setattr(mpi, "_mpi_prepare_run", prepare)
+    monkeypatch.setattr(
+        mpi,
+        "time",
+        SimpleNamespace(
+            perf_counter=lambda: clock[0],
+            monotonic=time.monotonic,
+            sleep=time.sleep,
+        ),
+    )
+    benchmark = BenchmarkOptions(warmup_rounds=1, measure_rounds=2)
+    result = mpi.run_mpi_image_pair_batch(
+        manifest_path=manifest,
+        output_root=tmp_path / "runs",
+        run_id="repeated",
+        aggregation_mode="mpi",
+        rank_timeout_sec=None,
+        attempt_id=None,
+        options=_options(),
+        benchmark=benchmark,
+    )
+
+    assert result is not None
+    assert loads == [True]
+    assert identities == ["cupy"]
+    assert len(comm.split_calls) == 1
+    if readiness_failure:
+        assert result.status == "failed"
+        assert result.summary["benchmark"]["rounds"] == []
+        assert result.summary["benchmark"]["measured_batch_wall_sec"] is None
+        assert "GPU initialization failed" in str(result.summary)
+        assert executions == []
+        return
+    assert result.status == "success"
+    assert result.summary["startup_ready_sec"] == 20.0
+    report = result.summary["benchmark"]
+    assert report["measured_batch_wall_sec"] == {
+        "min": 2.0,
+        "median": 2.0,
+        "max": 2.0,
+    }
+    assert len(executions) == 3 and len(set(executions)) == 3
+    pids = set()
+    for planned, round_result in zip(benchmark.rounds(), report["rounds"]):
+        assert round_result["round_id"] == planned.round_id
+        assert round_result["batch_wall_sec"] == 2.0
+        assert round_result["worker_wall_max_sec"] == 2.0
+        summary = mpi.read_json_mapping(
+            result.run_dir / round_result["summary_path"]
+        )
+        assert summary["run_id"] == planned.run_id("repeated")
+        assert summary["terminal_record_audit"]["ok"] is True
+        assert summary["rank_result_audit"]["ok"] is True
+        pids.add(summary["rank_results"][0]["provenance"]["pid"])
+    assert len(pids) == 1
+    assert result.summary["coordinator_wall_sec"] == 341.0
+
+
+@pytest.mark.parametrize("benchmark", [None, BenchmarkOptions()])
+def test_benchmark_plan_consensus_rejects_mismatched_enablement(
+    benchmark: BenchmarkOptions | None,
+) -> None:
+    class DivergentComm:
+        def gather(self, value, root):
+            other_plan = (
+                None if benchmark else BenchmarkOptions().to_payload()
+            )
+            return [value, {**value, "rank": 1, "benchmark": other_plan}]
+
+        def bcast(self, value, root):
+            return value
+
+    with pytest.raises(RuntimeError, match="benchmark options differ"):
+        mpi._mpi_manifest_consensus(
+            DivergentComm(),
+            mpi._RankContext(0, 0, 2, "host", "test"),
+            SimpleNamespace(sha256="manifest"),
+            None,
+            "run",
+            Path("/shared/run"),
+            _options(),
+            "4.test",
+            "Test MPI",
+            benchmark=benchmark,
+        )
+
+
+def test_benchmark_rejects_file_aggregation_before_loading_runtime(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setattr(
+        mpi, "_load_mpi_api", lambda: pytest.fail("loaded MPI")
+    )
+    with pytest.raises(ValueError, match="require aggregation_mode='mpi'"):
+        mpi.run_mpi_image_pair_batch(
+            manifest_path=tmp_path / "unused.json",
+            output_root=tmp_path / "runs",
+            run_id="repeated",
+            aggregation_mode="files",
+            rank_timeout_sec=1.0,
+            attempt_id="attempt",
+            options=_options(),
+            benchmark=BenchmarkOptions(),
+        )
+    assert not (tmp_path / "runs").exists()
+
+
+def test_benchmark_invalid_options_enter_startup_consensus(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    comm = _SingletonComm()
+    monkeypatch.setattr(
+        mpi, "_load_mpi_api", lambda: mpi._MPIAPI(None, comm, "4.test", "MPI")
+    )
+    with pytest.raises(RuntimeError, match="rank-startup.*BenchmarkOptions"):
+        mpi.run_mpi_image_pair_batch(
+            manifest_path=tmp_path / "unused.json",
+            output_root=tmp_path / "runs",
+            run_id="repeated",
+            aggregation_mode="mpi",
+            rank_timeout_sec=None,
+            attempt_id=None,
+            options=_options(),
+            benchmark=object(),
+        )
+    assert comm.gather_calls == 1
+    assert not (tmp_path / "runs").exists()
+
+
+@pytest.mark.parametrize(
+    "failure_round", [None, "warmup-0000", "measure-0000", "invalid-receipt"]
+)
+def test_benchmark_two_ranks_stop_together_and_preserve_failure(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure_round: str | None
+) -> None:
+    manifest = mpi.load_image_pair_manifest(_write_manifest(tmp_path, 2))
+    shards = mpi.partition_byte_balanced(manifest.work_items(), 2)
+    run_dir = tmp_path / "run"
+    run_dir.mkdir()
+    barrier = Barrier(2, timeout=5)
+    shared = SimpleNamespace(gathered=[None, None], broadcast=None)
+    rank_local = local()
+    identities = []
+    executed = []
+
+    class ConcurrentComm:
+        def __init__(self, rank):
+            self.rank = rank
+
+        def Get_rank(self):
+            return self.rank
+
+        def Get_size(self):
+            return 2
+
+        def gather(self, value, root):
+            shared.gathered[self.rank] = value
+            barrier.wait()
+            result = list(shared.gathered) if self.rank == root else None
+            barrier.wait()
+            return result
+
+        def bcast(self, value, root):
+            if self.rank == root:
+                shared.broadcast = value
+            barrier.wait()
+            result = shared.broadcast
+            barrier.wait()
+            return result
+
+    def identity(backend):
+        rank = rank_local.rank
+        identities.append(rank)
+        return _gpu(f"GPU-{rank}")
+
+    def execute(item, output, options):
+        executed.append((rank_local.rank, output.parent.parent.name))
+        if (
+            rank_local.rank == 1
+            and output.parent.parent.name == failure_round
+        ):
+            raise RuntimeError("scientific work failed")
+        return _item_runner(item, output, options)
+
+    monkeypatch.setattr(mpi, "_gpu_identity", identity)
+    monkeypatch.setattr(mpi, "run_image_pair_item", execute)
+    if failure_round == "invalid-receipt":
+        aggregate = mpi._mpi_aggregate
+
+        def invalid_receipt(*args, **kwargs):
+            result = aggregate(*args, **kwargs)
+            if result is not None:
+                return replace(result, summary_path=tmp_path / "outside.json")
+            return result
+
+        monkeypatch.setattr(mpi, "_mpi_aggregate", invalid_receipt)
+    benchmark = BenchmarkOptions(warmup_rounds=1, measure_rounds=2)
+
+    def run(rank):
+        rank_local.rank = rank
+        context = mpi._RankContext(rank, rank, 2, "host", "test")
+        try:
+            return mpi._run_mpi_benchmark(
+                mpi._MPIAPI(None, ConcurrentComm(rank), "4.test", "Test MPI"),
+                context,
+                run_dir,
+                "concurrent",
+                manifest,
+                _options(),
+                shards,
+                str(rank),
+                benchmark,
+                mpi.timestamp_utc(),
+                time.perf_counter(),
+                0.1,
+            )
+        except Exception as exc:
+            return exc
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        root, peer = list(pool.map(run, range(2)))
+    assert isinstance(root, mpi.MPIBatchResult)
+    assert sorted(identities) == [0, 1]
+    report = root.summary["benchmark"]
+    if failure_round is None:
+        assert root.status == "success"
+        assert peer is None
+        assert len(report["rounds"]) == 3
+        assert len(executed) == 6
+    else:
+        assert root.status == "failed"
+        assert isinstance(peer, RuntimeError)
+        assert "benchmark 'concurrent' failed" in str(peer)
+        if failure_round == "invalid-receipt":
+            assert report["rounds"] == []
+            assert report["errors"][0]["phase"] == "warmup-0000"
+            assert len(executed) == 2
+            return
+        assert report["rounds"][-1]["round_id"] == failure_round
+        assert report["rounds"][-1]["status"] == "failed"
+        assert report["measured_batch_wall_sec"] is None
+        assert len(executed) == len(report["rounds"]) * 2
+        failure = mpi.read_json_mapping(
+            run_dir / "rounds" / failure_round / "records" / "pair-1.json"
+        )
+        assert failure["error"]["message"] == "scientific work failed"
+
+
 def test_collective_generated_run_uses_fully_resolved_output_path(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
@@ -5169,3 +5485,18 @@ def runner(item, output, options):
         assert result.status == "failed"
         assert len(errors) == 1
         assert errors[0]["message"] == "record 0 has invalid field(s): solver"
+
+
+def test_item_record_filenames_bind_the_record_identity(tmp_path):
+    records = tmp_path / "records"
+    records.mkdir()
+    mpi.atomic_write_json(records / "one.json", {"item_id": "two"})
+    mpi.atomic_write_json(records / "two.json", {"item_id": "one"})
+    accepted, errors = mpi._read_mappings(
+        records, validate_item_filename=True
+    )
+    assert accepted == []
+    assert [error["path"] for error in errors] == ["one.json", "two.json"]
+    assert all(
+        "filename does not match" in error["message"] for error in errors
+    )