diff --git a/bin/ethlambda/src/benchmark/import/mod.rs b/bin/ethlambda/src/benchmark/import/mod.rs
index f8344b803..3b5d445b5 100644
--- a/bin/ethlambda/src/benchmark/import/mod.rs
+++ b/bin/ethlambda/src/benchmark/import/mod.rs
@@ -9,6 +9,7 @@ use eyre::WrapErr as _;
 
 pub(crate) mod corpus;
 pub(crate) mod fetch;
+pub(crate) mod process;
 pub(crate) mod replay;
 pub(crate) mod source;
 
diff --git a/bin/ethlambda/src/benchmark/import/process.rs b/bin/ethlambda/src/benchmark/import/process.rs
new file mode 100644
index 000000000..3a0f7a717
--- /dev/null
+++ b/bin/ethlambda/src/benchmark/import/process.rs
@@ -0,0 +1,117 @@
+//! Process-level resource readings for the import replay: CPU time and memory.
+//!
+//! Wall time says how long an import took; it does not say whether the time
+//! went to computation or to waiting, nor whether it ran on one core or
+//! several. CPU seconds beside it does: an import whose CPU time is well above
+//! its wall time was parallel (rayon in the merkle fold and the BLS batches),
+//! and one whose CPU time is well below it spent the difference blocked.
+//!
+//! Everything here reads the whole process, not the importing thread. That is
+//! deliberate: the work an import causes is not all on its own thread (the
+//! state writer, the merkle fold's worker pool, RocksDB's background
+//! compaction), and a per-thread reading would call a parallel import cheap.
+//! The same property is the caveat: anything else the process does during the
+//! import is in the number, which is why the replay runs nothing else.
+//!
+//! No dependency beyond `libc`, which the binary already carries for its file
+//! descriptor limit. `getrusage` is the one portable process-wide CPU clock
+//! the standard library does not wrap, so it is the one `unsafe` call here.
+
+/// User plus system CPU seconds this process has used so far, across all of
+/// its threads, or `None` if the kernel refuses the reading.
+pub(crate) fn cpu_seconds() -> Option<f64> {
+    let usage = rusage()?;
+    Some(seconds(usage.ru_utime) + seconds(usage.ru_stime))
+}
+
+/// The process's peak resident set size so far, in bytes.
+///
+/// A high-water mark, so it never goes down: between two blocks it moves only
+/// when a block set a new peak, and a replay's last value is the run's peak.
+pub(crate) fn peak_rss_bytes() -> Option<u64> {
+    let usage = rusage()?;
+    let maxrss = u64::try_from(usage.ru_maxrss).ok()?;
+    // `ru_maxrss` is bytes on macOS and kibibytes on Linux.
+    if cfg!(target_os = "macos") {
+        Some(maxrss)
+    } else {
+        maxrss.checked_mul(1024)
+    }
+}
+
+/// The process's resident set size right now, in bytes.
+///
+/// Read from `/proc/self/status`, so Linux only: macOS has no equivalent that
+/// needs no new dependency, and `None` there is honest where a stand-in would
+/// not be. A replay on a Mac still reports [`peak_rss_bytes`].
+pub(crate) fn rss_bytes() -> Option<u64> {
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    parse_vm_rss(&status)
+}
+
+/// The `VmRSS` line of a `/proc/<pid>/status` file, which the kernel reports in
+/// kibibytes.
+fn parse_vm_rss(status: &str) -> Option<u64> {
+    let line = status
+        .lines()
+        .find_map(|line| line.strip_prefix("VmRSS:"))?;
+    let kibibytes: u64 = line.trim().strip_suffix("kB")?.trim().parse().ok()?;
+    kibibytes.checked_mul(1024)
+}
+
+fn seconds(time: libc::timeval) -> f64 {
+    time.tv_sec as f64 + time.tv_usec as f64 / 1e6
+}
+
+fn rusage() -> Option<libc::rusage> {
+    let mut usage = std::mem::MaybeUninit::<libc::rusage>::uninit();
+    // SAFETY: `getrusage` writes a complete `rusage` through the pointer when
+    // it returns 0 and does not read it, and the pointer is to a live local;
+    // the value is only assumed initialized on that success path.
+    unsafe {
+        if libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) != 0 {
+            return None;
+        }
+        Some(usage.assume_init())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cpu_time_advances_while_the_process_computes() {
+        let before = cpu_seconds().expect("getrusage works on the host");
+        // Enough work that the clock's resolution cannot hide it; `black_box`
+        // keeps the loop from being folded away.
+        let mut acc = 0u64;
+        let started = std::time::Instant::now();
+        while started.elapsed() < std::time::Duration::from_millis(30) {
+            for i in 0..10_000u64 {
+                acc = std::hint::black_box(acc.wrapping_add(i * i));
+            }
+        }
+        let after = cpu_seconds().expect("getrusage works on the host");
+        assert!(after > before, "{after} should exceed {before}");
+    }
+
+    #[test]
+    fn a_process_has_a_nonzero_peak_resident_set() {
+        assert!(peak_rss_bytes().expect("getrusage works on the host") > 0);
+    }
+
+    #[test]
+    fn vm_rss_is_read_in_bytes_from_the_kibibyte_line() {
+        let status = "Name:\tethlambda\nVmHWM:\t  999 kB\nVmRSS:\t  2048 kB\nThreads:\t4\n";
+        assert_eq!(parse_vm_rss(status), Some(2048 * 1024));
+    }
+
+    #[test]
+    fn a_status_file_without_vm_rss_gives_none() {
+        // A kernel thread's status has no `VmRSS`, and a malformed line must
+        // not become a number.
+        assert_eq!(parse_vm_rss("Name:\tkthreadd\n"), None);
+        assert_eq!(parse_vm_rss("VmRSS:\tlots\n"), None);
+    }
+}
diff --git a/bin/ethlambda/src/benchmark/import/replay.rs b/bin/ethlambda/src/benchmark/import/replay.rs
index f0365571f..6d1cc497c 100644
--- a/bin/ethlambda/src/benchmark/import/replay.rs
+++ b/bin/ethlambda/src/benchmark/import/replay.rs
@@ -16,6 +16,7 @@
 //! only thing allowed to hold post-states, and a stray clone here would pin
 //! an entry the LRU believes it evicted, raising the real ceiling silently.
 
+use std::collections::BTreeMap;
 use std::path::Path;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -24,6 +25,7 @@ use ethlambda_blockchain::metrics::{BLOCK_ARRIVAL_PHASES, BLOCK_IMPORT_PHASES};
 use ethlambda_blockchain::{BlockChainServer, ImportOutcome};
 use ethlambda_p2p::beacon::decode;
 use ethlambda_state_transition::beacon::fork_choice;
+use ethlambda_state_transition::metrics::{BEACON_STF_PHASES, STF_CONTEXT_IMPORT};
 use ethlambda_storage::Store;
 use ethlambda_storage::backend::RocksDBBackend;
 use ethlambda_types::ShortRoot;
@@ -34,15 +36,101 @@ use ethlambda_types::primitives::H256;
 
 use super::ReplayOptions;
 use super::corpus::{ANCHOR_BLOCK_FILE, ANCHOR_STATE_FILE, Manifest, block_path, hex_root};
-use crate::benchmark::PhaseTimer;
+use super::process;
 use crate::benchmark::report::common::{Environment, format_ms};
-use crate::benchmark::report::import::{Params, Report, Sample};
+use crate::benchmark::report::import::{BlockClass, Caches, LookupCounts, Params, Report, Sample};
+use crate::benchmark::{PhaseTimer, read_counter};
 
 /// The histogram `ImportTimings` writes to, read back by [`PhaseTimer`] the
 /// same way the synthetic workload reads
 /// `lean_block_proposal_attestation_build_phase_seconds`.
 const IMPORT_PHASE_HISTOGRAM: &str = "lean_block_import_phase_seconds";
 
+/// The beacon state transition's own step histogram, read back over its
+/// `import` context only.
+///
+/// Reading every series would charge the epoch precompute worker's
+/// `process_epoch` to whichever block was importing when it finished. A replay
+/// runs no such worker, but the filter is what makes that a property of the
+/// reading rather than of how this binary happens to be wired.
+const STF_PHASE_HISTOGRAM: &str = "lean_beacon_stf_phase_seconds";
+
+/// The counters of the caches the state transition leans on, each labeled by
+/// `result`. Read around the measured blocks, so the report's hit rates cover
+/// the run and not the warm-up.
+const EPOCH_PRECOMPUTE_COUNTER: &str = "lean_beacon_epoch_precompute_lookups_total";
+const COMMITTEE_CACHE_COUNTER: &str = "lean_beacon_committee_cache_lookups_total";
+const TOTAL_ACTIVE_BALANCE_COUNTER: &str = "lean_beacon_total_active_balance_lookups_total";
+const JUSTIFIED_BALANCES_COUNTER: &str = "lean_beacon_justified_balances_lookups_total";
+
+/// The four cache counters' per-`result` values at one instant.
+struct CacheReading {
+    epoch_precompute: BTreeMap<String, u64>,
+    committee_cache: BTreeMap<String, u64>,
+    total_active_balance: BTreeMap<String, u64>,
+    justified_balances: BTreeMap<String, u64>,
+}
+
+impl CacheReading {
+    fn now() -> Self {
+        Self {
+            epoch_precompute: read_counter(EPOCH_PRECOMPUTE_COUNTER),
+            committee_cache: read_counter(COMMITTEE_CACHE_COUNTER),
+            total_active_balance: read_counter(TOTAL_ACTIVE_BALANCE_COUNTER),
+            justified_balances: read_counter(JUSTIFIED_BALANCES_COUNTER),
+        }
+    }
+
+    /// What each cache served between `self` and now.
+    fn deltas_since_start(self) -> Caches {
+        let now = Self::now();
+        Caches {
+            epoch_precompute: LookupCounts::from_results(
+                &delta(&self.epoch_precompute, &now.epoch_precompute),
+                false,
+            ),
+            committee_cache: LookupCounts::from_results(
+                &delta(&self.committee_cache, &now.committee_cache),
+                true,
+            ),
+            total_active_balance: LookupCounts::from_results(
+                &delta(&self.total_active_balance, &now.total_active_balance),
+                true,
+            ),
+            justified_balances: LookupCounts::from_results(
+                &delta(&self.justified_balances, &now.justified_balances),
+                false,
+            ),
+        }
+    }
+}
+
+/// `after - before` per result. Counters only go up, so a result missing from
+/// `before` started at zero.
+fn delta(before: &BTreeMap<String, u64>, after: &BTreeMap<String, u64>) -> BTreeMap<String, u64> {
+    after
+        .iter()
+        .map(|(result, value)| {
+            let start = before.get(result).copied().unwrap_or(0);
+            (result.clone(), value.saturating_sub(start))
+        })
+        .collect()
+}
+
+/// The seconds the `stf` import phase spent outside every STF step, or `None`
+/// when `stf` did not run.
+///
+/// The steps are nested inside `stf`, so their sum cannot exceed it; the clamp
+/// only absorbs the few microseconds two separate clock readings can disagree
+/// by, rather than reporting a negative duration.
+fn stf_unattributed(
+    phases: &BTreeMap<String, f64>,
+    stf_phases: &BTreeMap<String, f64>,
+) -> Option<f64> {
+    let stf = *phases.get("stf")?;
+    Some((stf - stf_phases.values().sum::<f64>()).max(0.0))
+}
+
 /// The phases a sample records: every per-block section, plus the
 /// per-arrival sections that do not enclose the others.
 ///
@@ -124,6 +212,11 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
 
     let mut samples = Vec::with_capacity(manifest.slots.len());
 
+    // The slot of the block each import is applied to. A corpus is linear
+    // (`fetch` refuses a block whose parent is not the previous one), so that
+    // is always the block imported just before, and the first is the anchor.
+    let mut parent_slot = manifest.anchor_slot;
+
     let warmup_total = manifest.warmup_slots.len();
     for (position, &slot) in manifest.warmup_slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
@@ -133,30 +226,58 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
             position + 1,
             format_ms(wall_seconds)
         );
+        parent_slot = slot;
         pause_between_blocks(options.block_delay).await;
     }
 
+    // Read once the warm-up is done, so a cache's warm-up misses (every cache
+    // starts cold) are not charged to the measured blocks' hit rates.
+    let caches_before = CacheReading::now();
+
     let total = manifest.slots.len();
     for (iteration, &slot) in manifest.slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
+        let class = BlockClass::classify(parent_slot, slot, preset::SLOTS_PER_EPOCH);
 
         let timer = PhaseTimer::start(IMPORT_PHASE_HISTOGRAM);
+        let stf_timer =
+            PhaseTimer::start_filtered(STF_PHASE_HISTOGRAM, ("context", STF_CONTEXT_IMPORT));
+        let cpu_before = process::cpu_seconds();
         let wall_seconds = import(&mut server, block, slot, block_root).await?;
+        // The CPU reading and the RSS go first, before the histogram reads,
+        // which gather the whole registry and would otherwise be in the CPU
+        // delta of the block they measure.
+        let cpu_seconds = cpu_before
+            .zip(process::cpu_seconds())
+            .map(|(before, after)| (after - before).max(0.0));
+        let rss_bytes = process::rss_bytes();
+        let peak_rss_bytes = process::peak_rss_bytes();
         let phases = timer.finish_at_most_once(sampled_phases())?;
+        let stf_phases = stf_timer.finish_summed(BEACON_STF_PHASES.iter().copied());
+        let stf_unattributed_seconds = stf_unattributed(&phases, &stf_phases);
 
         eprintln!(
-            "block {}/{total}: slot {slot} imported in {}",
+            "block {}/{total}: slot {slot} ({}) imported in {}",
             iteration + 1,
+            class.label(),
             format_ms(wall_seconds)
         );
         samples.push(Sample {
             iteration: iteration as u64,
             slot,
+            parent_slot,
+            class,
             block_root: hex_root(block_root),
             wall_seconds,
             phases,
+            stf_phases,
+            stf_unattributed_seconds,
+            cpu_seconds,
+            rss_bytes,
+            peak_rss_bytes,
             outcome: "imported",
         });
+        parent_slot = slot;
         // After `finish_at_most_once` and the sample are recorded, so the
         // sleep is in no phase delta and no `wall_seconds`.
         pause_between_blocks(options.block_delay).await;
@@ -166,6 +287,7 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
         Environment::collect(),
         params_from(&manifest, dir, options.block_delay),
         samples,
+        caches_before.deltas_since_start(),
     ))
 }
 
@@ -288,3 +410,55 @@ fn prepare_data_dir(data_dir: &Path, force: bool) -> eyre::Result<()> {
     std::fs::create_dir_all(data_dir)?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn map(entries: &[(&str, u64)]) -> BTreeMap<String, u64> {
+        entries
+            .iter()
+            .map(|(result, value)| ((*result).to_string(), *value))
+            .collect()
+    }
+
+    #[test]
+    fn a_counter_delta_excludes_what_was_counted_before_and_starts_new_results_at_zero() {
+        let before = map(&[("hit", 10), ("miss", 4)]);
+        let after = map(&[("hit", 15), ("miss", 4), ("unkeyable", 2)]);
+
+        let delta = delta(&before, &after);
+
+        assert_eq!(delta, map(&[("hit", 5), ("miss", 0), ("unkeyable", 2)]));
+    }
+
+    #[test]
+    fn the_unattributed_stf_time_is_what_the_steps_leave_of_the_stf_phase() {
+        let phases: BTreeMap<String, f64> = [("stf".to_string(), 1.0)].into_iter().collect();
+        let steps: BTreeMap<String, f64> = [
+            ("process_epoch".to_string(), 0.6),
+            ("state_root".to_string(), 0.3),
+        ]
+        .into_iter()
+        .collect();
+
+        let unattributed = stf_unattributed(&phases, &steps).expect("stf ran");
+        assert!((unattributed - 0.1).abs() < 1e-9);
+    }
+
+    #[test]
+    fn a_block_whose_stf_did_not_run_has_no_unattributed_time() {
+        // Absent, not zero: nothing ran, so nothing was left over.
+        assert_eq!(stf_unattributed(&BTreeMap::new(), &BTreeMap::new()), None);
+    }
+
+    #[test]
+    fn steps_that_overshoot_the_stf_phase_by_clock_skew_clamp_to_zero() {
+        let phases: BTreeMap<String, f64> = [("stf".to_string(), 1.0)].into_iter().collect();
+        let steps: BTreeMap<String, f64> = [("state_root".to_string(), 1.000_001)]
+            .into_iter()
+            .collect();
+
+        assert_eq!(stf_unattributed(&phases, &steps), Some(0.0));
+    }
+}
diff --git a/bin/ethlambda/src/benchmark/mod.rs b/bin/ethlambda/src/benchmark/mod.rs
index 4453bd562..5ee3b6057 100644
--- a/bin/ethlambda/src/benchmark/mod.rs
+++ b/bin/ethlambda/src/benchmark/mod.rs
@@ -327,6 +327,10 @@ struct PhaseTimer {
     /// The histogram this timer reads; the import workload uses a different
     /// one (`lean_block_import_phase_seconds`) than the synthetic build does.
     histogram: &'static str,
+    /// A `(label, value)` the series must carry to be read, for a histogram
+    /// with a label besides `phase` that splits one phase into several series.
+    /// `None` reads every series, as the original two workloads do.
+    only: Option<(&'static str, &'static str)>,
     /// Per-phase (sample_sum, sample_count) before the build.
     before: HashMap<String, (f64, u64)>,
 }
@@ -335,10 +339,47 @@ impl PhaseTimer {
     fn start(histogram: &'static str) -> Self {
         Self {
             histogram,
-            before: read(histogram),
+            only: None,
+            before: read(histogram, None),
         }
     }
 
+    /// [`Self::start`] over only the series that carry `only`'s label value.
+    ///
+    /// For `lean_beacon_stf_phase_seconds`, whose `context` label separates the
+    /// import path's observations from every other caller's: without the filter
+    /// a background `process_epoch` that finished mid-import would be charged
+    /// to the block that happened to be importing.
+    fn start_filtered(histogram: &'static str, only: (&'static str, &'static str)) -> Self {
+        Self {
+            histogram,
+            only: Some(only),
+            before: read(histogram, Some(only)),
+        }
+    }
+
+    /// Per-phase seconds accumulated since [`PhaseTimer::start`], for phases a
+    /// single import may observe any number of times.
+    ///
+    /// The counterpart of [`Self::finish_at_most_once`] for a phase that
+    /// legitimately repeats within one import: `process_epoch` once per epoch
+    /// boundary a long skip crosses, the gloas payload steps twice. The value is
+    /// the sum of those observations, which is the time the import spent in the
+    /// phase. A phase never observed is omitted rather than reported as zero,
+    /// for the reason given there.
+    fn finish_summed(self, expected: impl Iterator<Item = &'static str>) -> BTreeMap<String, f64> {
+        let after = read(self.histogram, self.only);
+        let mut phases = BTreeMap::new();
+        for phase in expected {
+            let (sum_before, count_before) = self.before.get(phase).copied().unwrap_or((0.0, 0));
+            let (sum_after, count_after) = after.get(phase).copied().unwrap_or((0.0, 0));
+            if count_after > count_before {
+                phases.insert(phase.to_string(), sum_after - sum_before);
+            }
+        }
+        phases
+    }
+
     /// Per-phase durations since [`PhaseTimer::start`] for `expected` phases.
     ///
     /// Each phase must have been observed exactly once — one build (and one
@@ -349,7 +390,7 @@ impl PhaseTimer {
         self,
         expected: impl Iterator<Item = &'static str>,
     ) -> eyre::Result<BTreeMap<String, f64>> {
-        let after = read(self.histogram);
+        let after = read(self.histogram, self.only);
         let mut phases = BTreeMap::new();
         for phase in expected {
             let (sum_before, count_before) = self.before.get(phase).copied().unwrap_or((0.0, 0));
@@ -377,7 +418,7 @@ impl PhaseTimer {
         self,
         expected: impl Iterator<Item = &'static str>,
     ) -> eyre::Result<BTreeMap<String, f64>> {
-        let after = read(self.histogram);
+        let after = read(self.histogram, self.only);
         let mut phases = BTreeMap::new();
         for phase in expected {
             let (sum_before, count_before) = self.before.get(phase).copied().unwrap_or((0.0, 0));
@@ -397,11 +438,23 @@ impl PhaseTimer {
 }
 
 /// Current (sample_sum, sample_count) per phase label on `histogram`.
-fn read(histogram: &str) -> HashMap<String, (f64, u64)> {
+///
+/// With `only`, series that do not carry that label value are skipped: the
+/// per-phase map is keyed by `phase` alone, so two series of one phase would
+/// otherwise overwrite each other and the reading would be whichever came last.
+fn read(histogram: &str, only: Option<(&str, &str)>) -> HashMap<String, (f64, u64)> {
     ethlambda_metrics::gather()
         .iter()
         .filter(|family| family.name() == histogram)
         .flat_map(|family| family.get_metric())
+        .filter(|metric| {
+            only.is_none_or(|(name, value)| {
+                metric
+                    .get_label()
+                    .iter()
+                    .any(|label| label.name() == name && label.value() == value)
+            })
+        })
         .filter_map(|metric| {
             let phase = metric
                 .get_label()
@@ -418,10 +471,103 @@ fn read(histogram: &str) -> HashMap<String, (f64, u64)> {
         .collect()
 }
 
+/// Current value of every series of the counter family `counter`, by the
+/// value of its `result` label.
+///
+/// A family nothing has incremented yet is not in the registry at all, so an
+/// absent family and an absent series both read as an empty map; the caller
+/// treats a missing result as zero lookups, which is what it is.
+fn read_counter(counter: &str) -> BTreeMap<String, u64> {
+    ethlambda_metrics::gather()
+        .iter()
+        .filter(|family| family.name() == counter)
+        .flat_map(|family| family.get_metric())
+        .filter_map(|metric| {
+            let result = metric
+                .get_label()
+                .iter()
+                .find(|label| label.name() == "result")?
+                .value()
+                .to_string();
+            Some((result, metric.get_counter().value() as u64))
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod phase_timer_tests {
     use super::*;
 
+    #[test]
+    fn a_filtered_timer_reads_only_the_series_it_names() {
+        // The shape of `lean_beacon_stf_phase_seconds`: one phase, two
+        // callers. A background caller's observation made during the span must
+        // not reach the import's reading.
+        let histogram = ethlambda_metrics::register_histogram_vec!(
+            "test_phase_timer_filtered_seconds",
+            "Phase histogram used only by the PhaseTimer tests",
+            &["context", "phase"]
+        )
+        .expect("registers once");
+
+        let timer =
+            PhaseTimer::start_filtered("test_phase_timer_filtered_seconds", ("context", "import"));
+        histogram
+            .with_label_values(&["import", "process_epoch"])
+            .observe(0.25);
+        histogram
+            .with_label_values(&["other", "process_epoch"])
+            .observe(9.0);
+
+        let phases = timer.finish_summed(["process_epoch", "randao"].into_iter());
+
+        assert!((phases["process_epoch"] - 0.25).abs() < 1e-9);
+        assert!(!phases.contains_key("randao"), "never observed, so absent");
+    }
+
+    #[test]
+    fn a_summed_phase_adds_every_observation_of_one_import() {
+        let histogram = ethlambda_metrics::register_histogram_vec!(
+            "test_phase_timer_summed_seconds",
+            "Phase histogram used only by the PhaseTimer tests",
+            &["context", "phase"]
+        )
+        .expect("registers once");
+        // Observed before the span: in neither the sum nor the count.
+        histogram
+            .with_label_values(&["import", "execution_payload"])
+            .observe(100.0);
+
+        let timer =
+            PhaseTimer::start_filtered("test_phase_timer_summed_seconds", ("context", "import"));
+        histogram
+            .with_label_values(&["import", "execution_payload"])
+            .observe(0.1);
+        histogram
+            .with_label_values(&["import", "execution_payload"])
+            .observe(0.2);
+
+        let phases = timer.finish_summed(["execution_payload"].into_iter());
+        assert!((phases["execution_payload"] - 0.3).abs() < 1e-9);
+    }
+
+    #[test]
+    fn a_counter_is_read_by_its_result_label() {
+        let counter = ethlambda_metrics::register_int_counter_vec!(
+            "test_read_counter_total",
+            "Counter used only by the read_counter test",
+            &["result"]
+        )
+        .expect("registers once");
+        counter.with_label_values(&["hit"]).inc_by(3);
+        counter.with_label_values(&["miss"]).inc();
+
+        let values = read_counter("test_read_counter_total");
+        assert_eq!(values["hit"], 3);
+        assert_eq!(values["miss"], 1);
+        assert!(read_counter("test_never_registered_total").is_empty());
+    }
+
     #[test]
     fn an_unobserved_phase_is_omitted_rather_than_reported_as_zero() {
         // An import runs a subset of BLOCK_IMPORT_PHASES: decode, defer,
diff --git a/bin/ethlambda/src/benchmark/report/common.rs b/bin/ethlambda/src/benchmark/report/common.rs
index a6dd70940..417031cbb 100644
--- a/bin/ethlambda/src/benchmark/report/common.rs
+++ b/bin/ethlambda/src/benchmark/report/common.rs
@@ -47,6 +47,10 @@ pub(crate) struct Stats {
     pub mean_seconds: f64,
     pub p50_seconds: f64,
     pub p90_seconds: f64,
+    /// Nearest-rank, like the others, so a short run reports its maximum for the
+    /// tail: up to 51 samples for the p99 and 11 for the p95.
+    pub p95_seconds: f64,
+    pub p99_seconds: f64,
     pub max_seconds: f64,
     /// Coefficient of variation (stddev / mean); NaN-free (0 when mean is 0).
     pub cv: f64,
@@ -64,6 +68,8 @@ pub(crate) fn stats(values: &[f64]) -> Stats {
             mean_seconds: 0.0,
             p50_seconds: 0.0,
             p90_seconds: 0.0,
+            p95_seconds: 0.0,
+            p99_seconds: 0.0,
             max_seconds: 0.0,
             cv: 0.0,
         };
@@ -88,6 +94,8 @@ pub(crate) fn stats(values: &[f64]) -> Stats {
         mean_seconds: mean,
         p50_seconds: percentile(&sorted, 0.50),
         p90_seconds: percentile(&sorted, 0.90),
+        p95_seconds: percentile(&sorted, 0.95),
+        p99_seconds: percentile(&sorted, 0.99),
         max_seconds: sorted[count - 1],
         cv,
     }
@@ -133,6 +141,29 @@ mod tests {
         assert!((stats.cv - 0.4).abs() < 1e-12);
     }
 
+    #[test]
+    fn the_tail_percentiles_are_nearest_rank_over_the_sorted_samples() {
+        // 1..=200 in a scrambled order: nearest rank of (n - 1) * q, rounded.
+        // p95 = index round(199 * 0.95) = 189 -> 190.0, p99 = index
+        // round(199 * 0.99) = 197 -> 198.0.
+        let mut values: Vec<f64> = (1..=200).map(f64::from).collect();
+        values.reverse();
+        values.swap(3, 150);
+        let stats = stats(&values);
+        assert_eq!(stats.p95_seconds, 190.0);
+        assert_eq!(stats.p99_seconds, 198.0);
+        assert_eq!(stats.max_seconds, 200.0);
+        assert!(stats.p90_seconds <= stats.p95_seconds);
+        assert!(stats.p95_seconds <= stats.p99_seconds);
+    }
+
+    #[test]
+    fn a_short_run_reports_its_maximum_as_the_tail() {
+        let stats = stats(&[0.1, 0.4, 0.2]);
+        assert_eq!(stats.p95_seconds, 0.4);
+        assert_eq!(stats.p99_seconds, 0.4);
+    }
+
     #[test]
     fn stats_on_empty_input_is_zeroed() {
         let stats = stats(&[]);
diff --git a/bin/ethlambda/src/benchmark/report/import.rs b/bin/ethlambda/src/benchmark/report/import.rs
index 74bb20e8e..bad2efc1c 100644
--- a/bin/ethlambda/src/benchmark/report/import.rs
+++ b/bin/ethlambda/src/benchmark/report/import.rs
@@ -7,6 +7,55 @@ use serde::Serialize;
 
 use super::common::{Environment, Stats, format_ms, stats};
 
+/// What a block's `process_slots` had to do, which is the first-order
+/// predictor of how long its state transition takes: crossing an epoch
+/// boundary runs `process_epoch` over the whole registry, and skipping slots
+/// runs the per-slot bookkeeping once per slot.
+///
+/// Decided from the block's own slot and its parent's, so it describes the
+/// work the block asked for, not what a cache happened to spare it: an
+/// `epoch_boundary` block whose epoch transition was precomputed is still one.
+///
+/// A replayed corpus is linear (`fetch` refuses a block whose parent is not
+/// the one before it), so a block's parent is always the previous block and
+/// there is no fourth class for a reorg: the state a block is applied to never
+/// sits on another branch.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum BlockClass {
+    /// `process_slots` crosses an epoch boundary: the block's slot is in a later
+    /// epoch than its parent's.
+    EpochBoundary,
+    /// Within one epoch, but the parent is more than one slot back.
+    AfterSkip,
+    /// The parent is the previous slot, in the same epoch.
+    Ordinary,
+}
+
+impl BlockClass {
+    /// The label a class is reported under, which is also its JSON value.
+    pub(crate) const fn label(self) -> &'static str {
+        match self {
+            BlockClass::EpochBoundary => "epoch_boundary",
+            BlockClass::AfterSkip => "after_skip",
+            BlockClass::Ordinary => "ordinary",
+        }
+    }
+
+    /// Classify a block at `slot` applied to a parent post-state at
+    /// `parent_slot`. An epoch crossing takes precedence over a skip: a block
+    /// that skipped slots across a boundary is dominated by the epoch work.
+    pub(crate) const fn classify(parent_slot: u64, slot: u64, slots_per_epoch: u64) -> Self {
+        if parent_slot / slots_per_epoch < slot / slots_per_epoch {
+            BlockClass::EpochBoundary
+        } else if slot > parent_slot + 1 {
+            BlockClass::AfterSkip
+        } else {
+            BlockClass::Ordinary
+        }
+    }
+}
+
 #[derive(Debug, Serialize)]
 pub(crate) struct Params {
     pub mode: &'static str,
@@ -30,6 +79,10 @@ pub(crate) struct Params {
 pub(crate) struct Sample {
     pub iteration: u64,
     pub slot: u64,
+    /// The slot of the block this one was applied to, which is what `class`
+    /// was decided from.
+    pub parent_slot: u64,
+    pub class: BlockClass,
     /// Determinism checksum: two runs over one corpus must produce the same
     /// roots in the same order.
     pub block_root: String,
@@ -37,44 +90,51 @@ pub(crate) struct Sample {
     /// Per-phase seconds from `lean_block_import_phase_seconds` sum deltas.
     /// A phase that did not run is absent, not zero.
     pub phases: BTreeMap<String, f64>,
+    /// Per-step seconds of the beacon state transition, from
+    /// `lean_beacon_stf_phase_seconds` sum deltas over the `import` context
+    /// only, so nothing a background thread did is in them. A step that did not
+    /// run is absent, not zero; one that ran several times (`process_epoch`
+    /// across a long skip) is the sum of its runs.
+    pub stf_phases: BTreeMap<String, f64>,
+    /// What the `stf` phase spent outside every step in `stf_phases`: cloning
+    /// the parent state, building the epoch-boundary balances, and anything
+    /// else not given a step of its own. Absent when `stf` did not run.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stf_unattributed_seconds: Option<f64>,
+    /// User plus system CPU seconds the whole process spent during the import,
+    /// all threads included. Absent if the host refused the reading.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cpu_seconds: Option<f64>,
+    /// Resident set size right after the import. Linux only: absent elsewhere.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub rss_bytes: Option<u64>,
+    /// The process's peak resident set size so far, a high-water mark.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub peak_rss_bytes: Option<u64>,
     /// `"imported"`. A run that reaches a report has no other outcome: a
     /// held or rejected block aborts it.
     pub outcome: &'static str,
 }
 
+/// The distributions of a set of samples: every block of the run, or the
+/// blocks of one class.
 #[derive(Debug, Serialize)]
-pub(crate) struct Summary {
+pub(crate) struct Distribution {
     pub phases: BTreeMap<String, Stats>,
+    pub stf_phases: BTreeMap<String, Stats>,
+    pub stf_unattributed: Stats,
     pub wall: Stats,
+    pub cpu: Stats,
 }
 
-#[derive(Debug, Serialize)]
-pub(crate) struct Report {
-    pub schema_version: u32,
-    pub environment: Environment,
-    pub params: Params,
-    pub samples: Vec<Sample>,
-    pub summary: Summary,
-}
-
-impl Report {
-    pub(crate) fn new(environment: Environment, params: Params, samples: Vec<Sample>) -> Self {
+impl Distribution {
+    fn of(samples: &[&Sample]) -> Self {
         // Union over every sample's phase keys, not just the first sample's:
         // unlike a synthetic build (where every phase runs on every block), a
         // corpus block legitimately skips phases (decode, defer, parent_wait,
         // columns_wait), and which ones it skips can differ block to block.
-        let mut phase_names: BTreeSet<&str> = BTreeSet::new();
-        for sample in &samples {
-            phase_names.extend(sample.phases.keys().map(String::as_str));
-        }
-        let mut phases: BTreeMap<String, Stats> = BTreeMap::new();
-        for phase in phase_names {
-            let values: Vec<f64> = samples
-                .iter()
-                .filter_map(|sample| sample.phases.get(phase).copied())
-                .collect();
-            phases.insert(phase.to_string(), stats(&values));
-        }
+        let phases = phase_stats(samples, |sample| &sample.phases);
+        let stf_phases = phase_stats(samples, |sample| &sample.stf_phases);
         // No coefficient-of-variation warning, unlike the synthetic report:
         // there every iteration is the same work, so spread is noise, while
         // here every sample is a different block, and the spread is mostly
@@ -86,13 +146,166 @@ impl Report {
                 .map(|sample| sample.wall_seconds)
                 .collect::<Vec<_>>(),
         );
+        let cpu = stats(
+            &samples
+                .iter()
+                .filter_map(|sample| sample.cpu_seconds)
+                .collect::<Vec<_>>(),
+        );
+        let stf_unattributed = stats(
+            &samples
+                .iter()
+                .filter_map(|sample| sample.stf_unattributed_seconds)
+                .collect::<Vec<_>>(),
+        );
+        Self {
+            phases,
+            stf_phases,
+            stf_unattributed,
+            wall,
+            cpu,
+        }
+    }
+}
+
+/// Stats per phase name over `samples`, for the phase map `select` picks.
+fn phase_stats(
+    samples: &[&Sample],
+    select: impl Fn(&Sample) -> &BTreeMap<String, f64>,
+) -> BTreeMap<String, Stats> {
+    let mut names: BTreeSet<&str> = BTreeSet::new();
+    for sample in samples {
+        names.extend(select(sample).keys().map(String::as_str));
+    }
+    names
+        .into_iter()
+        .map(|name| {
+            let values: Vec<f64> = samples
+                .iter()
+                .filter_map(|sample| select(sample).get(name).copied())
+                .collect();
+            (name.to_string(), stats(&values))
+        })
+        .collect()
+}
+
+/// One block class's share of the run.
+#[derive(Debug, Serialize)]
+pub(crate) struct ClassSummary {
+    pub blocks: usize,
+    #[serde(flatten)]
+    pub distribution: Distribution,
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct Summary {
+    #[serde(flatten)]
+    pub overall: Distribution,
+    /// Only the classes at least one block fell into.
+    pub by_class: BTreeMap<String, ClassSummary>,
+}
+
+/// The lookups one cache served during the measured blocks.
+#[derive(Debug, Serialize)]
+pub(crate) struct LookupCounts {
+    pub hit: u64,
+    pub miss: u64,
+    /// Lookups that could not be keyed and so were computed for one caller and
+    /// not cached. Absent for caches that have no such outcome.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub unkeyable: Option<u64>,
+    /// `hit / (hit + miss)`, absent when nothing was looked up. Unkeyable
+    /// lookups are not in the denominator: they never had a chance to hit.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub hit_rate: Option<f64>,
+}
+
+impl LookupCounts {
+    /// From the per-`result` counter deltas of one cache. `keyless` says
+    /// whether the cache has an `unkeyable` outcome to report.
+    pub(crate) fn from_results(results: &BTreeMap<String, u64>, keyless: bool) -> Self {
+        let get = |result: &str| results.get(result).copied().unwrap_or(0);
+        let (hit, miss) = (get("hit"), get("miss"));
+        let hit_rate = (hit + miss > 0).then(|| hit as f64 / (hit + miss) as f64);
+        Self {
+            hit,
+            miss,
+            unkeyable: keyless.then(|| get("unkeyable")),
+            hit_rate,
+        }
+    }
+}
+
+/// The lookup counters of the caches the state transition leans on, as deltas
+/// over the measured blocks (warm-up excluded). A replay has no epoch
+/// precompute worker, so `epoch_precompute` reports what an epoch-crossing
+/// import costs when nothing has precomputed its boundary: its misses are the
+/// epoch-boundary blocks.
+#[derive(Debug, Serialize)]
+pub(crate) struct Caches {
+    pub epoch_precompute: LookupCounts,
+    pub committee_cache: LookupCounts,
+    pub total_active_balance: LookupCounts,
+    pub justified_balances: LookupCounts,
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct Report {
+    pub schema_version: u32,
+    pub environment: Environment,
+    pub params: Params,
+    pub samples: Vec<Sample>,
+    pub summary: Summary,
+    pub caches: Caches,
+}
+
+/// The JSON layout version of this report.
+///
+/// 2: samples gained `parent_slot`, `class`, `stf_phases`,
+/// `stf_unattributed_seconds`, `cpu_seconds`, `rss_bytes` and
+/// `peak_rss_bytes`; every `Stats` gained `p95_seconds` and `p99_seconds`;
+/// the summary gained `stf_phases`, `stf_unattributed`, `cpu` and `by_class`;
+/// the report gained `caches`.
+pub(crate) const SCHEMA_VERSION: u32 = 2;
+
+impl Report {
+    pub(crate) fn new(
+        environment: Environment,
+        params: Params,
+        samples: Vec<Sample>,
+        caches: Caches,
+    ) -> Self {
+        let everyone: Vec<&Sample> = samples.iter().collect();
+        let overall = Distribution::of(&everyone);
+
+        let mut by_class: BTreeMap<String, ClassSummary> = BTreeMap::new();
+        for class in [
+            BlockClass::EpochBoundary,
+            BlockClass::AfterSkip,
+            BlockClass::Ordinary,
+        ] {
+            let members: Vec<&Sample> = samples
+                .iter()
+                .filter(|sample| sample.class == class)
+                .collect();
+            if !members.is_empty() {
+                by_class.insert(
+                    class.label().to_string(),
+                    ClassSummary {
+                        blocks: members.len(),
+                        distribution: Distribution::of(&members),
+                    },
+                );
+            }
+        }
 
         Self {
-            schema_version: 1,
+            schema_version: SCHEMA_VERSION,
             environment,
             params,
             samples,
-            summary: Summary { phases, wall },
+            summary: Summary { overall, by_class },
+            caches,
         }
     }
 
@@ -141,7 +354,11 @@ impl Report {
         for phase in &phases {
             let _ = write!(out, " {phase:>16}");
         }
-        let _ = writeln!(out, " {:>10} {:>12}", "wall", "root");
+        let _ = writeln!(
+            out,
+            " {:>14} {:>9} {:>10} {:>10} {:>12}",
+            "class", "cpu", "stf_unattr", "wall", "root"
+        );
 
         for sample in &self.samples {
             let _ = write!(out, "  {:<5}", sample.iteration);
@@ -157,35 +374,107 @@ impl Report {
             }
             let _ = writeln!(
                 out,
-                " {:>10} {:>12}",
+                " {:>14} {:>9} {:>10} {:>10} {:>12}",
+                sample.class.label(),
+                optional_ms(sample.cpu_seconds),
+                optional_ms(sample.stf_unattributed_seconds),
                 format_ms(sample.wall_seconds),
                 &sample.block_root[..10],
             );
         }
 
+        // The per-block STF steps are not columns above: there are close to
+        // twenty of them, so the table would stop fitting any terminal. They
+        // are in each sample's `stf_phases` in the JSON, and summarized here.
+        let _ = writeln!(out);
+        let overall = &self.summary.overall;
+        write_stats_header(&mut out, "phase");
+        for (phase, stats) in &overall.phases {
+            let _ = writeln!(out, "{}", stats_row(phase, stats));
+        }
+        let _ = writeln!(out, "{}", stats_row("wall", &overall.wall));
+        let _ = writeln!(out, "{}", stats_row("cpu", &overall.cpu));
+
         let _ = writeln!(out);
+        write_stats_header(&mut out, "stf step");
+        for (phase, stats) in &overall.stf_phases {
+            let _ = writeln!(out, "{}", stats_row(phase, stats));
+        }
         let _ = writeln!(
             out,
-            "  {:<18} {:>5} {:>10} {:>10} {:>10} {:>10} {:>10}",
-            "phase", "count", "min", "mean", "p50", "p90", "max"
+            "{}",
+            stats_row("(unattributed)", &overall.stf_unattributed)
         );
-        for (phase, stats) in &self.summary.phases {
-            let _ = writeln!(out, "{}", stats_row(phase, stats));
+
+        let _ = writeln!(out);
+        let _ = writeln!(
+            out,
+            "  {:<16} {:>6} {:>10} {:>10} {:>10} {:>10} {:>10}",
+            "class", "blocks", "wall p50", "wall p95", "wall p99", "stf p50", "stf p99"
+        );
+        for (class, summary) in &self.summary.by_class {
+            let stf = summary.distribution.phases.get("stf");
+            let _ = writeln!(
+                out,
+                "  {:<16} {:>6} {:>10} {:>10} {:>10} {:>10} {:>10}",
+                class,
+                summary.blocks,
+                format_ms(summary.distribution.wall.p50_seconds),
+                format_ms(summary.distribution.wall.p95_seconds),
+                format_ms(summary.distribution.wall.p99_seconds),
+                optional_ms(stf.map(|stats| stats.p50_seconds)),
+                optional_ms(stf.map(|stats| stats.p99_seconds)),
+            );
+        }
+
+        let _ = writeln!(out);
+        let caches = &self.caches;
+        for (name, counts) in [
+            ("epoch_precompute", &caches.epoch_precompute),
+            ("committee_cache", &caches.committee_cache),
+            ("total_active_balance", &caches.total_active_balance),
+            ("justified_balances", &caches.justified_balances),
+        ] {
+            let rate = counts
+                .hit_rate
+                .map_or_else(|| "-".to_string(), |rate| format!("{:.1}%", rate * 100.0));
+            let unkeyable = counts
+                .unkeyable
+                .map_or_else(String::new, |count| format!(" unkeyable={count}"));
+            let _ = writeln!(
+                out,
+                "  cache {name:<21} hit={} miss={}{unkeyable} hit_rate={rate}",
+                counts.hit, counts.miss
+            );
         }
-        let _ = writeln!(out, "{}", stats_row("wall", &self.summary.wall));
         out
     }
 }
 
+/// A duration in milliseconds, or `-` where the reading is absent.
+fn optional_ms(seconds: Option<f64>) -> String {
+    seconds.map_or_else(|| "-".to_string(), format_ms)
+}
+
+fn write_stats_header(out: &mut String, first: &str) {
+    let _ = writeln!(
+        out,
+        "  {:<26} {:>5} {:>10} {:>10} {:>10} {:>10} {:>10} {:>10} {:>10}",
+        first, "count", "min", "mean", "p50", "p90", "p95", "p99", "max"
+    );
+}
+
 fn stats_row(name: &str, stats: &Stats) -> String {
     format!(
-        "  {:<18} {:>5} {:>10} {:>10} {:>10} {:>10} {:>10}",
+        "  {:<26} {:>5} {:>10} {:>10} {:>10} {:>10} {:>10} {:>10} {:>10}",
         name,
         stats.count,
         format_ms(stats.min_seconds),
         format_ms(stats.mean_seconds),
         format_ms(stats.p50_seconds),
         format_ms(stats.p90_seconds),
+        format_ms(stats.p95_seconds),
+        format_ms(stats.p99_seconds),
         format_ms(stats.max_seconds),
     )
 }
@@ -194,13 +483,22 @@ fn stats_row(name: &str, stats: &Stats) -> String {
 mod tests {
     use super::*;
 
+    const SLOTS_PER_EPOCH: u64 = 32;
+
     fn sample(iteration: u64, phases: &[(&str, f64)]) -> Sample {
         Sample {
             iteration,
             slot: iteration,
+            parent_slot: iteration.saturating_sub(1),
+            class: BlockClass::Ordinary,
             block_root: format!("0x{iteration:064x}"),
             wall_seconds: 0.01,
             phases: phases.iter().map(|(k, v)| ((*k).to_string(), *v)).collect(),
+            stf_phases: BTreeMap::new(),
+            stf_unattributed_seconds: None,
+            cpu_seconds: None,
+            rss_bytes: None,
+            peak_rss_bytes: None,
             outcome: "imported",
         }
     }
@@ -220,11 +518,25 @@ mod tests {
         }
     }
 
+    fn no_caches() -> Caches {
+        let none = BTreeMap::new();
+        Caches {
+            epoch_precompute: LookupCounts::from_results(&none, false),
+            committee_cache: LookupCounts::from_results(&none, true),
+            total_active_balance: LookupCounts::from_results(&none, true),
+            justified_balances: LookupCounts::from_results(&none, false),
+        }
+    }
+
+    fn report(samples: Vec<Sample>) -> Report {
+        Report::new(Environment::collect(), params(), samples, no_caches())
+    }
+
     #[test]
     fn the_block_delay_is_recorded_in_the_json_and_the_header() {
         let mut params = params();
         params.block_delay_ms = 750;
-        let report = Report::new(Environment::collect(), params, Vec::new());
+        let report = Report::new(Environment::collect(), params, Vec::new(), no_caches());
 
         let json: serde_json::Value = serde_json::from_str(&report.to_json().unwrap()).unwrap();
         assert_eq!(json["params"]["block_delay_ms"], 750);
@@ -233,7 +545,7 @@ mod tests {
 
     #[test]
     fn the_range_prints_inclusive_as_fetch_takes_it() {
-        let report = Report::new(Environment::collect(), params(), Vec::new());
+        let report = report(Vec::new());
 
         assert!(
             report.human_table().contains("range=[1, 3]"),
@@ -248,13 +560,14 @@ mod tests {
             sample(1, &[("stf", 0.1), ("decode", 0.2)]),
             sample(2, &[("stf", 0.3)]),
         ];
-        let report = Report::new(Environment::collect(), params(), samples);
+        let report = report(samples);
 
         // `decode` ran on one of two samples: its Stats reflect only that one
         // observation, not a zero-filled second one.
-        assert_eq!(report.summary.phases["stf"].count, 2);
-        assert_eq!(report.summary.phases["decode"].count, 1);
-        assert!((report.summary.phases["decode"].mean_seconds - 0.2).abs() < 1e-9);
+        let phases = &report.summary.overall.phases;
+        assert_eq!(phases["stf"].count, 2);
+        assert_eq!(phases["decode"].count, 1);
+        assert!((phases["decode"].mean_seconds - 0.2).abs() < 1e-9);
 
         let table = report.human_table();
         assert!(table.contains("decode"), "column missing from human_table");
@@ -266,4 +579,183 @@ mod tests {
             "missing phase should render as a placeholder, not 0"
         );
     }
+
+    #[test]
+    fn a_block_crossing_an_epoch_boundary_is_an_epoch_boundary_whatever_it_skipped() {
+        let classify = |parent, slot| BlockClass::classify(parent, slot, SLOTS_PER_EPOCH);
+
+        // The last slot of an epoch to the first of the next, no skip.
+        assert_eq!(classify(31, 32), BlockClass::EpochBoundary);
+        // A skip that lands in the next epoch is still dominated by the epoch.
+        assert_eq!(classify(30, 33), BlockClass::EpochBoundary);
+        // Several epochs at once.
+        assert_eq!(classify(31, 100), BlockClass::EpochBoundary);
+    }
+
+    #[test]
+    fn a_block_that_skipped_slots_inside_one_epoch_is_after_skip() {
+        let classify = |parent, slot| BlockClass::classify(parent, slot, SLOTS_PER_EPOCH);
+
+        assert_eq!(classify(33, 35), BlockClass::AfterSkip);
+        // The skip ends exactly on the epoch's last slot: still one epoch.
+        assert_eq!(classify(32, 63), BlockClass::AfterSkip);
+    }
+
+    #[test]
+    fn a_block_one_slot_after_its_parent_in_the_same_epoch_is_ordinary() {
+        let classify = |parent, slot| BlockClass::classify(parent, slot, SLOTS_PER_EPOCH);
+
+        assert_eq!(classify(33, 34), BlockClass::Ordinary);
+        assert_eq!(classify(62, 63), BlockClass::Ordinary);
+    }
+
+    #[test]
+    fn the_class_label_is_the_json_value() {
+        for class in [
+            BlockClass::EpochBoundary,
+            BlockClass::AfterSkip,
+            BlockClass::Ordinary,
+        ] {
+            assert_eq!(
+                serde_json::to_value(class).unwrap(),
+                serde_json::Value::String(class.label().to_string())
+            );
+        }
+    }
+
+    #[test]
+    fn the_summary_splits_by_class_and_leaves_out_classes_nothing_fell_into() {
+        let mut boundary = sample(1, &[("stf", 0.9)]);
+        boundary.class = BlockClass::EpochBoundary;
+        boundary.wall_seconds = 1.0;
+        let ordinary_a = sample(2, &[("stf", 0.1)]);
+        let ordinary_b = sample(3, &[("stf", 0.2), ("decode", 0.01)]);
+        let report = report(vec![boundary, ordinary_a, ordinary_b]);
+
+        let by_class = &report.summary.by_class;
+        assert_eq!(
+            by_class.len(),
+            2,
+            "no after_skip block, so no after_skip row"
+        );
+        assert_eq!(by_class["epoch_boundary"].blocks, 1);
+        assert_eq!(by_class["ordinary"].blocks, 2);
+        assert!(!by_class.contains_key("after_skip"));
+
+        let ordinary = &by_class["ordinary"].distribution;
+        assert!((ordinary.phases["stf"].mean_seconds - 0.15).abs() < 1e-9);
+        assert_eq!(ordinary.phases["decode"].count, 1);
+        assert!((by_class["epoch_boundary"].distribution.wall.mean_seconds - 1.0).abs() < 1e-9);
+        // The whole-run summary still covers all three.
+        assert_eq!(report.summary.overall.phases["stf"].count, 3);
+    }
+
+    #[test]
+    fn the_report_carries_the_new_shape_at_the_documented_paths() {
+        let mut block = sample(1, &[("stf", 0.5)]);
+        block.stf_phases = [("process_epoch".to_string(), 0.3)].into_iter().collect();
+        block.stf_unattributed_seconds = Some(0.05);
+        block.cpu_seconds = Some(0.7);
+        block.rss_bytes = Some(1 << 30);
+        block.peak_rss_bytes = Some(2 << 30);
+        let mut bare = sample(2, &[]);
+        bare.class = BlockClass::AfterSkip;
+
+        let mut results = BTreeMap::new();
+        results.insert("hit".to_string(), 3);
+        results.insert("miss".to_string(), 1);
+        let caches = Caches {
+            epoch_precompute: LookupCounts::from_results(&results, false),
+            committee_cache: LookupCounts::from_results(&results, true),
+            ..no_caches()
+        };
+        let report = Report::new(Environment::collect(), params(), vec![block, bare], caches);
+
+        let json: serde_json::Value = serde_json::from_str(&report.to_json().unwrap()).unwrap();
+        assert_eq!(json["schema_version"], 2);
+
+        let first = &json["samples"][0];
+        assert_eq!(first["class"], "ordinary");
+        assert_eq!(first["parent_slot"], 0);
+        assert_eq!(first["stf_phases"]["process_epoch"], 0.3);
+        assert_eq!(first["cpu_seconds"], 0.7);
+        assert_eq!(first["rss_bytes"], 1u64 << 30);
+        assert_eq!(first["peak_rss_bytes"], 2u64 << 30);
+
+        // A reading that is absent is omitted from the sample, not null or 0.
+        let second = json["samples"][1].as_object().unwrap();
+        for key in [
+            "cpu_seconds",
+            "rss_bytes",
+            "peak_rss_bytes",
+            "stf_unattributed_seconds",
+        ] {
+            assert!(!second.contains_key(key), "{key} should be absent");
+        }
+        assert_eq!(second["class"], "after_skip");
+
+        // Summary paths: the old ones are where they were, flattened.
+        let summary = &json["summary"];
+        assert_eq!(summary["phases"]["stf"]["count"], 1);
+        assert!(summary["phases"]["stf"]["p95_seconds"].is_number());
+        assert!(summary["phases"]["stf"]["p99_seconds"].is_number());
+        assert_eq!(summary["stf_phases"]["process_epoch"]["count"], 1);
+        assert_eq!(summary["cpu"]["count"], 1, "only one sample had a reading");
+        assert_eq!(summary["stf_unattributed"]["count"], 1);
+        assert_eq!(summary["wall"]["count"], 2);
+        assert_eq!(summary["by_class"]["ordinary"]["blocks"], 1);
+        assert_eq!(summary["by_class"]["after_skip"]["blocks"], 1);
+
+        let caches = &json["caches"];
+        assert_eq!(caches["epoch_precompute"]["hit"], 3);
+        assert_eq!(caches["epoch_precompute"]["miss"], 1);
+        assert_eq!(caches["epoch_precompute"]["hit_rate"], 0.75);
+        assert!(
+            !caches["epoch_precompute"]
+                .as_object()
+                .unwrap()
+                .contains_key("unkeyable")
+        );
+        assert_eq!(caches["committee_cache"]["unkeyable"], 0);
+        assert!(
+            !caches["justified_balances"]
+                .as_object()
+                .unwrap()
+                .contains_key("hit_rate"),
+            "no lookups means no rate, not a rate of zero"
+        );
+    }
+
+    #[test]
+    fn the_hit_rate_ignores_unkeyable_lookups() {
+        let results: BTreeMap<String, u64> = [("hit", 1), ("miss", 1), ("unkeyable", 98)]
+            .into_iter()
+            .map(|(result, count)| (result.to_string(), count))
+            .collect();
+
+        let counts = LookupCounts::from_results(&results, true);
+        assert_eq!(counts.unkeyable, Some(98));
+        assert_eq!(counts.hit_rate, Some(0.5));
+    }
+
+    #[test]
+    fn the_human_table_names_the_new_columns_and_sections() {
+        let mut block = sample(1, &[("stf", 0.5)]);
+        block.stf_phases = [("state_root".to_string(), 0.2)].into_iter().collect();
+        block.cpu_seconds = Some(0.6);
+        let table = report(vec![block]).human_table();
+
+        for expected in [
+            "class",
+            "cpu",
+            "p95",
+            "p99",
+            "state_root",
+            "(unattributed)",
+            "epoch_precompute",
+            "justified_balances",
+        ] {
+            assert!(table.contains(expected), "`{expected}` missing:\n{table}");
+        }
+    }
 }
diff --git a/crates/blockchain/state_transition/src/beacon/fork_choice.rs b/crates/blockchain/state_transition/src/beacon/fork_choice.rs
index ac83d85fe..393cad698 100644
--- a/crates/blockchain/state_transition/src/beacon/fork_choice.rs
+++ b/crates/blockchain/state_transition/src/beacon/fork_choice.rs
@@ -1063,6 +1063,12 @@ fn transition_block(
     committees: &CommitteeCache,
     active_balances: &ActiveBalanceCache,
 ) -> (BeaconState, Result<()>) {
+    // Every STF phase observed below is the import's, and none of it belongs
+    // to the epoch precompute worker, which runs `process_epoch` on its own
+    // thread at the same time and so never sees this guard. See
+    // `metrics::StfContext` for why the series is split by caller.
+    let _import_context = metrics::import_context();
+
     let block_slot = signed_block.slot();
     let epoch = compute_epoch_at_slot(block_slot);
     let epoch_start = compute_start_slot_at_epoch(epoch);
diff --git a/crates/blockchain/state_transition/src/beacon/stf/altair.rs b/crates/blockchain/state_transition/src/beacon/stf/altair.rs
index 2b33e79be..505330c97 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/altair.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/altair.rs
@@ -304,6 +304,18 @@ pub fn process_sync_aggregate(
     state: &mut BeaconState,
     sync_aggregate: &altair::SyncAggregate,
     active_balances: &ActiveBalanceCache,
+) -> Result<()> {
+    // Timed here, not at the six forks' call sites, since every fork from
+    // altair on reaches this one function.
+    crate::metrics::timed(crate::metrics::StfPhase::SyncAggregate, || {
+        apply_sync_aggregate(state, sync_aggregate, active_balances)
+    })
+}
+
+fn apply_sync_aggregate(
+    state: &mut BeaconState,
+    sync_aggregate: &altair::SyncAggregate,
+    active_balances: &ActiveBalanceCache,
 ) -> Result<()> {
     let committee_bits = &sync_aggregate.sync_committee_bits;
 
diff --git a/crates/blockchain/state_transition/src/beacon/stf/bellatrix.rs b/crates/blockchain/state_transition/src/beacon/stf/bellatrix.rs
index e9287513e..7f91ff596 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/bellatrix.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/bellatrix.rs
@@ -41,6 +41,8 @@ use crate::beacon::primitives::{
     Bytes32, ExecutionAddress, ExecutionBlockHash, HashTreeRoot as _, Root, Slot, Uint256,
 };
 
+use crate::metrics::{StfPhase, timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -72,16 +74,23 @@ pub fn process_block(
     active_balances: &ActiveBalanceCache,
     options: super::block::BlockProcessingOptions,
 ) -> Result<()> {
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
-    if is_execution_enabled(state, &block.body.execution_payload)? {
-        process_execution_payload(state, &block.body.execution_payload, config, engine)?;
-    }
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        if is_execution_enabled(state, &block.body.execution_payload)? {
+            process_execution_payload(state, &block.body.execution_payload, config, engine)?;
+        }
+        Ok(())
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     super::operations::process_operations(
diff --git a/crates/blockchain/state_transition/src/beacon/stf/block.rs b/crates/blockchain/state_transition/src/beacon/stf/block.rs
index dcaa32684..cec84e143 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/block.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/block.rs
@@ -32,6 +32,8 @@ use crate::beacon::helpers::misc::compute_signing_root;
 use crate::beacon::preset;
 use crate::beacon::primitives::{BlsSignature, HashTreeRoot as _, Root, Slot, ValidatorIndex};
 
+use crate::metrics::{StfPhase, timed};
+
 use super::{ExecutionEngine, bellatrix, capella, deneb, electra, fulu, gloas};
 
 /// Dispatches on the block's fork and runs that fork's own block processing.
@@ -139,13 +141,17 @@ pub fn process_block_phase0(
     active_balances: &ActiveBalanceCache,
     options: BlockProcessingOptions,
 ) -> Result<()> {
-    process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
     process_randao(state, &block.body.randao_reveal, options.randao)?;
     process_eth1_data(state, &block.body.eth1_data)?;
     crate::beacon::stf::operations::process_operations(
@@ -176,13 +182,17 @@ pub fn process_block_altair(
     active_balances: &ActiveBalanceCache,
     options: BlockProcessingOptions,
 ) -> Result<()> {
-    process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
     process_randao(state, &block.body.randao_reveal, options.randao)?;
     process_eth1_data(state, &block.body.eth1_data)?;
     crate::beacon::stf::operations::process_operations(
@@ -287,6 +297,16 @@ pub fn process_randao(
     state: &mut BeaconState,
     randao_reveal: &BlsSignature,
     verification: RandaoVerification,
+) -> Result<()> {
+    timed(StfPhase::Randao, || {
+        apply_randao(state, randao_reveal, verification)
+    })
+}
+
+fn apply_randao(
+    state: &mut BeaconState,
+    randao_reveal: &BlsSignature,
+    verification: RandaoVerification,
 ) -> Result<()> {
     let epoch = get_current_epoch(state);
     let proposer_index = get_beacon_proposer_index(state)?;
@@ -316,6 +336,10 @@ pub fn process_randao(
 /// count used for the majority check includes the vote just appended, so a
 /// single block can supply the deciding vote.
 pub fn process_eth1_data(state: &mut BeaconState, eth1_data: &Eth1Data) -> Result<()> {
+    timed(StfPhase::Eth1Data, || apply_eth1_data(state, eth1_data))
+}
+
+fn apply_eth1_data(state: &mut BeaconState, eth1_data: &Eth1Data) -> Result<()> {
     state.eth1_data_votes_mut().push(eth1_data.clone())?;
 
     let matching_votes = state
diff --git a/crates/blockchain/state_transition/src/beacon/stf/capella.rs b/crates/blockchain/state_transition/src/beacon/stf/capella.rs
index f773d6da7..394877d87 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/capella.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/capella.rs
@@ -59,6 +59,8 @@ use crate::beacon::helpers::mutators::decrease_balance;
 use crate::beacon::preset;
 use crate::beacon::primitives::{H256, HashTreeRoot as _};
 
+use crate::metrics::{StfPhase, timed, try_for_each_timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -81,15 +83,21 @@ pub fn process_block(
     active_balances: &ActiveBalanceCache,
     options: super::block::BlockProcessingOptions,
 ) -> Result<()> {
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
-    process_withdrawals(state, &block.body.execution_payload)?;
-    process_execution_payload(state, &block.body.execution_payload, config, engine)?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        process_withdrawals(state, &block.body.execution_payload)?;
+        process_execution_payload(state, &block.body.execution_payload, config, engine)
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     process_operations(
@@ -381,9 +389,11 @@ pub fn process_operations(
         committees,
         active_balances,
     )?;
-    for signed_change in bls_to_execution_changes {
-        process_bls_to_execution_change(state, signed_change, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::BlsToExecutionChanges,
+        bls_to_execution_changes,
+        |signed_change| process_bls_to_execution_change(state, signed_change, config),
+    )?;
     Ok(())
 }
 
diff --git a/crates/blockchain/state_transition/src/beacon/stf/deneb.rs b/crates/blockchain/state_transition/src/beacon/stf/deneb.rs
index 72b5ac946..00044c933 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/deneb.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/deneb.rs
@@ -53,6 +53,8 @@ use crate::beacon::primitives::{
     Bytes32, Gwei, H256, HashTreeRoot as _, KzgCommitment, ParticipationFlags, ValidatorIndex,
 };
 
+use crate::metrics::{StfPhase, timed, try_for_each_timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -81,21 +83,27 @@ pub fn process_block(
     active_balances: &ActiveBalanceCache,
     options: super::block::BlockProcessingOptions,
 ) -> Result<()> {
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
-    process_withdrawals(state, &block.body.execution_payload)?;
-    process_execution_payload(
-        state,
-        &block.body.execution_payload,
-        &block.body.blob_kzg_commitments,
-        config,
-        engine,
-    )?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        process_withdrawals(state, &block.body.execution_payload)?;
+        process_execution_payload(
+            state,
+            &block.body.execution_payload,
+            &block.body.blob_kzg_commitments,
+            config,
+            engine,
+        )
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     process_operations(
@@ -165,24 +173,32 @@ fn process_operations(
         "len(body.deposits) == min(MAX_DEPOSITS, eth1_data.deposit_count - eth1_deposit_index)",
     )?;
 
-    for proposer_slashing in proposer_slashings {
-        super::operations::process_proposer_slashing(state, proposer_slashing, config)?;
-    }
-    for attester_slashing in attester_slashings {
-        super::operations::process_attester_slashing(state, attester_slashing, config)?;
-    }
-    for attestation in attestations {
-        process_attestation(state, attestation, committees, active_balances)?;
-    }
-    for deposit in deposits {
-        super::operations::process_deposit(state, deposit, config)?;
-    }
-    for voluntary_exit in voluntary_exits {
-        process_voluntary_exit(state, voluntary_exit, config)?;
-    }
-    for signed_change in bls_to_execution_changes {
-        super::capella::process_bls_to_execution_change(state, signed_change, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::ProposerSlashings,
+        proposer_slashings,
+        |slashing| super::operations::process_proposer_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::AttesterSlashings,
+        attester_slashings,
+        |slashing| super::operations::process_attester_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(StfPhase::Attestations, attestations, |attestation| {
+        process_attestation(state, attestation, committees, active_balances)
+    })?;
+    try_for_each_timed(StfPhase::Deposits, deposits, |deposit| {
+        super::operations::process_deposit(state, deposit, config)
+    })?;
+    try_for_each_timed(StfPhase::VoluntaryExits, voluntary_exits, |exit| {
+        process_voluntary_exit(state, exit, config)
+    })?;
+    try_for_each_timed(
+        StfPhase::BlsToExecutionChanges,
+        bls_to_execution_changes,
+        |signed_change| {
+            super::capella::process_bls_to_execution_change(state, signed_change, config)
+        },
+    )?;
 
     Ok(())
 }
diff --git a/crates/blockchain/state_transition/src/beacon/stf/electra.rs b/crates/blockchain/state_transition/src/beacon/stf/electra.rs
index 9b3519caf..b53265737 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/electra.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/electra.rs
@@ -82,6 +82,8 @@ use crate::beacon::primitives::{
     WithdrawalIndex,
 };
 
+use crate::metrics::{StfPhase, timed, try_for_each_timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -1802,28 +1804,40 @@ pub fn process_operations(
         verify(body.deposits.is_empty(), "len(body.deposits) == 0")?;
     }
 
-    for proposer_slashing in body.proposer_slashings.iter() {
-        super::operations::process_proposer_slashing(state, proposer_slashing, config)?;
-    }
-    for attester_slashing in body.attester_slashings.iter() {
-        process_attester_slashing(state, attester_slashing, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::ProposerSlashings,
+        body.proposer_slashings.iter(),
+        |slashing| super::operations::process_proposer_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::AttesterSlashings,
+        body.attester_slashings.iter(),
+        |slashing| process_attester_slashing(state, slashing, config),
+    )?;
     // [Modified in Electra:EIP7549]
-    for attestation in body.attestations.iter() {
-        process_attestation(state, attestation, committees, active_balances)?;
-    }
-    for deposit in body.deposits.iter() {
-        process_deposit(state, deposit, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::Attestations,
+        body.attestations.iter(),
+        |attestation| process_attestation(state, attestation, committees, active_balances),
+    )?;
+    try_for_each_timed(StfPhase::Deposits, body.deposits.iter(), |deposit| {
+        process_deposit(state, deposit, config)
+    })?;
     // [Modified in Electra:EIP7251]
-    for voluntary_exit in body.voluntary_exits.iter() {
-        process_voluntary_exit(state, voluntary_exit, config)?;
-    }
-    for signed_change in body.bls_to_execution_changes.iter() {
-        process_bls_to_execution_change(state, signed_change, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::VoluntaryExits,
+        body.voluntary_exits.iter(),
+        |exit| process_voluntary_exit(state, exit, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::BlsToExecutionChanges,
+        body.bls_to_execution_changes.iter(),
+        |signed_change| process_bls_to_execution_change(state, signed_change, config),
+    )?;
     // [New in Electra:EIP6110:EIP7002:EIP7251]
-    process_execution_requests(state, &body.execution_requests, config)?;
+    timed(StfPhase::ExecutionRequests, || {
+        process_execution_requests(state, &body.execution_requests, config)
+    })?;
 
     Ok(())
 }
@@ -1854,15 +1868,21 @@ pub fn process_block(
     active_balances: &ActiveBalanceCache,
     options: super::block::BlockProcessingOptions,
 ) -> Result<()> {
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
-    process_withdrawals(state, &block.body.execution_payload)?;
-    process_execution_payload(state, &block.body, config, engine)?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        process_withdrawals(state, &block.body.execution_payload)?;
+        process_execution_payload(state, &block.body, config, engine)
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     process_operations(state, &block.body, config, committees, active_balances)?;
diff --git a/crates/blockchain/state_transition/src/beacon/stf/fulu.rs b/crates/blockchain/state_transition/src/beacon/stf/fulu.rs
index 1a8bc3b58..b237c6b9c 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/fulu.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/fulu.rs
@@ -50,6 +50,8 @@ use crate::beacon::helpers::electra::pending_queue_fields;
 use crate::beacon::helpers::fulu::{fulu_state, fulu_state_ref};
 use crate::beacon::primitives::{Bytes32, HashTreeRoot as _};
 
+use crate::metrics::{StfPhase, timed, try_for_each_timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -70,15 +72,21 @@ pub fn process_block(
     active_balances: &ActiveBalanceCache,
     options: super::block::BlockProcessingOptions,
 ) -> Result<()> {
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
-    super::electra::process_withdrawals(state, &block.body.execution_payload)?;
-    process_execution_payload(state, &block.body, config, engine)?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        super::electra::process_withdrawals(state, &block.body.execution_payload)?;
+        process_execution_payload(state, &block.body, config, engine)
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     process_operations(state, &block.body, config, committees, active_balances)?;
@@ -228,24 +236,40 @@ pub fn process_operations(
     // [Modified in Fulu]
     verify(body.deposits.is_empty(), "len(body.deposits) == 0")?;
 
-    for proposer_slashing in body.proposer_slashings.iter() {
-        super::operations::process_proposer_slashing(state, proposer_slashing, config)?;
-    }
-    for attester_slashing in body.attester_slashings.iter() {
-        super::electra::process_attester_slashing(state, attester_slashing, config)?;
-    }
-    for attestation in body.attestations.iter() {
-        super::electra::process_attestation(state, attestation, committees, active_balances)?;
-    }
+    try_for_each_timed(
+        StfPhase::ProposerSlashings,
+        body.proposer_slashings.iter(),
+        |slashing| super::operations::process_proposer_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::AttesterSlashings,
+        body.attester_slashings.iter(),
+        |slashing| super::electra::process_attester_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::Attestations,
+        body.attestations.iter(),
+        |attestation| {
+            super::electra::process_attestation(state, attestation, committees, active_balances)
+        },
+    )?;
     // [Modified in Fulu]
     // Removed `process_deposit`
-    for voluntary_exit in body.voluntary_exits.iter() {
-        super::electra::process_voluntary_exit(state, voluntary_exit, config)?;
-    }
-    for signed_change in body.bls_to_execution_changes.iter() {
-        super::electra::process_bls_to_execution_change(state, signed_change, config)?;
-    }
-    process_execution_requests(state, &body.execution_requests, config)?;
+    try_for_each_timed(
+        StfPhase::VoluntaryExits,
+        body.voluntary_exits.iter(),
+        |exit| super::electra::process_voluntary_exit(state, exit, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::BlsToExecutionChanges,
+        body.bls_to_execution_changes.iter(),
+        |signed_change| {
+            super::electra::process_bls_to_execution_change(state, signed_change, config)
+        },
+    )?;
+    timed(StfPhase::ExecutionRequests, || {
+        process_execution_requests(state, &body.execution_requests, config)
+    })?;
 
     Ok(())
 }
diff --git a/crates/blockchain/state_transition/src/beacon/stf/gloas.rs b/crates/blockchain/state_transition/src/beacon/stf/gloas.rs
index 0ef4c1c8c..ca244edac 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/gloas.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/gloas.rs
@@ -77,6 +77,8 @@ use crate::beacon::primitives::{
     ValidatorIndex, WithdrawalIndex,
 };
 
+use crate::metrics::{StfPhase, timed, try_for_each_timed};
+
 use super::ExecutionEngine;
 
 // ---------------------------------------------------------------------------
@@ -109,20 +111,32 @@ pub fn process_block(
     // [New in Gloas:EIP7732]
     let parent_slot = state.latest_block_header().slot;
 
+    // Gloas has no execution payload step of its own: the parent's payload, the
+    // withdrawals and the bid are its equivalents, so they are observed as
+    // `execution_payload`, which therefore reaches the histogram twice per block.
+    //
     // [New in Gloas:EIP7732]
-    process_parent_execution_payload(state, block, config)?;
-    super::block::process_block_header(
-        state,
-        block.slot,
-        block.proposer_index,
-        block.parent_root,
-        block.body.hash_tree_root(),
-    )?;
+    timed(StfPhase::ExecutionPayload, || {
+        process_parent_execution_payload(state, block, config)
+    })?;
+    // The body's root is an argument, so it is hashed inside the timed span:
+    // it is part of what a header costs.
+    timed(StfPhase::BlockHeader, || {
+        super::block::process_block_header(
+            state,
+            block.slot,
+            block.proposer_index,
+            block.parent_root,
+            block.body.hash_tree_root(),
+        )
+    })?;
     // [Modified in Gloas:EIP7732]
-    process_withdrawals(state)?;
     // [Modified in Gloas:EIP7732] Removed `process_execution_payload`.
     // [New in Gloas:EIP7732]
-    process_execution_payload_bid(state, &block.body.signed_execution_payload_bid, config)?;
+    timed(StfPhase::ExecutionPayload, || -> Result<()> {
+        process_withdrawals(state)?;
+        process_execution_payload_bid(state, &block.body.signed_execution_payload_bid, config)
+    })?;
     super::block::process_randao(state, &block.body.randao_reveal, options.randao)?;
     super::block::process_eth1_data(state, &block.body.eth1_data)?;
     // [Modified in Gloas:EIP7732]
@@ -1621,28 +1635,48 @@ pub fn process_operations(
     )?;
 
     // [Modified in Gloas:EIP7732]
-    for proposer_slashing in body.proposer_slashings.iter() {
-        process_proposer_slashing(state, proposer_slashing, config)?;
-    }
-    for attester_slashing in body.attester_slashings.iter() {
-        process_attester_slashing(state, attester_slashing, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::ProposerSlashings,
+        body.proposer_slashings.iter(),
+        |slashing| process_proposer_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::AttesterSlashings,
+        body.attester_slashings.iter(),
+        |slashing| process_attester_slashing(state, slashing, config),
+    )?;
     // [Modified in Gloas:EIP7732]
-    for attestation in body.attestations.iter() {
-        process_attestation(state, attestation, parent_slot, committees, active_balances)?;
-    }
-    for voluntary_exit in body.voluntary_exits.iter() {
-        crate::beacon::stf::electra::process_voluntary_exit(state, voluntary_exit, config)?;
-    }
-    for signed_change in body.bls_to_execution_changes.iter() {
-        crate::beacon::stf::capella::process_bls_to_execution_change(state, signed_change, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::Attestations,
+        body.attestations.iter(),
+        |attestation| {
+            process_attestation(state, attestation, parent_slot, committees, active_balances)
+        },
+    )?;
+    try_for_each_timed(
+        StfPhase::VoluntaryExits,
+        body.voluntary_exits.iter(),
+        |exit| crate::beacon::stf::electra::process_voluntary_exit(state, exit, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::BlsToExecutionChanges,
+        body.bls_to_execution_changes.iter(),
+        |signed_change| {
+            crate::beacon::stf::capella::process_bls_to_execution_change(
+                state,
+                signed_change,
+                config,
+            )
+        },
+    )?;
     // [Modified in Gloas:EIP7732] Removed `process_deposit_request`,
     // `process_withdrawal_request`, and `process_consolidation_request`.
     // [New in Gloas:EIP7732]
-    for payload_attestation in body.payload_attestations.iter() {
-        process_payload_attestation(state, payload_attestation, config)?;
-    }
+    try_for_each_timed(
+        StfPhase::PayloadAttestations,
+        body.payload_attestations.iter(),
+        |payload_attestation| process_payload_attestation(state, payload_attestation, config),
+    )?;
 
     Ok(())
 }
diff --git a/crates/blockchain/state_transition/src/beacon/stf/mod.rs b/crates/blockchain/state_transition/src/beacon/stf/mod.rs
index 3dba6eef8..4aeab5e5e 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/mod.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/mod.rs
@@ -73,6 +73,8 @@ use crate::beacon::lean_state_unreachable;
 use crate::beacon::preset;
 use crate::beacon::primitives::{HashTreeRoot as _, Slot};
 use crate::beacon::{bls, config::Config, constants};
+use crate::metrics::{StfPhase, observe_stf_phase, timed};
+use std::time::{Duration, Instant};
 
 /// What the execution layer would answer for a block's payload.
 ///
@@ -178,10 +180,10 @@ pub fn apply_block(
     )?;
 
     if validate_result {
-        verify(
-            verify_block_signature(state, signed_block),
-            "block signature",
-        )?;
+        let signature_valid = timed(StfPhase::BlockSignature, || {
+            verify_block_signature(state, signed_block)
+        });
+        verify(signature_valid, "block signature")?;
     }
 
     block::process_block(
@@ -193,15 +195,24 @@ pub fn apply_block(
         active_balances,
         block::BlockProcessingOptions::default(),
     )?;
-    // As in `process_slot`: the post-state's root, checked below and needed
-    // again in the next slot, is then computed on the state's own nodes.
-    state.apply_pending_mutations();
 
+    // As in `process_slot`: the post-state's root, checked below and needed
+    // again in the next slot, is then computed on the state's own nodes. The
+    // flush and the hash are one phase: the flush is what makes the hash run
+    // on, and cache into, those nodes, and a caller that does not validate
+    // pays the flush alone without ever naming it, so it is not worth a series
+    // of its own.
     if validate_result {
+        let state_root_matches = timed(StfPhase::StateRoot, || {
+            state.apply_pending_mutations();
+            signed_block.state_root() == state.hash_tree_root()
+        });
         verify(
-            signed_block.state_root() == state.hash_tree_root(),
+            state_root_matches,
             "block state root matches the post-state",
         )?;
+    } else {
+        state.apply_pending_mutations();
     }
 
     Ok(())
@@ -228,12 +239,22 @@ pub fn verify_block_signature(
 pub fn process_slots(state: &mut BeaconState, slot: Slot, config: &Config) -> Result<()> {
     verify(state.slot() < slot, "target slot is after the current slot")?;
 
+    // The epoch work is timed on its own and taken out of this call's total,
+    // so the two phases add up to the call instead of the second containing
+    // the first.
+    let started = Instant::now();
+    let mut epoch_work = Duration::ZERO;
+
     while state.slot() < slot {
         process_slot(state)?;
         // Epoch processing runs on the last slot of an epoch, before the counter
         // moves into the next one.
         if (state.slot() + 1).is_multiple_of(preset::SLOTS_PER_EPOCH) {
+            let epoch_started = Instant::now();
             epoch::process_epoch(state, config)?;
+            let elapsed = epoch_started.elapsed();
+            observe_stf_phase(StfPhase::ProcessEpoch, elapsed);
+            epoch_work += elapsed;
         }
         *state.slot_mut() += 1;
         upgrade_at_fork_boundary(state, config)?;
@@ -247,6 +268,13 @@ pub fn process_slots(state: &mut BeaconState, slot: Slot, config: &Config) -> Re
     // pending.
     state.apply_pending_mutations();
 
+    // Only a call that got this far is observed: a rejected target slot or a
+    // failed epoch leaves no state worth attributing time to.
+    observe_stf_phase(
+        StfPhase::ProcessSlots,
+        started.elapsed().saturating_sub(epoch_work),
+    );
+
     Ok(())
 }
 
@@ -486,6 +514,39 @@ mod tests {
         assert!(!state.has_pending_mutations());
     }
 
+    /// The epoch work is observed once per boundary and under its own label,
+    /// and the plain slot work once per call, so the two never double count.
+    /// Which context they land in follows the thread's guard, which is what
+    /// keeps the precompute worker's `process_epoch` out of an import's.
+    #[test]
+    fn process_slots_observes_epoch_work_apart_from_slot_work() {
+        use crate::metrics::{import_context, take_observation_log};
+
+        let config = Config::mainnet();
+        let mut state = test_state::with_validators(4);
+        let two_epochs = state.slot() + 2 * preset::SLOTS_PER_EPOCH;
+
+        take_observation_log();
+        process_slots(&mut state, two_epochs, &config).unwrap();
+        assert_eq!(
+            take_observation_log(),
+            [
+                ("other", StfPhase::ProcessEpoch),
+                ("other", StfPhase::ProcessEpoch),
+                ("other", StfPhase::ProcessSlots),
+            ]
+        );
+
+        let _import = import_context();
+        let one_slot = state.slot() + 1;
+        process_slots(&mut state, one_slot, &config).unwrap();
+        assert_eq!(
+            take_observation_log(),
+            [("import", StfPhase::ProcessSlots)],
+            "a call that crosses no boundary runs no epoch work"
+        );
+    }
+
     #[test]
     fn process_slots_flushes_the_roots_it_wrote_in_its_last_slot() {
         for fork in BEACON_FORKS {
diff --git a/crates/blockchain/state_transition/src/beacon/stf/operations.rs b/crates/blockchain/state_transition/src/beacon/stf/operations.rs
index 5f9add7cc..8f7c86a18 100644
--- a/crates/blockchain/state_transition/src/beacon/stf/operations.rs
+++ b/crates/blockchain/state_transition/src/beacon/stf/operations.rs
@@ -39,6 +39,7 @@ use crate::beacon::helpers::predicates::{
 use crate::beacon::lean_state_unreachable;
 use crate::beacon::preset;
 use crate::beacon::primitives::{Epoch, Gwei, HashTreeRoot as _, ValidatorIndex};
+use crate::metrics::{StfPhase, try_for_each_timed};
 
 /// Runs every operation in a block, in the specification's order.
 ///
@@ -117,13 +118,17 @@ pub fn process_operations(
         "len(body.deposits) == min(MAX_DEPOSITS, eth1_data.deposit_count - eth1_deposit_index)",
     )?;
 
-    for proposer_slashing in proposer_slashings {
-        process_proposer_slashing(state, proposer_slashing, config)?;
-    }
-    for attester_slashing in attester_slashings {
-        process_attester_slashing(state, attester_slashing, config)?;
-    }
-    for attestation in attestations {
+    try_for_each_timed(
+        StfPhase::ProposerSlashings,
+        proposer_slashings,
+        |slashing| process_proposer_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(
+        StfPhase::AttesterSlashings,
+        attester_slashings,
+        |slashing| process_attester_slashing(state, slashing, config),
+    )?;
+    try_for_each_timed(StfPhase::Attestations, attestations, |attestation| {
         // Phase0 defers an attestation's reward to the epoch boundary, so it
         // needs its own version of this step; altair scores one the moment
         // it is processed instead, and nothing about that changed through
@@ -139,17 +144,17 @@ pub fn process_operations(
                 attestation,
                 committees,
                 active_balances,
-            )?;
+            )
         } else {
-            process_attestation(state, attestation, config, committees)?;
+            process_attestation(state, attestation, config, committees)
         }
-    }
-    for deposit in deposits {
-        process_deposit(state, deposit, config)?;
-    }
-    for voluntary_exit in voluntary_exits {
-        process_voluntary_exit(state, voluntary_exit, config)?;
-    }
+    })?;
+    try_for_each_timed(StfPhase::Deposits, deposits, |deposit| {
+        process_deposit(state, deposit, config)
+    })?;
+    try_for_each_timed(StfPhase::VoluntaryExits, voluntary_exits, |exit| {
+        process_voluntary_exit(state, exit, config)
+    })?;
 
     Ok(())
 }
diff --git a/crates/blockchain/state_transition/src/metrics.rs b/crates/blockchain/state_transition/src/metrics.rs
index 63c029166..c3a7dd2b9 100644
--- a/crates/blockchain/state_transition/src/metrics.rs
+++ b/crates/blockchain/state_transition/src/metrics.rs
@@ -162,6 +162,214 @@ pub fn inc_epoch_precompute_started(trigger: &str) {
         .inc();
 }
 
+// --- Beacon state transition phases ---
+
+/// Declares [`StfPhase`], its label, [`StfPhase::ALL`] and
+/// [`BEACON_STF_PHASES`] from one list, so a phase cannot exist in the code
+/// without being in the declared label set or the other way round.
+macro_rules! stf_phases {
+    ($($(#[$doc:meta])* $variant:ident => $label:literal,)+) => {
+        /// A step of the beacon state transition that
+        /// `lean_beacon_stf_phase_seconds` times.
+        #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+        pub enum StfPhase {
+            $($(#[$doc])* $variant,)+
+        }
+
+        impl StfPhase {
+            /// Every phase, in declaration order.
+            pub const ALL: &'static [StfPhase] = &[$(StfPhase::$variant,)+];
+
+            /// The `phase` label value this phase is observed under.
+            pub const fn label(self) -> &'static str {
+                match self {
+                    $(StfPhase::$variant => $label,)+
+                }
+            }
+        }
+
+        /// The `phase` label values of `lean_beacon_stf_phase_seconds`, in the
+        /// order a block crosses them. Generated from [`StfPhase`], so it is
+        /// the only list there is.
+        pub const BEACON_STF_PHASES: &[&str] = &[$($label,)+];
+    };
+}
+
+stf_phases! {
+    /// Advancing the state to the block's slot, everything but the epoch
+    /// work: per-slot root bookkeeping, the fork upgrade and the closing
+    /// flush.
+    ProcessSlots => "process_slots",
+    /// One observation per epoch boundary crossed (a long skip crosses
+    /// several).
+    ProcessEpoch => "process_epoch",
+    /// The proposer's signature over the block, when the caller asked for it.
+    BlockSignature => "block_signature",
+    BlockHeader => "block_header",
+    Randao => "randao",
+    Eth1Data => "eth1_data",
+    /// Withdrawals and the execution payload; gloas: the parent payload, the
+    /// withdrawals and the bid.
+    ExecutionPayload => "execution_payload",
+    ProposerSlashings => "proposer_slashings",
+    AttesterSlashings => "attester_slashings",
+    Attestations => "attestations",
+    Deposits => "deposits",
+    VoluntaryExits => "voluntary_exits",
+    BlsToExecutionChanges => "bls_to_execution_changes",
+    ExecutionRequests => "execution_requests",
+    PayloadAttestations => "payload_attestations",
+    SyncAggregate => "sync_aggregate",
+    /// The post-state flush and `hash_tree_root` that checks
+    /// `block.state_root`.
+    StateRoot => "state_root",
+}
+
+/// Which caller a beacon STF observation is charged to.
+///
+/// The epoch precompute worker runs `process_epoch` on another thread while an
+/// import is running, and the import replay reads this histogram before and
+/// after every block, so one global series would charge the worker's seconds
+/// to whichever block happened to be importing. Observations therefore carry
+/// the context of the thread that made them: only a thread inside an
+/// [`import_context`] guard writes `import`, and every other caller (the
+/// precompute worker, checkpoint-state advances, block production, gossip
+/// validation, the spec runners) writes `other`.
+///
+/// A thread-local rather than a parameter because the phases are observed deep
+/// inside the per-fork block processing, and the context is a property of who
+/// asked for the transition, which none of those functions know. Every
+/// observation is made on the calling thread, never inside a rayon task, so
+/// the thread-local is the right scope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum StfContext {
+    Import,
+    Other,
+}
+
+impl StfContext {
+    const fn label(self) -> &'static str {
+        match self {
+            StfContext::Import => STF_CONTEXT_IMPORT,
+            StfContext::Other => "other",
+        }
+    }
+}
+
+thread_local! {
+    static STF_CONTEXT: std::cell::Cell<StfContext> =
+        const { std::cell::Cell::new(StfContext::Other) };
+}
+
+/// The `context` label value of the block import path, for readers of the
+/// histogram.
+pub const STF_CONTEXT_IMPORT: &str = "import";
+
+/// Keeps the current thread's STF observations in the `import` series until
+/// dropped, then restores whatever context it replaced.
+#[must_use = "the guard is the scope of the import context"]
+pub struct ImportContextGuard {
+    previous: StfContext,
+}
+
+impl Drop for ImportContextGuard {
+    fn drop(&mut self) {
+        STF_CONTEXT.with(|context| context.set(self.previous));
+    }
+}
+
+/// Charge this thread's STF phase observations to block import until the
+/// returned guard is dropped. See [`StfContext`].
+pub fn import_context() -> ImportContextGuard {
+    let previous = STF_CONTEXT.with(|context| context.replace(StfContext::Import));
+    ImportContextGuard { previous }
+}
+
+static LEAN_BEACON_STF_PHASE_SECONDS: LazyLock<HistogramVec> = LazyLock::new(|| {
+    register_histogram_vec!(
+        "lean_beacon_stf_phase_seconds",
+        "Time one step of the beacon state transition took, per observation. `phase` is one of \
+         BEACON_STF_PHASES; `context` is `import` for the block import path and `other` for \
+         every other caller (epoch precompute, checkpoint states, block production, gossip \
+         validation). A step that did not run writes nothing, and a step that runs more than \
+         once in a transition (process_epoch across a long skip) writes once per run.",
+        &["context", "phase"],
+        vec![
+            0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 4.0
+        ]
+    )
+    .unwrap()
+});
+
+#[cfg(test)]
+thread_local! {
+    /// What this thread observed, as `(context, phase)`. The histogram is
+    /// process-global and the lib tests run in parallel, so a test reading
+    /// series deltas would count its neighbours' observations; each thread's
+    /// own log cannot be written by anyone else.
+    static OBSERVATION_LOG: std::cell::RefCell<Vec<(&'static str, StfPhase)>> =
+        const { std::cell::RefCell::new(Vec::new()) };
+}
+
+/// Drain this thread's observation log.
+#[cfg(test)]
+pub fn take_observation_log() -> Vec<(&'static str, StfPhase)> {
+    OBSERVATION_LOG.with(|log| std::mem::take(&mut *log.borrow_mut()))
+}
+
+/// Record `elapsed` against `phase` in the calling thread's context.
+pub fn observe_stf_phase(phase: StfPhase, elapsed: std::time::Duration) {
+    let context = STF_CONTEXT.with(|context| context.get());
+    #[cfg(test)]
+    OBSERVATION_LOG.with(|log| log.borrow_mut().push((context.label(), phase)));
+    LEAN_BEACON_STF_PHASE_SECONDS
+        .with_label_values(&[context.label(), phase.label()])
+        .observe(elapsed.as_secs_f64());
+}
+
+/// Run `step`, observing its duration as `phase`: one `Instant` pair, however
+/// long the step.
+///
+/// A step that fails inside `step` is observed too, since the closure returns
+/// before the observation. That is deliberate: a rejected block's cost is
+/// still cost.
+pub fn timed<T>(phase: StfPhase, step: impl FnOnce() -> T) -> T {
+    let started = std::time::Instant::now();
+    let result = step();
+    observe_stf_phase(phase, started.elapsed());
+    result
+}
+
+/// Run `step` over every item of `items`, stopping at the first error, and
+/// observe the whole loop as one `phase` observation.
+///
+/// An empty list observes nothing: a block with no deposits did not run the
+/// deposits step, which is a different statement from a step that took no
+/// time. The timer starts at the first item rather than before the iterator is
+/// built, so it does not charge the list's own construction.
+pub fn try_for_each_timed<I, E>(
+    phase: StfPhase,
+    items: I,
+    mut step: impl FnMut(I::Item) -> Result<(), E>,
+) -> Result<(), E>
+where
+    I: IntoIterator,
+{
+    let mut started = None;
+    let mut outcome = Ok(());
+    for item in items {
+        started.get_or_insert_with(std::time::Instant::now);
+        outcome = step(item);
+        if outcome.is_err() {
+            break;
+        }
+    }
+    if let Some(started) = started {
+        observe_stf_phase(phase, started.elapsed());
+    }
+    outcome
+}
+
 static LEAN_STATE_TRANSITION_TIME_SECONDS: LazyLock<Histogram> = LazyLock::new(|| {
     register_histogram!(
         "lean_state_transition_time_seconds",
@@ -244,4 +452,89 @@ pub fn time_data_column_kzg_verify() -> TimingGuard {
 /// observation.
 pub fn init() {
     LazyLock::force(&LEAN_DATA_COLUMN_KZG_VERIFY_SECONDS);
+    LazyLock::force(&LEAN_BEACON_STF_PHASE_SECONDS);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_label_list_is_the_enum() {
+        let labels: Vec<&str> = StfPhase::ALL.iter().map(|phase| phase.label()).collect();
+        assert_eq!(labels, BEACON_STF_PHASES);
+
+        let mut unique = labels.clone();
+        unique.sort_unstable();
+        unique.dedup();
+        assert_eq!(unique.len(), labels.len(), "a label is declared twice");
+    }
+
+    #[test]
+    fn only_a_thread_inside_the_guard_writes_the_import_series() {
+        take_observation_log();
+        let phase = StfPhase::ProposerSlashings;
+
+        timed(phase, || ());
+        {
+            let _guard = import_context();
+            timed(phase, || ());
+            // A thread spawned inside the guard (the precompute worker's
+            // situation) does not inherit it.
+            let spawned = std::thread::spawn(move || {
+                timed(phase, || ());
+                take_observation_log()
+            })
+            .join()
+            .unwrap();
+            assert_eq!(spawned, [("other", phase)]);
+        }
+        timed(phase, || ());
+
+        assert_eq!(
+            take_observation_log(),
+            [("other", phase), ("import", phase), ("other", phase)]
+        );
+    }
+
+    #[test]
+    fn a_nested_guard_restores_the_context_it_replaced() {
+        take_observation_log();
+        let phase = StfPhase::Deposits;
+        let outer = import_context();
+        drop(import_context());
+        timed(phase, || ());
+        drop(outer);
+        timed(phase, || ());
+        assert_eq!(
+            take_observation_log(),
+            [("import", phase), ("other", phase)]
+        );
+    }
+
+    #[test]
+    fn an_empty_list_observes_nothing_and_a_failure_stops_the_loop() {
+        take_observation_log();
+        let phase = StfPhase::VoluntaryExits;
+
+        let empty: [u8; 0] = [];
+        try_for_each_timed(phase, empty, |_| Ok::<(), ()>(())).unwrap();
+        assert!(
+            take_observation_log().is_empty(),
+            "an empty list ran no step"
+        );
+
+        let mut seen = 0;
+        let outcome = try_for_each_timed(phase, [1, 2, 3], |item| {
+            seen += 1;
+            if item == 2 { Err("second") } else { Ok(()) }
+        });
+        assert_eq!(outcome, Err("second"));
+        assert_eq!(seen, 2, "items after the failure are not run");
+        assert_eq!(
+            take_observation_log(),
+            [("other", phase)],
+            "one observation per loop"
+        );
+    }
 }
diff --git a/docs/benchmarking.md b/docs/benchmarking.md
index ae50db389..0f1ca5be4 100644
--- a/docs/benchmarking.md
+++ b/docs/benchmarking.md
@@ -277,6 +277,90 @@ is state-transition work and the two phases do not overlap. When blocks import
 faster than the writer persists them, the wait shows up here rather than in
 `stf`.
 
+### Where `stf` goes: the report's `stf_phases`, `class` and resource fields
+
+`stf` is one number for a block's whole state transition, and on a mainnet
+block it is most of the import. Each sample therefore also carries what it was
+made of, so a baseline says which step to attack and for which kind of block.
+The report's `schema_version` is 2; version 1 had none of what follows, and every
+field below that a sample or the summary can lack is **absent, not zero**.
+
+**`stf_phases`** (per sample, seconds). The steps of the beacon state
+transition, read from `lean_beacon_stf_phase_seconds` the same way `phases`
+reads the import histogram: its `sum` before and after each import. The names are
+`BEACON_STF_PHASES`, described in [`metrics.md`](metrics.md#beacon-state-transition-phases):
+`process_slots` (excluding epoch work), `process_epoch`, `block_signature`,
+`block_header`, `randao`, `eth1_data`, `execution_payload`, one per operation list
+(`proposer_slashings`, `attester_slashings`, `attestations`, `deposits`,
+`voluntary_exits`, `bls_to_execution_changes`, `execution_requests`,
+`payload_attestations`), `sync_aggregate` and `state_root`. Unlike `phases`, a
+step may be observed more than once in one import (`process_epoch` once per
+boundary a long skip crosses, the gloas payload steps twice), so a value is the
+**sum** of that import's observations, not a single one.
+
+Only the import path's observations are read. The epoch precompute worker runs
+`process_epoch` on another thread while an import runs, and the histogram has a
+`context` label (`import` or `other`) so the replay reads `context="import"`
+alone; see the metrics page for what sets each. A replay runs no precompute
+worker, so for these blocks the filter changes nothing; it is what keeps the
+same reading honest on any binary that does.
+
+For the same reason an `epoch_boundary` block in a replay always pays
+`process_epoch` inline: nothing has precomputed it (the report's
+`caches.epoch_precompute` shows the misses). On a live follower a precompute hit
+takes that cost off the import path, so the replay is the upper bound.
+
+**`stf_unattributed_seconds`** is `stf` minus the sum of `stf_phases`: the part
+of the transition no step covers, such as cloning the parent's state and
+building the epoch-boundary balances. Absent when `stf` did not run.
+
+**`class`** is what the block's `process_slots` had to do, decided from the
+block's slot and its parent's (`parent_slot`, also in the sample):
+
+| `class` | Meaning |
+|---------|---------|
+| `epoch_boundary` | The block's slot is in a later epoch than its parent's, so `process_epoch` runs. Takes precedence over a skip |
+| `after_skip` | Same epoch, but the parent is more than one slot back |
+| `ordinary` | The parent is the previous slot, in the same epoch |
+
+A corpus is linear (`fetch` refuses a block whose parent is not the one before
+it), so there is no class for a reorg: a block is always applied to the block
+imported just before it. The class describes the work the block asked for, not
+what a cache spared it.
+
+**`cpu_seconds`** is the user plus system CPU time the whole process used during
+the import (`getrusage(RUSAGE_SELF)`, delta around the call). It is the process,
+not the importing thread: the merkle fold's workers, the state writer and
+RocksDB's background threads are in it, which is the point (an import whose CPU
+time is well above its wall time ran in parallel) and also the caveat (nothing
+else may run in the process). **`rss_bytes`** is the resident set size right
+after the import, read from `/proc/self/status`, so it is **Linux only** and
+absent elsewhere. **`peak_rss_bytes`** is the process's high-water mark so far
+(`ru_maxrss`), present on Linux and macOS; being a running maximum it moves only
+on the block that set a new peak.
+
+**Summary.** `summary` keeps its old paths (`phases`, `wall`) and gains
+`stf_phases`, `stf_unattributed` and `cpu`, each a distribution over the blocks
+that have the field, and `by_class`: for each class at least one block fell into
+(`epoch_boundary`, `after_skip`, `ordinary`) its `blocks` count and the same
+distributions (`phases`, `stf_phases`, `stf_unattributed`, `wall`, `cpu`)
+restricted to that class. Every distribution (`Stats`) now has `p95_seconds` and
+`p99_seconds` beside `p50_seconds` and `p90_seconds`, in both workloads' reports.
+Percentiles are nearest-rank, so a short run reports its maximum for the tail:
+the p99 is the maximum up to 51 samples, the p95 up to 11.
+
+**`caches`** (per run, not per block) is the lookups the state transition's
+caches served during the measured blocks, as deltas of the counters in
+[`metrics.md`](metrics.md) taken after the warm-up so each cache's cold start is
+not counted: `epoch_precompute`, `committee_cache`, `total_active_balance` and
+`justified_balances`, each with `hit`, `miss`, `hit_rate` (`hit / (hit + miss)`,
+absent when nothing was looked up) and, for the two caches that have the outcome,
+`unkeyable` (computed for one caller and not cached, so not in the rate).
+
+The human table shows the per-block `phases`, the class, CPU and unattributed
+time, then the summaries; the per-block `stf_phases` are in the JSON only,
+since close to twenty more columns would not fit a terminal.
+
 ### What is excluded, and why
 
 Replay runs nothing that a corpus already answers for:
diff --git a/docs/metrics.md b/docs/metrics.md
index 9ac92fd4f..86dbc0d68 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -559,6 +559,42 @@ block raced it, see `lean_beacon_epoch_precompute_seconds` against the slot
 duration), the node was syncing, or the entry was evicted from the state cache.
 A `timer` start with no matching `head` start means the last slot was skipped.
 
+### Beacon State Transition Phases
+
+Where the time of one beacon state transition goes. `lean_block_import_phase_seconds{phase="stf"}`
+says how long an import's transition took; this says which step.
+
+| Name | Type | Usage | Sample collection event | Labels | Buckets |
+|------|------|-------|-------------------------|--------|---------|
+| `lean_beacon_stf_phase_seconds` | Histogram | Time one step of the beacon state transition took | Once per step per transition, when the step finishes (a step that did not run writes nothing) | context=import,other; phase (see below) | 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 4 |
+
+`phase` is one of `BEACON_STF_PHASES` (`crates/blockchain/state_transition/src/metrics.rs`):
+
+| Phase | Covers |
+|-------|--------|
+| `process_slots` | Advancing the state to the block's slot, **excluding** epoch work: per-slot root bookkeeping, fork upgrades, the closing flush |
+| `process_epoch` | The epoch transition, once per boundary crossed (a long skip observes it several times) |
+| `block_signature` | The proposer's signature over the block, when the caller asked for result validation |
+| `block_header` | The header check, including the hash of the block body it commits to |
+| `randao` | The reveal's verification and the mix update |
+| `eth1_data` | The eth1 vote |
+| `execution_payload` | Withdrawals and the execution payload; gloas: the parent payload, then the withdrawals and the bid, two observations per block |
+| `proposer_slashings`, `attester_slashings`, `attestations`, `deposits`, `voluntary_exits`, `bls_to_execution_changes`, `payload_attestations` | One observation per block for the whole list, absent when the list is empty; the fork must have the list |
+| `execution_requests` | Electra and fulu's requests of the execution layer |
+| `sync_aggregate` | The sync aggregate, altair onward |
+| `state_root` | The post-state flush and `hash_tree_root` that checks `block.state_root`; only observed when the caller validates the result |
+
+**`context` keeps background work out of an import's numbers.** The epoch
+precompute worker runs `process_epoch` on its own thread while an import runs,
+so one series would charge the worker's seconds to whichever import was in
+flight. `context="import"` is written only by the import path
+(`fork_choice::on_block`'s transition); every other caller (the precompute
+worker, checkpoint-state advances, block production, gossip validation) writes
+`context="other"`. A dashboard that wants an import's cost reads
+`context="import"`; one that wants the precompute worker's reads `other`, which
+also holds the other callers, so `lean_beacon_epoch_precompute_seconds` is the
+cleaner series for the worker alone.
+
 ### Beacon Pubkey Cache
 
 Every BLS signature check `ethlambda beacon` runs (block import, fork choice,