obzenflow_runtime 0.2.5

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

//! Latest observation maps and registration of observation producers.
//! This module has no journal, publication scope, credit, or settlement capability.

use crate::execution::RuntimeExecution;
use obzenflow_core::event::observability::families::{
    any_observation_family, observation_families as split, ObservationFamily,
};
#[cfg(test)]
use obzenflow_core::event::observability::RuntimeObservability;
use obzenflow_core::event::observability::*;
use obzenflow_core::{FlowId, MiddlewareExecutionScope, WriterId};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::{SystemTime, UNIX_EPOCH};

const MAX_KEYS: usize = 4096;
const MAX_OWNERS: usize = 1024;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct Key {
    observer: WriterId,
    kind: ObservationFamily,
}

#[derive(Debug, Clone, Default)]
struct View {
    active_scope: Option<CaptureScope>,
    latest: HashMap<Key, ObservabilityContext>,
}

impl View {
    fn can_replace(&self, key: &Key, capture: CaptureStamp, recorded: bool) -> bool {
        let scope = capture.capture_scope;
        if let Some(active) = self.active_scope {
            if (!recorded && scope != active)
                || (scope.flow_id == active.flow_id
                    && scope.resume_generation > active.resume_generation)
            {
                return false;
            }
        }
        if let Some(previous) = self.latest.get(key) {
            if previous.capture.capture_scope == scope {
                return previous.capture.capture_seq < capture.capture_seq;
            }
            if self.active_scope != Some(scope) {
                let previous_scope = previous.capture.capture_scope;
                return recorded
                    && previous_scope.flow_id == scope.flow_id
                    && previous_scope.resume_generation < scope.resume_generation;
            }
        }
        true
    }
}

/// A bounded, mutex-protected hash map of the latest observation per observer
/// and measurement family/subject. A newer capture replaces that key's value;
/// sequence gaps are allowed and reading the map does not consume its entries.
#[derive(Debug, Default)]
pub struct LatestObservationMap {
    view: Mutex<View>,
    dropped: AtomicU64,
}

/// Run-owned capture-sequence and stage-instrumentation registries, together
/// with the latest observations offered by those producers.
#[derive(Debug, Default)]
pub struct ObservationRegistry {
    latest: LatestObservationMap,
    capture_sequences: Mutex<HashMap<(CaptureScope, WriterId), Arc<AtomicU64>>>,
    stages: Mutex<HashMap<WriterId, Weak<super::instrumentation::StageInstrumentation>>>,
}

impl ObservationRegistry {
    pub fn latest(&self) -> &LatestObservationMap {
        &self.latest
    }

    pub(crate) fn live_counters(&self) -> HashMap<obzenflow_core::StageId, (CaptureScope, u64)> {
        let stages: Vec<_> = self
            .stages
            .try_lock()
            .map(|stages| {
                stages
                    .iter()
                    .map(|(writer, stage)| (*writer, stage.clone()))
                    .collect()
            })
            .unwrap_or_default();
        stages
            .into_iter()
            .filter_map(|(writer, weak)| {
                let stage_id = *writer.as_stage()?;
                Some((stage_id, weak.upgrade()?.live_counter_sample()?))
            })
            .collect()
    }

    pub fn activate_scope(&self, scope: CaptureScope) {
        self.latest.activate_scope(scope);
    }

    pub(crate) fn register_stage(
        &self,
        writer: WriterId,
        stage: &Arc<super::instrumentation::StageInstrumentation>,
    ) {
        if let Ok(mut stages) = self.stages.lock() {
            if stages.len() < MAX_OWNERS || stages.contains_key(&writer) {
                stages.insert(writer, Arc::downgrade(stage));
            }
        }
    }

    /// Explicit live-only capture. Journal-derived exports do not call this;
    /// it neither records an attachment nor consumes a journal allowance.
    pub fn capture_registered(&self, reason: CaptureReason) {
        let stages: Vec<_> = self
            .stages
            .try_lock()
            .map(|stages| stages.values().cloned().collect())
            .unwrap_or_default();
        for stage in stages.into_iter().filter_map(|stage| stage.upgrade()) {
            stage.offer_capture(reason);
        }
    }

    pub(crate) fn capture_owner(
        self: &Arc<Self>,
        scope: CaptureScope,
        observer: WriterId,
        execution: RuntimeExecution,
    ) -> ObservationOwner {
        let sequence = self
            .capture_sequences
            .lock()
            .ok()
            .and_then(|mut sequences| {
                if !sequences.contains_key(&(scope, observer)) && sequences.len() >= MAX_OWNERS {
                    return None;
                }
                Some(sequences.entry((scope, observer)).or_default().clone())
            });
        ObservationOwner {
            scope,
            observer,
            sequence,
            registry: self.clone(),
            execution,
        }
    }
}

impl ObservationSink for ObservationRegistry {
    fn offer(&self, observation: ObservabilityContext) -> ObservationOffer {
        self.latest.offer(observation)
    }
}

impl ObservationSource for ObservationRegistry {
    fn active_scope(&self) -> Option<CaptureScope> {
        self.latest.active_scope()
    }

    fn snapshot(&self) -> Vec<ObservabilityContext> {
        self.latest.snapshot()
    }
}

impl LatestObservationMap {
    pub fn activate_scope(&self, scope: CaptureScope) {
        if let Ok(mut view) = self.view.lock() {
            view.active_scope = Some(scope);
        }
    }

    fn drop_sample(&self) -> ObservationOffer {
        self.dropped.fetch_add(1, Ordering::Relaxed);
        ObservationOffer::Dropped
    }

    /// Restore only evidence already admitted through a consumer's journal cut.
    /// Recorded attachments never activate an execution generation.
    pub(crate) fn offer_recorded(&self, packet: &ObservabilityContext) {
        if self.could_update(packet, true).unwrap_or(false) {
            let _ = self.select_inner(packet.clone(), true, false);
        }
    }

    fn could_update(
        &self,
        packet: &ObservabilityContext,
        recorded: bool,
    ) -> Result<bool, ObservationOffer> {
        let view = self.view.try_lock().map_err(|_| self.drop_sample())?;
        Ok(any_observation_family(packet, |kind, capture| {
            view.can_replace(
                &Key {
                    observer: capture.observer,
                    kind,
                },
                capture,
                recorded,
            )
        }))
    }

    /// An independent copy of the retained values and active scope.
    pub fn retained_copy(&self) -> Self {
        let view = self
            .view
            .try_lock()
            .map(|view| view.clone())
            .unwrap_or_default();
        Self {
            view: Mutex::new(view),
            ..Default::default()
        }
    }

    /// Select the changed families for an existing backend consumer. All
    /// callers use the same scope and per-subject ordering rule.
    pub fn select(
        &self,
        observation: ObservabilityContext,
    ) -> Result<Vec<ObservabilityContext>, ObservationOffer> {
        self.select_inner(observation, false, true)
    }

    pub fn select_recorded(
        &self,
        observation: ObservabilityContext,
    ) -> Result<Vec<ObservabilityContext>, ObservationOffer> {
        self.select_inner(observation, true, true)
    }

    fn select_inner(
        &self,
        observation: ObservabilityContext,
        recorded: bool,
        include_deltas: bool,
    ) -> Result<Vec<ObservabilityContext>, ObservationOffer> {
        let Some(observation) = observation.validated() else {
            return Err(self.drop_sample());
        };
        let Some(families) = split(observation) else {
            return Err(self.drop_sample());
        };
        let Ok(mut view) = self.view.try_lock() else {
            return Err(self.drop_sample());
        };
        let mut selected = Vec::new();
        let mut retained = false;
        let mut capacity_dropped = false;
        for (kind, packet) in families {
            let key = Key {
                observer: packet.capture.observer,
                kind,
            };
            // Validation runs outside the lock. Recheck against concurrent live captures.
            if !view.can_replace(&key, packet.capture, recorded) {
                continue;
            }
            if !view.latest.contains_key(&key) && view.latest.len() >= MAX_KEYS {
                self.drop_sample();
                capacity_dropped = true;
                continue;
            }
            if include_deltas {
                selected.push(packet.clone());
            }
            retained = true;
            view.latest.insert(key, packet);
        }
        if !retained && capacity_dropped {
            Err(ObservationOffer::Dropped)
        } else {
            Ok(selected)
        }
    }
}

impl ObservationSink for LatestObservationMap {
    fn offer(&self, observation: ObservabilityContext) -> ObservationOffer {
        match self.could_update(&observation, false) {
            Ok(false) => return ObservationOffer::Accepted,
            Err(dropped) => return dropped,
            Ok(true) => {}
        }
        match self.select_inner(observation, false, false) {
            Ok(_) => ObservationOffer::Accepted,
            Err(dropped) => dropped,
        }
    }
}

impl ObservationSource for LatestObservationMap {
    fn active_scope(&self) -> Option<CaptureScope> {
        match self.view.try_lock() {
            Ok(view) => view.active_scope,
            Err(_) => None,
        }
    }
    fn snapshot(&self) -> Vec<ObservabilityContext> {
        let mut packets: Vec<_> = match self.view.try_lock() {
            Ok(view) => view.latest.values().cloned().collect(),
            Err(_) => return Vec::new(),
        };
        // Families are independently retained; overlapping projected fields
        // apply in capture order, never HashMap iteration order.
        packets.sort_by_key(|packet| {
            (
                packet.capture.capture_scope.resume_generation,
                packet.capture.capture_seq,
            )
        });
        packets
    }
}

/// Sequence ownership survives capture-helper recreation and spans the owner's
/// data, error, and system records. Allocating it never allocates an EventId.
#[derive(Debug, Clone)]
pub struct ObservationOwner {
    scope: CaptureScope,
    observer: WriterId,
    sequence: Option<Arc<AtomicU64>>,
    registry: Arc<ObservationRegistry>,
    execution: RuntimeExecution,
}

impl ObservationOwner {
    pub(crate) fn scope(&self) -> CaptureScope {
        self.scope
    }

    pub fn measurements_allowed(&self) -> bool {
        match self.observer.as_stage() {
            Some(stage) => !self.execution.stage_scope(*stage).is_deterministic_replay(),
            None => self.execution.host_observations_allowed(),
        }
    }

    pub fn capture(&self, reason: CaptureReason) -> Option<ObservabilityContext> {
        if !self.measurements_allowed() {
            return None;
        }
        self.allocate_capture(reason)
    }

    pub(crate) fn capture_in_scope(
        &self,
        reason: CaptureReason,
        scope: MiddlewareExecutionScope,
    ) -> Option<ObservabilityContext> {
        if scope.is_deterministic_replay() || !self.execution.host_observations_allowed() {
            return None;
        }
        self.allocate_capture(reason)
    }

    fn allocate_capture(&self, reason: CaptureReason) -> Option<ObservabilityContext> {
        let sequence = self
            .sequence
            .as_ref()?
            .fetch_update(
                Ordering::Relaxed,
                Ordering::Relaxed,
                |current_capture_seq| current_capture_seq.checked_add(1),
            )
            .ok()?
            + 1;
        Some(ObservabilityContext::new(CaptureStamp {
            capture_scope: self.scope,
            observer: self.observer,
            capture_seq: CaptureSeq(sequence),
            capture_reason: reason,
            observed_at_ms: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .ok()?
                .as_millis() as u64,
        }))
    }

    pub fn offer(&self, packet: ObservabilityContext) {
        self.registry.offer(packet);
    }
}

impl ObservationRecorder for ObservationOwner {
    fn observe(&self, record: ObservationRecord) {
        self.observe_with_reason(record, CaptureReason::Record);
    }
    fn observe_with_reason(&self, record: ObservationRecord, reason: CaptureReason) {
        if let Some(mut packet) = self.capture(reason) {
            packet.records.push(record);
            self.offer(packet);
        }
    }
}

pub(crate) fn scope(execution: &RuntimeExecution, flow_id: FlowId) -> CaptureScope {
    CaptureScope {
        flow_id,
        resume_generation: execution
            .resume_control()
            .map(|control| control.resume_generation())
            .unwrap_or_default(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::execution::RuntimeMode;
    use crate::metrics::instrumentation::StageInstrumentation;
    use obzenflow_core::event::observability::{MeasurementWindow, TimingMeasurements};
    use obzenflow_core::event::ChainEventFactory;
    use obzenflow_core::{ReaderGeneration, StageId};
    use std::time::Duration;

    fn packet(
        scope: CaptureScope,
        observer: WriterId,
        sequence: u64,
        in_flight: u32,
    ) -> ObservabilityContext {
        let mut packet = ObservabilityContext::new(CaptureStamp {
            capture_scope: scope,
            observer,
            capture_seq: CaptureSeq(sequence),
            capture_reason: CaptureReason::Record,
            observed_at_ms: sequence,
        });
        packet.runtime = Some(RuntimeObservability {
            in_flight: Some(in_flight),
            ..Default::default()
        });
        packet
    }

    #[test]
    fn retained_candidates_use_family_stamps_and_validate_the_whole_packet() {
        use obzenflow_core::event::observability::{ExecutionProgress, RuntimeSnapshot};

        let observations = LatestObservationMap::default();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        let writer = StageId::new().into();
        let local = StageId::new().into();
        observations.activate_scope(scope);
        observations.offer_recorded(&packet(scope, writer, 100, 4));

        // The outer family is old, but the independently stamped nested family is new.
        let mut carrier = packet(scope, writer, 99, 9);
        carrier.runtime_snapshot = Some(RuntimeSnapshot {
            capture: packet(scope, local, 7, 0).capture,
            progress: ExecutionProgress::default(),
            fsm_state: "Running".into(),
        });
        observations.offer_recorded(&carrier);
        let retained = observations.snapshot();
        assert_eq!(retained.len(), 2);
        assert_eq!(
            retained.iter().find_map(|p| p.runtime.as_ref()?.in_flight),
            Some(4)
        );
        assert_eq!(
            retained
                .iter()
                .find_map(|p| p.runtime_snapshot.as_ref())
                .unwrap()
                .capture
                .capture_seq,
            CaptureSeq(7)
        );

        // Even a stale family must be validated when any part of its original
        // packet is a candidate. Splitting first would wrongly admit in_flight.
        carrier.capture.capture_seq = CaptureSeq(101);
        carrier.runtime_snapshot.as_mut().unwrap().fsm_state = "x".repeat(65_536);
        observations.offer_recorded(&carrier);
        assert_eq!(observations.dropped.load(Ordering::Relaxed), 1);
        assert_eq!(
            serde_json::to_value(observations.snapshot()).unwrap(),
            serde_json::to_value(&retained).unwrap()
        );

        // Entirely superseded optional measurements need no admission validation.
        carrier.capture.capture_seq = CaptureSeq(98);
        observations.offer_recorded(&carrier);
        assert_eq!(observations.dropped.load(Ordering::Relaxed), 1);
        assert_eq!(
            serde_json::to_value(observations.snapshot()).unwrap(),
            serde_json::to_value(retained).unwrap()
        );
    }

    #[test]
    fn runtime_snapshot_uses_its_own_stamp_and_replaces_the_whole_family() {
        use obzenflow_core::event::observability::{ExecutionProgress, RuntimeSnapshot};

        let observations = LatestObservationMap::default();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        observations.activate_scope(scope);
        let upstream = StageId::new().into();
        let local = StageId::new().into();
        let mut carrier = packet(scope, upstream, 100, 9);
        let local_stamp = packet(scope, local, 5, 0).capture;
        carrier.runtime_snapshot = Some(RuntimeSnapshot {
            capture: local_stamp,
            progress: ExecutionProgress {
                reader_seq: 12,
                last_consumed_event_id: Some(obzenflow_core::EventId::new()),
                ..Default::default()
            },
            fsm_state: "Running".into(),
        });
        assert_eq!(
            observations.select_recorded(carrier.clone()).unwrap().len(),
            2
        );
        let first = observations
            .snapshot()
            .into_iter()
            .find_map(|packet| packet.runtime_snapshot)
            .unwrap();
        assert_eq!(first.capture, local_stamp);
        assert_eq!(first.progress.reader_seq, 12);

        carrier.capture.capture_seq = CaptureSeq(101);
        carrier
            .runtime_snapshot
            .as_mut()
            .unwrap()
            .capture
            .capture_seq = CaptureSeq(4);
        carrier.runtime_snapshot.as_mut().unwrap().fsm_state = "Created".into();
        let selected = observations.select_recorded(carrier).unwrap();
        assert_eq!(selected.len(), 1);
        assert!(selected[0].runtime_snapshot.is_none());

        let mut newer = ObservabilityContext::new(local_stamp);
        newer.runtime_snapshot = Some(RuntimeSnapshot {
            capture: CaptureStamp {
                capture_seq: CaptureSeq(6),
                ..local_stamp
            },
            progress: ExecutionProgress::default(),
            fsm_state: "Drained".into(),
        });
        assert_eq!(
            observations.select_recorded(newer.clone()).unwrap().len(),
            1
        );
        assert!(observations
            .select_recorded(newer.clone())
            .unwrap()
            .is_empty());
        let latest = observations
            .snapshot()
            .into_iter()
            .find_map(|packet| packet.runtime_snapshot)
            .unwrap();
        assert_eq!(latest.capture.capture_seq, CaptureSeq(6));
        assert_eq!(latest.progress.reader_seq, 0);
        assert!(latest.progress.last_consumed_event_id.is_none());
        assert_eq!(latest.fsm_state, "Drained");

        let resumed = CaptureScope {
            resume_generation: ReaderGeneration(1),
            ..scope
        };
        observations.activate_scope(resumed);
        assert!(observations.select(newer.clone()).unwrap().is_empty());
        let snapshot = newer.runtime_snapshot.as_mut().unwrap();
        snapshot.capture.capture_scope = resumed;
        snapshot.capture.capture_seq = CaptureSeq(1);
        assert_eq!(observations.select(newer).unwrap().len(), 1);
        let latest = observations
            .snapshot()
            .into_iter()
            .find_map(|packet| packet.runtime_snapshot)
            .unwrap();
        assert_eq!(latest.capture.capture_scope, resumed);
        assert_eq!(latest.capture.capture_seq, CaptureSeq(1));
    }

    #[test]
    fn owners_share_sequence_across_helper_recreation_and_drop_on_contention() {
        let execution = RuntimeExecution::new(RuntimeMode::Live, None);
        let observations = execution.observations();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        let writer = WriterId::from(StageId::new());
        let first = observations.capture_owner(scope, writer, execution.clone());
        let restarted = observations.capture_owner(scope, writer, execution.clone());
        assert_eq!(
            first
                .capture(CaptureReason::Initial)
                .unwrap()
                .capture
                .capture_seq,
            CaptureSeq(1)
        );
        assert_eq!(
            restarted
                .capture(CaptureReason::Periodic)
                .unwrap()
                .capture
                .capture_seq,
            CaptureSeq(2)
        );
        let held = observations.latest.view.lock().unwrap();
        assert_eq!(
            observations.offer(packet(scope, writer, 3, 4)),
            ObservationOffer::Dropped
        );
        drop(held);
        assert!(observations.snapshot().is_empty());
        assert_eq!(observations.latest.dropped.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn family_freshness_and_active_generation_are_independent_of_arrival_order() {
        let observations = LatestObservationMap::default();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        let writer = WriterId::from(StageId::new());
        observations.activate_scope(scope);
        assert_eq!(
            observations
                .select(packet(scope, writer, 10, 4))
                .unwrap()
                .len(),
            1
        );
        assert!(observations
            .select(packet(scope, writer, 9, 9))
            .unwrap()
            .is_empty());
        assert!(observations
            .select(packet(scope, writer, 10, 9))
            .unwrap()
            .is_empty());
        let mut timing = packet(scope, writer, 3, 0);
        timing.runtime.as_mut().unwrap().in_flight = None;
        timing.runtime.as_mut().unwrap().timing = Some(TimingMeasurements {
            processing_time_count: 0,
            processing_time_sum_nanos: 0,
            recent_p50_ms: None,
            recent_p90_ms: None,
            recent_p95_ms: None,
            recent_p99_ms: None,
            recent_p999_ms: None,
            window: MeasurementWindow {
                started_at_ms: 0,
                ended_at_ms: 3,
            },
        });
        assert_eq!(observations.select(timing).unwrap().len(), 1);
        assert_eq!(observations.snapshot().len(), 2);
        let resumed = CaptureScope {
            resume_generation: ReaderGeneration(1),
            ..scope
        };
        assert!(
            observations
                .select(packet(resumed, StageId::new().into(), 1, 2))
                .unwrap()
                .is_empty(),
            "an unknown owner cannot activate a future generation"
        );
        assert!(observations
            .select(packet(resumed, writer, 1, 2))
            .unwrap()
            .is_empty());
        observations.activate_scope(resumed);
        assert_eq!(
            observations
                .select(packet(resumed, writer, 1, 2))
                .unwrap()
                .len(),
            1
        );
        assert!(observations
            .select(packet(scope, writer, 100, 9))
            .unwrap()
            .is_empty());
        assert_eq!(
            observations
                .snapshot()
                .iter()
                .filter_map(|packet| packet.runtime.as_ref()?.in_flight)
                .collect::<Vec<_>>(),
            vec![2]
        );
        assert!(
            observations
                .snapshot()
                .iter()
                .any(|packet| packet.capture.capture_scope == scope
                    && packet.runtime.as_ref().unwrap().timing.is_some()),
            "an absent family retains the recorded sample identity"
        );
    }

    #[test]
    fn recorded_generations_restore_without_activating_execution_or_regressing() {
        let observations = LatestObservationMap::default();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        let resumed = CaptureScope {
            resume_generation: ReaderGeneration(1),
            ..scope
        };
        let writer = StageId::new().into();
        observations.offer_recorded(&packet(resumed, writer, 1, 2));
        observations.offer_recorded(&packet(scope, writer, 1000, 9));
        assert_eq!(observations.active_scope(), None);
        assert_eq!(observations.snapshot()[0].capture.capture_scope, resumed);
        assert_eq!(observations.snapshot()[0].capture.observed_at_ms, 1);
        assert_eq!(
            observations.snapshot()[0]
                .runtime
                .as_ref()
                .unwrap()
                .in_flight,
            Some(2)
        );
    }

    #[test]
    fn capacity_and_unavailable_final_capture_drop_only_optional_samples() {
        let execution = RuntimeExecution::new(RuntimeMode::Live, None);
        let observations = execution.observations();
        let scope = CaptureScope {
            flow_id: FlowId::new(),
            resume_generation: ReaderGeneration(0),
        };
        observations.activate_scope(scope);
        for _ in 0..MAX_KEYS {
            assert_eq!(
                observations.offer(packet(scope, StageId::new().into(), 1, 0)),
                ObservationOffer::Accepted
            );
        }
        assert_eq!(
            observations.offer(packet(scope, StageId::new().into(), 1, 0)),
            ObservationOffer::Dropped
        );
        assert_eq!(observations.snapshot().len(), MAX_KEYS);
        for _ in 0..MAX_OWNERS {
            assert!(observations
                .capture_owner(scope, StageId::new().into(), execution.clone())
                .capture(CaptureReason::Initial)
                .is_some());
        }
        let stage = StageId::new();
        let instrumentation = Arc::new(StageInstrumentation::new());
        instrumentation.bind_observations(scope.flow_id, stage.into(), &execution);
        instrumentation.record_output_event(&ChainEventFactory::data_event(
            stage.into(),
            "business.fact",
            serde_json::Value::Null,
        ));
        let before = instrumentation.snapshot();
        assert!(instrumentation
            .capture_observability(CaptureReason::Final)
            .is_none());
        assert_eq!(
            serde_json::to_value(instrumentation.snapshot()).unwrap(),
            serde_json::to_value(before).unwrap()
        );
        assert_eq!(observations.snapshot().len(), MAX_KEYS);
    }

    #[test]
    fn timing_contention_omits_that_family_and_zero_duration_is_a_measurement() {
        let execution = RuntimeExecution::new(RuntimeMode::Live, None);
        let instrumentation = Arc::new(StageInstrumentation::new());
        instrumentation.bind_observations(FlowId::new(), StageId::new().into(), &execution);
        instrumentation.record_processing_time(Duration::ZERO);
        let captured = instrumentation
            .capture_observability(CaptureReason::Record)
            .unwrap();
        let timing = captured.runtime.unwrap().timing.unwrap();
        assert_eq!(timing.processing_time_count, 1);
        assert_eq!(timing.processing_time_sum_nanos, 0);
        assert_eq!(timing.recent_p50_ms, Some(0));
        let held = instrumentation.processing_time_histogram.write().unwrap();
        let partial = instrumentation
            .capture_observability(CaptureReason::Periodic)
            .unwrap()
            .runtime
            .unwrap();
        assert!(partial.timing.is_none());
        assert_eq!(partial.in_flight, Some(0));
        drop(held);
    }

    #[test]
    fn strict_replay_does_not_capture_new_measurements() {
        let execution = RuntimeExecution::new(RuntimeMode::Replay, None);
        let instrumentation = Arc::new(StageInstrumentation::new());
        instrumentation.bind_observations(FlowId::new(), StageId::new().into(), &execution);
        instrumentation.record_processing_time(Duration::from_millis(5));
        assert!(instrumentation
            .capture_observability(CaptureReason::Final)
            .is_none());
        assert!(execution.observations().snapshot().is_empty());
    }
}