diff --git a/AGENTS.md b/AGENTS.md
index 1fa7a18f..35bcd09e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -84,6 +84,10 @@ Key files:
   change to the recorded sections has to land on both sides of the package boundary. Benchmark
   names are Bencher's history key, so renaming or moving a `bench_*` test orphans its tracked
   series.
+- **`rounds > 1` overlaps two rounds' live memory** (`setup=` runs before the prior round's
+  teardown) — pin `--bench-rounds=1`; `record_memory` measures that construction transient,
+  not per-op cost (use `op_memory`).
+- **Peak memory is `HighWaterMark`, not sampled PSS** — exact, unlike `/proc/self/smaps_rollup`.
 
 ### Core abstractions (the propagation backbone)
 
@@ -137,6 +141,8 @@ mp = MajoranaPropagator(operator, initial_state, cutoff=4)
 - Fixture msgpack schema is documented in `tests/data/README.md`
 - Tests validate against exact solutions for small systems
 - Heavy use of `@parametrize_with_cases` decorators
+- **pytest's fd-level capture hides C++ stderr** (e.g. `COMMPROF`) — rerun with `-s` to see it.
+- **A slow CTest run on an MPI build is `MPI_Init` fabric probing, not slow tests** — see `monoprop_TEST_EXCLUDE_MPI_FABRIC` in `cpp/tests/CMakeLists.txt`.
 
 ## Key Dependencies & Integration
 
@@ -176,5 +182,7 @@ When changing behavior, APIs, build/test workflows, paths, or developer conventi
 - Check `build/*/compile_commands.json` for compilation flags
 - Use `rm -rf build` to clear environment-specific builds
 - Verify `monoprop_MAX_NUM_MODES` matches your use case (default: 250)
+- **`uv sync` does not relink `bin/monoprop_unit_tests.x`** — check its mtime against the
+  source's; recipe for a standalone C++ build tree in `docs/content/docs/building.mdx`.
 
 This is a sophisticated scientific computing project requiring careful attention to template instantiation, build system configuration, and the C++/Python boundary.
diff --git a/benches/bench_models.py b/benches/bench_models.py
index c45e70f2..7a702335 100644
--- a/benches/bench_models.py
+++ b/benches/bench_models.py
@@ -12,21 +12,37 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Fixed-model benchmarks: heavy, Heisenberg-only, in-place simulations.
+"""Fixed-model benchmarks: heavy, Heisenberg-only simulations.
 
 The 120-qubit Fermi-Hubbard trajectory and the 127-qubit Pauli-basis kicked-Ising
 circuit, at fixed sizes. The registry (config class, builder, steps-per-run) lives
 in :data:`monoprop_bench_tools.models.MODELS`; each config field is overridable via
 ``--<model>-<field>``.
+
+Run one group per pytest process: build_graph/propagate and energy/gradient do not fit a node together.
 """
 
 from __future__ import annotations
 
+import os
 from typing import Any
 
 import pytest
 from monoprop_bench_tools.memory.cpu import resting_rss_bytes
-from monoprop_bench_tools.models import MODELS, barriered
+from monoprop_bench_tools.models import MODELS, barrier_setup, barriered
+
+# `build_graph` extends the graph, so a driver that re-applies its circuit retains one layer-set per step
+MAX_GRAPH_STEPS = 2
+
+
+def skip_if_graph_will_not_fit(model: str, steps: int) -> None:
+    """Skip a graph-holding benchmark whose retained graph is known not to fit."""
+    allow = os.environ.get("monoprop_BENCH_ALLOW_BIG_GRAPH")  # noqa: SIM112
+    if steps > MAX_GRAPH_STEPS and allow != "1":
+        pytest.skip(
+            f"{model}: {steps} successive build_graph calls retain {steps} layer-sets, "
+            f"measured to exceed 242 GiB. Set monoprop_BENCH_ALLOW_BIG_GRAPH=1 to run it."
+        )
 
 
 @pytest.mark.slow
@@ -59,9 +75,138 @@ def run(built, n_steps):
         return propagator.expectation_value()
 
     result = benchmark.pedantic(
-        barriered(run, bench_comm), setup=setup, rounds=1, iterations=1
+        barriered(run, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=1,
+        iterations=1,
     )
     assert isinstance(result, float)
 
     propagator, _circuit = state["built"]
     record_model_stats(model, propagator, state["baseline_rss"])
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("model", list(MODELS))
+def test_model_build_graph(
+    benchmark,
+    bench_comm,
+    bench_rounds,
+    model_configs,
+    model,
+    record_model_config,
+    op_memory,
+    record_opsize,
+):
+    """Benchmark building a fixed model's propagation graph, from a fresh propagator."""
+    _config_cls, build_fn, steps_fn = MODELS[model]
+    config = model_configs[model]
+    steps = steps_fn(config)
+    skip_if_graph_will_not_fit(model, steps)
+    record_model_config(model, config)
+
+    last = []
+
+    def setup():
+        built = build_fn(config, comm=bench_comm)
+        last[:] = [built[0]]
+        op_memory.open()  # inside setup, so the window excludes the construction
+        return (built, steps), {}
+
+    def build(built, n_steps):
+        propagator, circuit = built
+        for _ in range(n_steps):
+            propagator.build_graph(circuit)
+
+    benchmark.pedantic(
+        barriered(build, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=bench_rounds,
+        iterations=1,
+    )
+    op_memory.close(last[0])
+    assert record_opsize(last[0]) > 0
+    assert last[0].graph_layers > 0
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("model", list(MODELS))
+def test_model_propagate(
+    benchmark,
+    bench_comm,
+    bench_rounds,
+    model_configs,
+    model,
+    record_model_config,
+    op_memory,
+    record_opsize,
+):
+    """Benchmark a fixed model's in-place evolution alone -- no expectation value, no graph."""
+    _config_cls, build_fn, steps_fn = MODELS[model]
+    config = model_configs[model]
+    steps = steps_fn(config)
+    record_model_config(model, config)
+
+    last = []
+
+    def setup():
+        built = build_fn(config, comm=bench_comm)
+        last[:] = [built[0]]
+        op_memory.open()
+        return (built, steps), {}
+
+    def run(built, n_steps):
+        propagator, circuit = built
+        for _ in range(n_steps):
+            propagator.propagate(circuit)
+
+    benchmark.pedantic(
+        barriered(run, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=bench_rounds,
+        iterations=1,
+    )
+    op_memory.close(last[0])
+    assert record_opsize(last[0]) > 0
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("model", list(MODELS))
+def test_model_energy(
+    benchmark, model_graph, model, model_configs, bench_comm, bench_rounds, op_memory
+):
+    """Benchmark evaluating a fixed model's expectation-value functional."""
+    skip_if_graph_will_not_fit(model, MODELS[model][2](model_configs[model]))
+    propagator, parameters = model_graph(model)
+    functional = propagator.expectation_value_functional()
+    op_memory.open()
+    result = benchmark.pedantic(
+        barriered(functional, bench_comm),
+        args=(parameters,),
+        setup=barrier_setup(bench_comm),
+        rounds=bench_rounds,
+        iterations=1,
+    )
+    op_memory.close(propagator)
+    assert isinstance(result, float)
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("model", list(MODELS))
+def test_model_gradient(
+    benchmark, model_graph, model, model_configs, bench_comm, bench_rounds, op_memory
+):
+    """Benchmark evaluating a fixed model's expectation-value-and-gradient functional."""
+    skip_if_graph_will_not_fit(model, MODELS[model][2](model_configs[model]))
+    propagator, parameters = model_graph(model)
+    functional = propagator.expectation_value_and_gradient_functional()
+    op_memory.open()
+    _value, gradient = benchmark.pedantic(
+        barriered(functional, bench_comm),
+        args=(parameters,),
+        setup=barrier_setup(bench_comm),
+        rounds=bench_rounds,
+        iterations=1,
+    )
+    op_memory.close(propagator)
+    assert len(gradient) == len(parameters)
diff --git a/benches/bench_random.py b/benches/bench_random.py
index f691bbd8..31f5e75d 100644
--- a/benches/bench_random.py
+++ b/benches/bench_random.py
@@ -16,27 +16,73 @@
 
 from __future__ import annotations
 
-from monoprop_bench_tools.models import barriered
+from monoprop_bench_tools.models import barrier_setup, barriered
 
 PARE_THRESHOLD = 1e-10
 INPLACE_LOWER_ATOL = 1e-5
 
 
 def test_random_build_graph(
-    benchmark, make_random_propagator, bench_comm, bench_rounds
+    benchmark,
+    make_random_propagator,
+    bench_comm,
+    bench_rounds,
+    op_memory,
+    record_opsize,
 ):
     """Benchmark building the propagation graph from a fresh propagator."""
+    last = []
 
     def setup():
-        return (make_random_propagator(),), {}
+        built = make_random_propagator()
+        last[:] = [built[0]]
+        op_memory.open()  # inside setup, so the window excludes the construction
+        return (built,), {}
 
     def build(built):
         propagator, circuit = built
         propagator.build_graph(circuit)
 
     benchmark.pedantic(
-        barriered(build, bench_comm), setup=setup, rounds=bench_rounds, iterations=1
+        barriered(build, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=bench_rounds,
+        iterations=1,
+    )
+    op_memory.close(last[0])
+    assert record_opsize(last[0]) > 0
+    assert last[0].graph_layers > 0
+
+
+def test_random_propagate(
+    benchmark,
+    make_random_propagator,
+    bench_comm,
+    bench_rounds,
+    op_memory,
+    record_opsize,
+):
+    """Benchmark in-place evolution alone, with no expectation value and no graph."""
+    last = []
+
+    def setup():
+        built = make_random_propagator()
+        last[:] = [built[0]]
+        op_memory.open()
+        return (built,), {}
+
+    def run(built):
+        propagator, circuit = built
+        propagator.propagate(circuit)
+
+    benchmark.pedantic(
+        barriered(run, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=bench_rounds,
+        iterations=1,
     )
+    op_memory.close(last[0])
+    assert record_opsize(last[0]) > 0
 
 
 def test_random_pare(benchmark, built_graph, bench_comm, bench_rounds):
@@ -47,34 +93,52 @@ def pare():
             pare_threshold=PARE_THRESHOLD,
         )
 
-    benchmark.pedantic(barriered(pare, bench_comm), rounds=bench_rounds, iterations=1)
+    benchmark.pedantic(
+        barriered(pare, bench_comm),
+        setup=barrier_setup(bench_comm),
+        rounds=bench_rounds,
+        iterations=1,
+    )
 
 
 def test_random_energy(
-    benchmark, built_graph, random_problem, bench_comm, bench_rounds
+    benchmark, built_graph, random_problem, bench_comm, bench_rounds, op_memory
 ):
     """Benchmark evaluating the expectation-value functional."""
     functional = built_graph.expectation_value_functional()
+    op_memory.open()
+
+    # The entry barrier must stay untimed, so it lives in a setup, and a setup that returns args forbids args=.
+    def setup():
+        return (random_problem.parameters,), {}
+
     result = benchmark.pedantic(
         barriered(functional, bench_comm),
-        args=(random_problem.parameters,),
+        setup=barrier_setup(bench_comm, setup),
         rounds=bench_rounds,
         iterations=1,
     )
+    op_memory.close(built_graph)
     assert isinstance(result, float)
 
 
 def test_random_gradient(
-    benchmark, built_graph, random_problem, bench_comm, bench_rounds
+    benchmark, built_graph, random_problem, bench_comm, bench_rounds, op_memory
 ):
     """Benchmark evaluating the expectation-value-and-gradient functional."""
     functional = built_graph.expectation_value_and_gradient_functional()
+    op_memory.open()
+
+    def setup():
+        return (random_problem.parameters,), {}
+
     _value, gradient = benchmark.pedantic(
         barriered(functional, bench_comm),
-        args=(random_problem.parameters,),
+        setup=barrier_setup(bench_comm, setup),
         rounds=bench_rounds,
         iterations=1,
     )
+    op_memory.close(built_graph)
     assert len(gradient) == len(random_problem.parameters)
 
 
@@ -90,6 +154,9 @@ def run(built):
         return propagator.expectation_value()
 
     result = benchmark.pedantic(
-        barriered(run, bench_comm), setup=setup, rounds=bench_rounds, iterations=1
+        barriered(run, bench_comm),
+        setup=barrier_setup(bench_comm, setup),
+        rounds=bench_rounds,
+        iterations=1,
     )
     assert isinstance(result, float)
diff --git a/benches/conftest.py b/benches/conftest.py
index b469ca41..ef22dd80 100644
--- a/benches/conftest.py
+++ b/benches/conftest.py
@@ -32,6 +32,8 @@
 
 from __future__ import annotations
 
+import gc
+import hashlib
 import json
 import os
 import socket
@@ -43,8 +45,7 @@
 import pytest
 from monoprop_bench_tools.memory.cpu import (
     HighWaterMark,
-    PssSampler,
-    merge_peak_of_sum,
+    pinned_thread_summary,
     resting_rss_bytes,
 )
 from monoprop_bench_tools.models import (
@@ -88,6 +89,33 @@ def _reduce_sum(comm: Any, value: int) -> int:
     return value
 
 
+def _reduce_max(comm: Any, value: int) -> int:
+    """Return the largest ``value`` over ranks. Collective; serial returns ``value``."""
+    if comm is not None and comm.Get_size() > 1:
+        return comm.allreduce(value, op=MPI.MAX)
+    return value
+
+
+def _reduce_min(comm: Any, value: int) -> int:
+    """Return the smallest ``value`` over ranks. Collective; serial returns ``value``."""
+    if comm is not None and comm.Get_size() > 1:
+        return comm.allreduce(value, op=MPI.MIN)
+    return value
+
+
+def _gather_lists(comm: Any, values: list[int]) -> list[list[int]]:
+    """Gather per-rank CPU-id lists to rank 0. Collective; off root returns ``[]``."""
+    if comm is None or comm.Get_size() == 1:
+        return [values]
+    gathered = comm.gather(values, root=0)
+    return gathered if gathered is not None else []
+
+
+def _spread(comm: Any, value: int) -> dict[str, int]:
+    """Reduce a per-rank number to ``sum`` (bounds the job) and ``max`` (bounds a node)."""
+    return {"sum": _reduce_sum(comm, value), "max": _reduce_max(comm, value)}
+
+
 _RANDOM_OPTIONS = (
     ("gen-length", 4, "Majorana operators per generator."),
     ("obs-terms", 10000, "Observable terms."),
@@ -98,17 +126,21 @@ def _reduce_sum(comm: Any, value: int) -> int:
     ("bench-rounds", 1, "Fixed timing rounds (MPI-safe)."),
 )
 
-# Non-timing results, written at session end (rank 0) to ``results/<label>.json``.
 _RESULTS: dict[str, Any] = {
     "meta": {},  # run configuration (ranks, threads, host, ...)
     "params": {},  # resolved random-problem hyperparameters
-    "mem": {},  # node id -> peak-of-sum PSS bytes (per operation; MPI lower bound)
-    "memhwm": {},  # node id -> summed per-rank peak RSS bytes (per operation; may fall back to exit RSS if VmHWM can't be reset)  # noqa: E501
-    "opsize": {},  # picture / model -> {"terms": n}
+    "memhwm": {},  # node id -> summed peak RSS, whole test, setup() included
+    "opsize": {},  # picture / model / node id -> {"terms": n}
     "memrest": {},  # picture / model -> resting RSS bytes
     "membase": {},  # fixed model -> resting RSS bytes before the model is built
     "configs": {},  # fixed model -> config dataclass fields
     "opmem": {},  # fixed model -> per-field operator memory split (bytes)
+    # Timed call only (see ``OpMemory``), each {"sum", "max"}.
+    "opmemdelta": {},  # node id -> peak above the operation's floor
+    "opmempeak": {},  # node id -> peak including resident bytes
+    "opmembase": {},  # node id -> that floor
+    "opbytes": {},  # node id -> {"operator": n, "graph": n}
+    "opmembreak": {},  # node id -> operator memory split (bytes)
 }
 
 
@@ -127,20 +159,6 @@ def _results_path() -> Path | None:
     return Path(results, f"{label}.json")
 
 
-def _peak_of_sum(comm: Any, samples: list[tuple[float, int]]) -> int:
-    """Reduce per-rank PSS timelines to the job's peak summed footprint (bytes).
-
-    Gathers every rank's ``(wall_clock, pss)`` samples to rank 0 and merges them
-    via :func:`_memory.merge_peak_of_sum`. Collective; returns ``0`` off root.
-    """
-    if comm is None or comm.Get_size() == 1:
-        return merge_peak_of_sum([samples])
-    gathered = comm.gather(samples, root=0)
-    if comm.Get_rank() != 0:
-        return 0
-    return merge_peak_of_sum(gathered)
-
-
 def pytest_addoption(parser: pytest.Parser) -> None:
     """Register benchmark configuration options for the random benchmarks."""
     group = parser.getgroup("monoprop-bench", "monoprop random benchmark sizing")
@@ -158,6 +176,19 @@ def pytest_addoption(parser: pytest.Parser) -> None:
             )
 
 
+def _core_md5() -> str:
+    """Return the md5 of the extension module this run imported.
+
+    The install-time version stamp is not rewritten by a later ``cmake --build`` and
+    site-packages holds its own copies, so hash what ``_core`` actually resolved to.
+    """
+    try:
+        path = Path(monoprop._core.__file__)
+        return hashlib.md5(path.read_bytes()).hexdigest()  # noqa: S324
+    except (AttributeError, OSError, TypeError):
+        return "unavailable"
+
+
 def _meta() -> dict[str, Any]:
     """Return this run's configuration metadata for the report."""
     return {
@@ -168,8 +199,14 @@ def _meta() -> dict[str, Any]:
         "cpu_count_physical": psutil.cpu_count(logical=False),
         "hostname": socket.gethostname(),
         "monoprop_version": monoprop.__version__,
+        "monoprop_core_md5": _core_md5(),
         "monoprop_variant": monoprop.__variant__,
         "monoprop_compiler_flags": monoprop.__compiler_flags__,
+        "monoprop_max_num_modes": monoprop.MAX_NUM_MODES,
+        "malloc_arena_max": os.environ.get("MALLOC_ARENA_MAX", "default"),
+        "omp_num_threads": os.environ.get("OMP_NUM_THREADS", "default"),
+        # Filled by _record_placement: the threads exist only once a propagator does.
+        "pinning": {},
     }
 
 
@@ -185,11 +222,7 @@ def _params(config: pytest.Config) -> dict[str, Any]:
 def pytest_configure(config: pytest.Config) -> None:
     """Record run metadata on rank 0; silence the other ranks under MPI.
 
-    Every rank runs the whole session, so without intervention they interleave
-    output and race on the shared ``--benchmark-json`` file. Rank 0 prints, writes
-    the JSON, and records metadata; the others go silent. The memory fixture still
-    runs everywhere (for the collective reduce), but only rank 0 records.
-
+    Every rank runs the whole session, so rank 0 alone prints, writes and records.
     ``trylast`` so the terminal reporter exists before non-root ranks unregister it.
     """
     if _rank() == 0:
@@ -200,9 +233,8 @@ def pytest_configure(config: pytest.Config) -> None:
     reporter = config.pluginmanager.getplugin("terminalreporter")
     if reporter is not None:
         config.pluginmanager.unregister(reporter)
-    # pytest-benchmark opens --benchmark-json at parse time and writes it at
-    # session finish. Null the session handle to skip the write, then close the
-    # file so it does not leak (closing alone leaves ``with self.json`` raising).
+    # Nulling the session handle skips the write; closing alone leaves ``with self.json``
+    # raising, so do both.
     bench_session = getattr(config, "_benchmarksession", None)
     if bench_session is not None:
         bench_session.json = None
@@ -253,6 +285,30 @@ def model_configs(request: pytest.FixtureRequest) -> dict[str, Any]:
     }
 
 
+def _record_placement(comm: Any) -> None:
+    """Record engine thread placement over ranks; every rank must enter the reductions and gather.
+
+    min/max because a partial failure -- some ranks placed, some not -- is the shape a cpuset
+    confinement takes and is invisible in rank 0's view alone.
+    """
+    summary = pinned_thread_summary()
+    placed = summary["single_cpu_threads"]
+    mask = summary["affinity_cpus"]
+    spread = {
+        "single_cpu_threads_min": _reduce_min(comm, placed),
+        "single_cpu_threads_max": _reduce_max(comm, placed),
+        "affinity_cpus_min": _reduce_min(comm, mask),
+        "affinity_cpus_max": _reduce_max(comm, mask),
+    }
+    pinned_cpus_by_rank = _gather_lists(comm, summary["pinned_cpus"])
+    if _rank() == 0:
+        _RESULTS["meta"]["pinning"] = {
+            **summary,
+            **spread,
+            "pinned_cpus_by_rank": pinned_cpus_by_rank,
+        }
+
+
 @pytest.fixture
 def record_model_config() -> Callable[[str, Any], None]:
     """Return ``record(model, config)`` recording a model's resolved config."""
@@ -263,61 +319,135 @@ def _do(model: str, config: Any) -> None:
     return _do
 
 
+def _record_model_stats(
+    comm: Any, key: str, propagator: Any, baseline_rss: int | None = None
+) -> None:
+    """Record term count, operator memory breakdown and footprint under ``key``."""
+    _record("opsize", key, {"terms": _reduce_sum(comm, propagator.size())})
+
+    # Placement is only observable while the propagator's threads are alive.
+    _record_placement(comm)
+
+    # The Python front-end does not re-export the C++ accounting.
+    breakdown = getattr(propagator._simulator, "operator_memory_breakdown", None)
+    if breakdown is not None:
+        _record(
+            "opmem",
+            key,
+            {k: _reduce_sum(comm, v) for k, v in breakdown().items()},
+        )
+
+    resting = _reduce_sum(comm, resting_rss_bytes())
+    if resting:  # 0 => /proc unavailable; skip rather than record 0 MiB
+        _record("memrest", key, resting)
+
+    if baseline_rss is not None:
+        baseline = _reduce_sum(comm, baseline_rss)
+        if baseline:
+            _record("membase", key, baseline)
+
+
 @pytest.fixture
 def record_model_stats(bench_comm: Any) -> Callable[..., None]:
     """Return ``record(model, propagator, baseline_rss)`` for fixed-model runs."""
 
     def _do(model: str, propagator: Any, baseline_rss: int) -> None:
-        _record("opsize", model, {"terms": _reduce_sum(bench_comm, propagator.size())})
+        _record_model_stats(bench_comm, model, propagator, baseline_rss)
 
-        # Sparse InvertedIndex columns cost a TermIndex (4B) per set bit against ~1-2B per
-        # set bit in the rows, so which of the two dominates is what sizing decisions turn on.
-        # The Python front-end does not re-export the C++ accounting; it hangs off ._simulator.
-        breakdown = getattr(propagator._simulator, "operator_memory_breakdown", None)
-        # None => binding predates operator_memory_breakdown()
+    return _do
+
+
+class OpMemory:
+    """Records one benchmarked operation's memory over the timed call alone.
+
+    The window opens in ``setup`` and closes when ``pedantic`` returns, so ``delta`` is the
+    operation's own cost, not the construction transient. ``settle=False`` deliberately:
+    settling is seconds of ``gc.collect()``/``malloc_trim`` beside a timed region.
+    """
+
+    def __init__(self, key: str, comm: Any) -> None:
+        """Bind the recorder to one benchmark's report key and communicator."""
+        self._key = key
+        self._comm = comm
+        self._window: HighWaterMark | None = None
+
+    def open(self) -> None:
+        """Start the window. Safe to call once per round; the last round wins."""
+        self._window = HighWaterMark(settle=False)
+        self._window.start()
+
+    def close(self, propagator: Any = None) -> None:
+        """Record the window and ``propagator``'s byte counts; collective, so every rank calls it equally often."""
+        if self._window is None:
+            return
+        self._window.stop()
+        window, self._window = self._window, None
+
+        _record("opmemdelta", self._key, _spread(self._comm, window.delta_bytes))
+        _record("opmempeak", self._key, _spread(self._comm, window.peak_bytes))
+        _record("opmembase", self._key, _spread(self._comm, window.baseline_bytes))
+
+        _record_placement(self._comm)
+
+        if propagator is None:
+            return
+        simulator = getattr(propagator, "_simulator", None)
+        counts = {
+            name: getattr(simulator, f"{name}_memory_bytes", None)
+            for name in ("operator", "graph")
+        }
+        if all(fn is not None for fn in counts.values()):
+            _record(
+                "opbytes",
+                self._key,
+                {
+                    name: _reduce_sum(self._comm, int(fn()))
+                    for name, fn in counts.items()
+                },
+            )
+        breakdown = getattr(simulator, "operator_memory_breakdown", None)
         if breakdown is not None:
             _record(
-                "opmem",
-                model,
-                {k: _reduce_sum(bench_comm, v) for k, v in breakdown().items()},
+                "opmembreak",
+                self._key,
+                {k: _reduce_sum(self._comm, v) for k, v in breakdown().items()},
             )
 
-        resting = _reduce_sum(bench_comm, resting_rss_bytes())
-        if resting:  # 0 => /proc unavailable; skip rather than record 0 MiB
-            _record("memrest", model, resting)
 
-        baseline = _reduce_sum(bench_comm, baseline_rss)
-        if baseline:
-            _record("membase", model, baseline)
+@pytest.fixture
+def op_memory(request: pytest.FixtureRequest, bench_comm: Any) -> Iterator[OpMemory]:
+    """Return this test's :class:`OpMemory` recorder, keyed as the timing report keys it."""
+    recorder = OpMemory(request.node.nodeid.split("/")[-1], bench_comm)
+    yield recorder
+    # A test that raised before closing would desync the collectives.
+    recorder.close()
+
+
+@pytest.fixture
+def record_opsize(
+    request: pytest.FixtureRequest, bench_comm: Any
+) -> Callable[[Any], int]:
+    """Return ``record(propagator)`` storing and returning the global term count."""
+    key = request.node.nodeid.split("/")[-1]
+
+    def _do(propagator: Any) -> int:
+        terms = _reduce_sum(bench_comm, propagator.size())
+        _record("opsize", key, {"terms": terms})
+        return terms
 
     return _do
 
 
 @pytest.fixture(autouse=True)
 def record_memory(request: pytest.FixtureRequest, bench_comm: Any) -> Iterator[None]:
-    """Record each benchmark's peak physical-memory footprint for the report.
-
-    Two numbers, because neither alone is both exact and MPI-aware:
-
-    ``memhwm``
-        The kernel's exact peak RSS (:class:`HighWaterMark`) summed over ranks. Exact per
-        rank, and an upper bound on the job total, since ranks that peak at different
-        moments are added as though they had peaked together.
-    ``mem``
-        The peak-of-sum of the sampled per-rank PSS timelines, which is the quantity that
-        actually bounds a node's RAM. A lower bound: the sampler only advances when a rank
-        drops the GIL, and monoprop's ``propagate()`` holds it for the whole call.
-
-    Both are footprints, not deltas: they include structures already resident when the
-    operation starts (e.g. the shared :func:`built_graph`). The gather is collective, but
-    only rank 0 records.
+    """Record ``memhwm``: peak RSS over the whole test, summed over ranks.
+
+    It spans ``setup``, so it predicts an OOM kill but is the wrong number for comparing
+    operations -- ``opmemdelta`` is that. The reduce is collective; only rank 0 records.
     """
-    with HighWaterMark() as window, PssSampler() as sampler:
+    with HighWaterMark() as window:
         yield
     key = request.node.nodeid.split("/")[-1]
-    mem = _peak_of_sum(bench_comm, sampler.samples)
-    if mem:  # 0 => non-root rank or /proc unavailable: nothing to record
-        _record("mem", key, mem)
     hwm = _reduce_sum(bench_comm, window.peak_bytes)
     if hwm:
         _record("memhwm", key, hwm)
@@ -336,7 +466,7 @@ def picture(request: pytest.FixtureRequest) -> str:
 def random_problem(request: pytest.FixtureRequest) -> RandomProblem:
     """Build the random observable/circuit problem from the CLI options."""
     opt = request.config.getoption
-    return make_random_problem(
+    problem = make_random_problem(
         gen_length=opt("--gen-length"),
         obs_terms=opt("--obs-terms"),
         num_generators=opt("--num-generators"),
@@ -344,6 +474,9 @@ def random_problem(request: pytest.FixtureRequest) -> RandomProblem:
         cutoff=opt("--cutoff"),
         seed=opt("--seed"),
     )
+    # Keeps every later HighWaterMark's gc.collect() off these session-lifetime tuples.
+    gc.freeze()
+    return problem
 
 
 @pytest.fixture
@@ -394,3 +527,36 @@ def built_graph(
         _record("memrest", picture, resting)
 
     return mp
+
+
+@pytest.fixture(scope="session")
+def model_graph(
+    model_configs: dict[str, Any], bench_comm: Any
+) -> Callable[[str], tuple[Any, list[float]]]:
+    """Return ``get(model)`` giving a fixed model's built graph and its parameter vector.
+
+    Hubbard's circuit is one Trotter step re-applied, so the graph takes ``steps`` successive
+    ``build_graph`` calls and the parameter vector is the circuit's repeated to match.
+    """
+    cache: dict[str, tuple[Any, list[float]]] = {}
+
+    def _get(model: str) -> tuple[Any, list[float]]:
+        if model not in cache:
+            _config_cls, build_fn, steps_fn = MODELS[model]
+            config = model_configs[model]
+            steps = steps_fn(config)
+            propagator, circuit = build_fn(config, comm=bench_comm)
+            for _ in range(steps):
+                propagator.build_graph(circuit)
+            parameters = list(circuit.parameters) * steps
+            # A mismatch benchmarks a different graph than the build_graph cell measured.
+            assert len(parameters) == propagator.n_parameters
+
+            # No test here calls record_model_config, so record the config from this side.
+            _record("configs", model, asdict(config))
+            _record_model_stats(bench_comm, model, propagator)
+            gc.freeze()  # same reason as random_problem
+            cache[model] = (propagator, parameters)
+        return cache[model]
+
+    return _get
diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt
index fe5c7c0b..9a34fe1a 100644
--- a/cpp/tests/CMakeLists.txt
+++ b/cpp/tests/CMakeLists.txt
@@ -51,10 +51,34 @@ target_link_libraries(
 
 include(${CMAKE_CURRENT_LIST_DIR}/boost-test.cmake)
 
+# CTest launches each Boost case as a world-size-1 process, so excluding fabric components cuts
+# MPI_Init from 2.03 s to 0.61 s without affecting communication; real MPI tests must keep the
+# full component set.
+#
+# Use SERIAL_ENVIRONMENT with exclusions (^), not a positive component list: Open MPI component
+# names change across versions, and unknown OMPI_MCA_* variables are harmless under other MPIs.
+option(
+  monoprop_TEST_EXCLUDE_MPI_FABRIC
+  "Skip fabric init in the single-process (serial) test variants to cut MPI_Init cost"
+  ON
+)
+set(_monoprop_serial_env_entries)
+if(monoprop_TEST_EXCLUDE_MPI_FABRIC AND monoprop_ENABLE_MPI)
+  list(
+    APPEND _monoprop_serial_env_entries
+    "OMPI_MCA_pml=^ucx"
+    "OMPI_MCA_btl=^openib,ofi,uct"
+  )
+endif()
+
 # Automatic discovery of unit tests.
 # Default CTest run includes per-case serial tests plus suite-level MPI variants
 # for monoprop_MPI_TEST_PROCS.
-discover_tests(monoprop_unit_tests.x
+discover_tests(
+  monoprop_unit_tests.x
   PROPERTIES
-    LABELS "unit"
+  LABELS
+  "unit"
+  SERIAL_ENVIRONMENT
+  ${_monoprop_serial_env_entries}
 )
diff --git a/cpp/tests/boost-test.cmake b/cpp/tests/boost-test.cmake
index 4f686b2b..7d0d52c9 100644
--- a/cpp/tests/boost-test.cmake
+++ b/cpp/tests/boost-test.cmake
@@ -21,12 +21,13 @@ if(NOT _monoprop_mpiexec_numproc_flag)
   set(_monoprop_mpiexec_numproc_flag "-n")
 endif()
 
+# SERIAL_ENVIRONMENT: VAR=value entries applied to the per-case `serial` variants only.
 function(discover_tests TARGET)
   cmake_parse_arguments(
     ""
     ""
     "WORKING_DIRECTORY"
-    "EXTRA_ARGS;PROPERTIES"
+    "EXTRA_ARGS;PROPERTIES;SERIAL_ENVIRONMENT"
     ${ARGN}
   )
 
@@ -66,6 +67,7 @@ function(discover_tests TARGET)
       "TEST_EXECUTABLE=$<TARGET_FILE:${TARGET}>" -D
       "TEST_WORKING_DIR=${_WORKING_DIRECTORY}" -D
       "TEST_EXTRA_ARGS=${_EXTRA_ARGS}" -D "TEST_PROPERTIES=${_PROPERTIES}" -D
+      "TEST_SERIAL_ENVIRONMENT=${_SERIAL_ENVIRONMENT}" -D
       "TEST_LIST=${_TEST_LIST}" -D "CTEST_FILE=${ctest_tests_file}" -D
       "TEST_ENABLE_MPI_VARIANTS=${_enable_mpi_variants}" -D
       "TEST_MPI_NUMPROCS=${monoprop_MPI_TEST_PROCS}" -D
diff --git a/cpp/tests/boostAddTests.cmake b/cpp/tests/boostAddTests.cmake
index f0fd71b2..74b5ff00 100644
--- a/cpp/tests/boostAddTests.cmake
+++ b/cpp/tests/boostAddTests.cmake
@@ -39,6 +39,7 @@ endif()
 
 set(extra_args ${TEST_EXTRA_ARGS})
 set(properties ${TEST_PROPERTIES})
+set(serial_env ${TEST_SERIAL_ENVIRONMENT})
 set(script)
 set(tests)
 
@@ -205,6 +206,8 @@ foreach(LINE ${LINES})
         ${extra_args}
       LABELS
         serial
+      ENVIRONMENT
+        ${serial_env}
     )
   endif()
 endforeach()
diff --git a/cspell.json b/cspell.json
index 4f86409b..b1f6a54a 100644
--- a/cspell.json
+++ b/cspell.json
@@ -17,6 +17,7 @@
     "backend",
     "barriered",
     "bitstring",
+    "btl",
     "coeffs",
     "conj",
     "ctest",
@@ -46,13 +47,19 @@
     "memray",
     "microbenchmarks",
     "mpiexec",
+    "ofi",
+    "openib",
     "oversubscribe",
     "Pauli",
     "Paulis",
+    "pml",
     "qubit",
     "qubits",
     "Remigio",
     "tracemalloc",
+    "uct",
+    "ucx",
+    "vader",
     "schrodinger",
     "simulable",
     "unnormalized",
diff --git a/docs/content/docs/testing.mdx b/docs/content/docs/testing.mdx
index 8390991b..51326cff 100644
--- a/docs/content/docs/testing.mdx
+++ b/docs/content/docs/testing.mdx
@@ -85,6 +85,12 @@ A 64-bit `TermIndex` build is the only configuration that compiles the
 just test-wide   # rebuilds via uv sync with monoprop_WIDE_TERM_INDEX=ON, then runs CTest
 ```
 
+CTest runs each Boost case as its own process, so an MPI build's `MPI_Init` probes every
+fabric device per case, whether or not the test sends anything. `monoprop_TEST_EXCLUDE_MPI_FABRIC=ON`
+(default) skips that probe for the single-process `serial` variants only; multi-rank variants
+keep the full component set, since they exchange real messages. Turn it off by adding
+`-Dmonoprop_TEST_EXCLUDE_MPI_FABRIC=OFF` to the `SKBUILD_CMAKE_ARGS` MPI build above.
+
 CTest registers every Boost case individually as a `serial` variant. When the
 build has MPI enabled and a launcher is found, it also registers the whole suite
 once per rank count in `monoprop_MPI_TEST_PROCS`, labelled `mpi` and `mpi-<n>` —
diff --git a/packages/monoprop-bench-tools/src/monoprop_bench_tools/bmf.py b/packages/monoprop-bench-tools/src/monoprop_bench_tools/bmf.py
index abd08429..cd8a64ca 100644
--- a/packages/monoprop-bench-tools/src/monoprop_bench_tools/bmf.py
+++ b/packages/monoprop-bench-tools/src/monoprop_bench_tools/bmf.py
@@ -57,6 +57,10 @@
 # describe a built operator rather than one timed call.
 _OPERATOR = "operator"
 
+# ``opsize``/``memrest`` are also keyed by pytest node id (``::``, as report.py reads
+# them) for a private A/B harness; those are not operators, so they are skipped here.
+_NODE_ID_SEP = "::"
+
 Metric = dict[str, float]
 Bmf = dict[str, dict[str, Metric]]
 
@@ -118,17 +122,17 @@ def measure(benchmark: str, name: str, value: float) -> None:
     for benchmark, latency in _timings(results_dir, label):
         bmf.setdefault(benchmark, {})["latency"] = latency
 
-    # ``memhwm`` is the kernel's exact peak RSS per operation (summed over ranks
-    # under MPI), which is what REPORT.md shows; ``mem`` is the peak-of-sum PSS
-    # lower bound for the same window and would only duplicate the signal.
+    # ``memhwm`` is the kernel's exact peak RSS per operation, summed over ranks under MPI.
     for benchmark, peak in results.get("memhwm", {}).items():
         measure(benchmark, "peak-memory", peak)
 
     for key, size in results.get("opsize", {}).items():
-        measure(f"{_OPERATOR}[{key}]", "terms", size["terms"])
+        if _NODE_ID_SEP not in key:
+            measure(f"{_OPERATOR}[{key}]", "terms", size["terms"])
 
     for key, resting in results.get("memrest", {}).items():
-        measure(f"{_OPERATOR}[{key}]", "resting-memory", resting)
+        if _NODE_ID_SEP not in key:
+            measure(f"{_OPERATOR}[{key}]", "resting-memory", resting)
 
     return bmf
 
diff --git a/packages/monoprop-bench-tools/src/monoprop_bench_tools/memory/cpu.py b/packages/monoprop-bench-tools/src/monoprop_bench_tools/memory/cpu.py
index 055954c0..7405e2ab 100644
--- a/packages/monoprop-bench-tools/src/monoprop_bench_tools/memory/cpu.py
+++ b/packages/monoprop-bench-tools/src/monoprop_bench_tools/memory/cpu.py
@@ -34,6 +34,7 @@
 
 import contextlib
 import gc
+import os
 import threading
 import time
 from pathlib import Path
@@ -124,6 +125,55 @@ def resting_rss_bytes() -> int:
     return rss_bytes()
 
 
+def _parse_cpu_list(spec: str) -> set[int]:
+    """Expand a kernel CPU list (``0-3,8``) into a set of CPU numbers."""
+    cpus: set[int] = set()
+    for part in spec.split(","):
+        if not part:
+            continue
+        lo, _, hi = part.partition("-")
+        cpus.update(range(int(lo), int(hi or lo) + 1))
+    return cpus
+
+
+def pinned_thread_summary() -> dict[str, int | list[int]]:
+    """Count this rank's own threads bound to a single CPU; all-zero means ``/proc`` was unreadable, not unpinned."""
+    try:
+        tasks = list(Path("/proc/self/task").iterdir())
+    except OSError:  # pragma: no cover - non-Linux or restricted /proc
+        return {
+            "threads": 0,
+            "single_cpu_threads": 0,
+            "distinct_pinned_cpus": 0,
+            "affinity_cpus": 0,
+            "pinned_cpus": [],
+        }
+
+    threads = 0
+    pinned: set[int] = set()
+    single = 0
+    for task in tasks:
+        try:
+            text = (task / "status").read_text()
+        except OSError:  # thread exited between listing and reading
+            continue
+        threads += 1
+        for line in text.splitlines():
+            if line.startswith("Cpus_allowed_list:"):
+                cpus = _parse_cpu_list(line.split(":", 1)[1].strip())
+                if len(cpus) == 1:
+                    single += 1
+                    pinned |= cpus
+                break
+    return {
+        "threads": threads,
+        "single_cpu_threads": single,
+        "distinct_pinned_cpus": len(pinned),
+        "affinity_cpus": len(os.sched_getaffinity(0)),
+        "pinned_cpus": sorted(pinned),
+    }
+
+
 class HighWaterMark:
     """Exact peak RSS over the enclosed block, straight from the kernel.
 
@@ -173,6 +223,14 @@ def __exit__(
         observed = peak_rss_bytes() if self.exact else rss_bytes()
         self.peak_bytes = max(self.baseline_bytes, observed)
 
+    def start(self) -> Self:
+        """Open the window explicitly; a ``pedantic`` run cannot be wrapped in a ``with``."""
+        return self.__enter__()
+
+    def stop(self) -> None:
+        """Close the window explicitly (same as ``__exit__``)."""
+        self.__exit__(None, None, None)
+
     @property
     def delta_bytes(self) -> int:
         """Return the peak measured above the block's own starting floor."""
diff --git a/packages/monoprop-bench-tools/src/monoprop_bench_tools/models.py b/packages/monoprop-bench-tools/src/monoprop_bench_tools/models.py
index f8bb86f1..1c9ae567 100644
--- a/packages/monoprop-bench-tools/src/monoprop_bench_tools/models.py
+++ b/packages/monoprop-bench-tools/src/monoprop_bench_tools/models.py
@@ -50,10 +50,9 @@
 
 
 def barriered(fn: Callable[..., _T], comm: Any | None) -> Callable[..., _T]:
-    """Wrap ``fn`` so all MPI ranks enter and leave the call together.
+    """Wrap ``fn`` in the timed *exit* barrier, making each measurement the makespan.
 
-    The barriers make each rank's measured time reflect the makespan (the slowest
-    rank). A serial run returns ``fn`` unchanged, so there is no overhead.
+    The entry barrier lives untimed in :func:`barrier_setup`. Serial runs return ``fn``.
 
     Args:
         fn: The callable to wrap.
@@ -63,7 +62,6 @@ def barriered(fn: Callable[..., _T], comm: Any | None) -> Callable[..., _T]:
         return fn
 
     def wrapped(*args: object, **kwargs: object) -> _T:
-        comm.Barrier()
         result = fn(*args, **kwargs)
         comm.Barrier()
         return result
@@ -71,6 +69,21 @@ def wrapped(*args: object, **kwargs: object) -> _T:
     return wrapped
 
 
+def barrier_setup(
+    comm: Any | None, setup: Callable[[], _T] | None = None
+) -> Callable[[], _T] | None:
+    """Untimed *entry* barrier for ``pedantic(setup=...)``; serial runs return ``setup``, ``None`` included."""
+    if comm is None or comm.Get_size() == 1:
+        return setup
+
+    def wrapped_setup() -> _T:
+        result = setup() if setup is not None else None
+        comm.Barrier()
+        return result  # type: ignore[return-value]
+
+    return wrapped_setup
+
+
 @dataclass(frozen=True, slots=True)
 class RandomProblem:
     """A randomly generated observable and circuit ready for a propagator.
@@ -126,6 +139,21 @@ def _random_terms(
     ]
 
 
+def _random_majorana_operator(
+    terms: list[tuple[int, ...]],
+    coefficients: list[float],
+    num_modes: int,
+) -> MajoranaOperator:
+    """Build a :class:`MajoranaOperator` from already-canonical :func:`_random_terms` output.
+
+    Bypasses ``__init__``, whose sign re-derivation duplicates the whole term mapping.
+    """
+    operator = MajoranaOperator.__new__(MajoranaOperator)
+    operator.num_modes = num_modes
+    operator.terms = dict(zip(terms, coefficients, strict=True))
+    return operator
+
+
 def make_random_problem(
     *,
     gen_length: int = 4,
@@ -154,7 +182,7 @@ def make_random_problem(
 
     obs_majoranas = _random_terms(rng, obs_terms, gen_length, num_majorana_indices)
     obs_coeffs = rng.standard_normal(obs_terms).tolist()  # Hermitian -> real
-    observable = MajoranaOperator(dict(zip(obs_majoranas, obs_coeffs)), num_modes)
+    observable = _random_majorana_operator(obs_majoranas, obs_coeffs, num_modes)
 
     gen_majoranas = _random_terms(rng, num_generators, gen_length, num_majorana_indices)
     gen_coeffs = rng.standard_normal(num_generators).tolist()
diff --git a/packages/monoprop-bench-tools/tests/test_bmf.py b/packages/monoprop-bench-tools/tests/test_bmf.py
index 72b594b8..9478d7cf 100644
--- a/packages/monoprop-bench-tools/tests/test_bmf.py
+++ b/packages/monoprop-bench-tools/tests/test_bmf.py
@@ -97,6 +97,29 @@ def test_operator_metrics_are_grouped_per_picture_and_model(tmp_path: Path) -> N
     assert result["operator[hubbard]"] == {"terms": {"value": 12.0}}
 
 
+def test_operator_metrics_skip_node_id_keyed_entries(tmp_path: Path) -> None:
+    build_graph = "bench_random.py::test_random_build_graph[heisenberg]"
+    _write(
+        tmp_path,
+        opsize={"heisenberg": {"terms": 12}, build_graph: {"terms": 34}},
+        memrest={"heisenberg": 4096, build_graph: 8192},
+    )
+    result = bmf.build_bmf(tmp_path, "ci")
+
+    # record_opsize keys by node id for a private A/B harness; Bencher's history keys
+    # on the name forever, so those entries must not reach it under any name at all.
+    assert set(result) == {_ENERGY, "operator[heisenberg]"}
+    assert result == {
+        _ENERGY: {
+            "latency": {"value": 0.5e9, "lower_value": 0.49e9, "upper_value": 0.51e9}
+        },
+        "operator[heisenberg]": {
+            "terms": {"value": 12.0},
+            "resting-memory": {"value": 4096.0},
+        },
+    }
+
+
 def test_missing_sections_are_skipped(tmp_path: Path) -> None:
     _write(tmp_path)