Measure repeated XPOIS batches in persistent Dragon and MPI workers

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

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

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