obzenflow_runtime 0.1.2

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

//! Metrics aggregator FSM types and state machine definition
//!
//! The metrics aggregator follows a simple lifecycle:
//! Initializing -> Running -> Draining -> Drained
//! Event processing happens directly without FSM state tracking

use obzenflow_core::event::chain_event::ChainEventContent;
use obzenflow_core::event::context::StageType;
use obzenflow_core::event::ingestion::IngestionTelemetrySnapshot;
use obzenflow_core::event::observability::HttpPullTelemetry;
use obzenflow_core::event::payloads::observability_payload::{
    CircuitBreakerEvent, MetricsLifecycle, MiddlewareLifecycle, ObservabilityPayload,
};
use obzenflow_core::event::status::processing_status::{ErrorKind, ProcessingStatus};
use obzenflow_core::event::{JournalEvent, WriterId};
use obzenflow_core::id::{FlowId, StageId, SystemId};
use obzenflow_core::metrics::{ContractMetricsSnapshot, Percentile, StageMetadata};
use obzenflow_core::time::MetricsDuration;
use obzenflow_core::{ChainEvent, EventId, Journal};
use obzenflow_fsm::{
    fsm, EventVariant, FsmAction, FsmContext, StateMachine, StateVariant, Transition,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;

use crate::metrics::tail_read;

/// FSM states for metrics aggregator lifecycle
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum MetricsAggregatorState {
    /// Initial state
    Initializing,

    /// Normal operation - processing events
    Running,

    /// Processing final events before shutdown
    Draining,

    /// Terminal state - all events processed
    Drained { last_event_id: Option<EventId> },

    /// Terminal state - error occurred
    Failed { error: String },
}

impl StateVariant for MetricsAggregatorState {
    fn variant_name(&self) -> &str {
        match self {
            MetricsAggregatorState::Initializing => "Initializing",
            MetricsAggregatorState::Running => "Running",
            MetricsAggregatorState::Draining => "Draining",
            MetricsAggregatorState::Drained { .. } => "Drained",
            MetricsAggregatorState::Failed { .. } => "Failed",
        }
    }
}

/// Events that drive state transitions
#[derive(Clone, Debug)]
pub enum MetricsAggregatorEvent {
    /// Initialization complete, start processing
    StartRunning,

    /// Process a batch of events
    ProcessBatch {
        events: Vec<obzenflow_core::EventEnvelope<obzenflow_core::ChainEvent>>,
    },

    /// Process a system event (FLOWIP-059b)
    ProcessSystemEvent {
        envelope: Box<obzenflow_core::EventEnvelope<obzenflow_core::event::SystemEvent>>,
    },

    /// Time to export metrics
    ExportMetrics,

    /// Start draining process (from journal control event)
    StartDraining,

    /// Flow + stages have reached terminal lifecycle; perform final export and shutdown
    FlowTerminal,

    /// Error occurred (e.g., journal corruption)
    Error(String),
}

impl EventVariant for MetricsAggregatorEvent {
    fn variant_name(&self) -> &str {
        match self {
            MetricsAggregatorEvent::StartRunning => "StartRunning",
            MetricsAggregatorEvent::ProcessBatch { .. } => "ProcessBatch",
            MetricsAggregatorEvent::ProcessSystemEvent { .. } => "ProcessSystemEvent",
            MetricsAggregatorEvent::ExportMetrics => "ExportMetrics",
            MetricsAggregatorEvent::StartDraining => "StartDraining",
            MetricsAggregatorEvent::FlowTerminal => "FlowTerminal",
            MetricsAggregatorEvent::Error(_) => "Error",
        }
    }
}

/// Actions performed during transitions
#[derive(Clone, Debug)]
pub enum MetricsAggregatorAction {
    /// Initialize metrics collection
    Initialize,

    /// Update metrics from an event
    UpdateMetrics {
        envelope: Box<obzenflow_core::EventEnvelope<obzenflow_core::ChainEvent>>,
    },

    /// Process system events from the system journal (FLOWIP-059b)
    ProcessSystemEvent {
        envelope: Box<obzenflow_core::EventEnvelope<obzenflow_core::event::SystemEvent>>,
    },

    /// Export metrics snapshot
    ExportMetrics,

    /// Publish drain complete event to journal
    PublishDrainComplete { last_event_id: Option<EventId> },
}

/// Context for the FSM - contains everything actions need to do their work
pub struct MetricsAggregatorContext {
    /// System journal for reporting
    pub system_journal: Arc<dyn Journal<obzenflow_core::event::SystemEvent>>,

    /// Mapping of stage IDs to their data journals for tail reads.
    pub stage_data_journals: HashMap<StageId, Arc<dyn Journal<ChainEvent>>>,

    /// Mapping of stage IDs to their error journals for tail reads.
    pub stage_error_journals: HashMap<StageId, Arc<dyn Journal<ChainEvent>>>,

    /// Flow-scoped backpressure registry for observability (FLOWIP-086k).
    pub backpressure_registry: Option<Arc<crate::backpressure::BackpressureRegistry>>,

    /// Whether to include error journals in metrics collection
    pub include_error_journals: bool,

    pub exporter: Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>>,
    pub metrics_store: MetricsStore,
    pub export_interval_secs: u64,
    pub system_id: SystemId,
    pub stage_metadata: HashMap<StageId, StageMetadata>,
}

pub(crate) struct MetricsAggregatorIo {
    pub(crate) data_subscription:
        crate::messaging::upstream_subscription::UpstreamSubscription<ChainEvent>,
    pub(crate) error_subscription:
        Option<crate::messaging::upstream_subscription::UpstreamSubscription<ChainEvent>>,
    pub(crate) system_subscription: crate::messaging::system_subscription::SystemSubscription<
        obzenflow_core::event::SystemEvent,
    >,
}

/// Simple metrics storage
#[derive(Default)]
pub struct MetricsStore {
    pub stage_metrics: std::collections::HashMap<StageId, StageMetrics>,
    pub last_event_id: Option<EventId>,
    pub flow_start_time: Option<std::time::Instant>,
    pub first_event_time: Option<std::time::Instant>,
    pub last_event_time: Option<std::time::Instant>,
    pub total_events_processed: u64,

    /// Per-stage vector clock watermark (FLOWIP-059c)
    /// Tracks the highest writer sequence observed for each stage's journal writer.
    pub stage_vector_clocks: HashMap<StageId, u64>,

    /// Per-system vector clock watermark (FLOWIP-059c).
    ///
    /// Tracks the highest writer sequence observed for each system writer in the system journal.
    /// We intentionally track only `WriterId::System` clocks to avoid confusing stage-journal
    /// freshness with system-journal stage lifecycle events (separate clock domains).
    pub system_vector_clocks: HashMap<SystemId, u64>,

    // Middleware observability accumulation (FLOWIP-059a)
    pub circuit_breaker_state: HashMap<StageId, f64>,
    pub circuit_breaker_rejection_rate: HashMap<StageId, f64>,
    pub circuit_breaker_consecutive_failures: HashMap<StageId, f64>,
    pub circuit_breaker_requests_total: HashMap<StageId, u64>,
    pub circuit_breaker_rejections_total: HashMap<StageId, u64>,
    pub circuit_breaker_opened_total: HashMap<StageId, u64>,
    pub circuit_breaker_successes_total: HashMap<StageId, u64>,
    pub circuit_breaker_failures_total: HashMap<StageId, u64>,
    pub circuit_breaker_time_in_state_seconds_total: HashMap<(StageId, String), f64>,
    pub circuit_breaker_state_transitions_total: HashMap<(StageId, String, String), u64>,
    circuit_breaker_last_state: HashMap<StageId, String>,

    pub rate_limiter_events_total: HashMap<StageId, u64>,
    pub rate_limiter_delayed_total: HashMap<StageId, u64>,
    pub rate_limiter_tokens_consumed_total: HashMap<StageId, f64>,
    pub rate_limiter_delay_seconds_total: HashMap<StageId, f64>,
    // Bucket state for gauge metrics (FLOWIP-059a-3 Issue 3)
    pub rate_limiter_bucket_tokens: HashMap<StageId, f64>,
    pub rate_limiter_bucket_capacity: HashMap<StageId, f64>,

    // Contract metrics accumulation (FLOWIP-059a)
    pub contract_metrics: ContractMetricsSnapshot,

    // HTTP ingestion telemetry (FLOWIP-084d)
    pub ingestion_metrics: HashMap<String, IngestionTelemetrySnapshot>,

    // HTTP pull telemetry (FLOWIP-084e)
    pub http_pull_metrics: HashMap<StageId, HttpPullTelemetry>,

    // System event tracking (FLOWIP-059b - essential events only)
    // Track all states each stage has been in: (StageId, state_name) -> true
    pub stage_lifecycle_states: HashMap<(StageId, String), bool>,
    pub pipeline_state: String,
}

#[derive(Clone, Default)]
pub struct StageMetrics {
    pub errors_by_kind: HashMap<ErrorKind, u64>,
    // Runtime context metrics (FLOWIP-056c / FLOWIP-059 Phase 6)
    pub last_in_flight: Option<u32>,
    pub last_failures_total: Option<u64>,
    // Join-only gauge (Live join): number of reference events processed since the last stream event.
    pub join_reference_since_last_stream: Option<u64>,
    // Wide-event snapshot counters (Phase 6)
    pub latest_events_processed_total: Option<u64>,
    pub latest_events_accumulated_total: Option<u64>,
    pub latest_events_emitted_total: Option<u64>,
    pub latest_errors_total: Option<u64>,
    pub event_loops_total: u64,
    pub event_loops_with_work_total: u64,
    // Wide-event snapshot percentiles (Phase 6) - pre-computed by stage, in milliseconds
    pub snapshot_p50_ms: Option<u64>,
    pub snapshot_p90_ms: Option<u64>,
    pub snapshot_p95_ms: Option<u64>,
    pub snapshot_p99_ms: Option<u64>,
    pub snapshot_p999_ms: Option<u64>,
    // Actual sum of processing times (nanoseconds) - never reconstructed from percentiles
    pub processing_time_sum_nanos: Option<u64>,
    // Stage-specific timing for accurate rate calculation
    pub first_event_time: Option<std::time::Instant>,
    pub last_event_time: Option<std::time::Instant>,
}

impl MetricsAggregatorContext {
    pub(crate) async fn new(
        inputs: crate::metrics::inputs::MetricsInputs,
        system_journal: Arc<dyn Journal<obzenflow_core::event::SystemEvent>>,
        exporter: Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>>,
        export_interval_secs: u64,
        system_id: SystemId,
        stage_metadata: HashMap<StageId, StageMetadata>,
    ) -> Result<(Self, MetricsAggregatorIo), String> {
        // Initialize in-memory metrics store so we can seed snapshot fields
        // before constructing subscriptions (FLOWIP-059 Phase 6).
        let mut metrics_store = MetricsStore::default();

        // Helper to attach stage names to journals for better diagnostics
        let with_names = |journals: &[(StageId, Arc<dyn Journal<ChainEvent>>)]| {
            journals
                .iter()
                .map(|(id, journal)| {
                    let name = stage_metadata
                        .get(id)
                        .map(|m| m.name.clone())
                        .unwrap_or_else(|| format!("{id:?}"));
                    (*id, name, journal.clone())
                })
                .collect::<Vec<_>>()
        };

        // Phase 6/059d: wide-event snapshot seeding and tail-aware start for data journals.
        let data_with_names = with_names(&inputs.stage_data_journals);
        let mut data_start_positions = Vec::with_capacity(data_with_names.len());

        for (stage_id, stage_name, journal) in &data_with_names {
            // Tail-read last event with runtime_context, if any.
            let tail_snapshot = match journal.read_last_n(1).await {
                Ok(mut events) => events
                    .drain(..)
                    .find(|env| env.event.runtime_context.is_some()),
                Err(e) => {
                    tracing::warn!(
                        target: "flowip-059",
                        owner = "metrics_aggregator",
                        stage_id = ?stage_id,
                        stage_name = stage_name,
                        error = ?e,
                        "Failed to tail-read data journal for snapshot; seeding skipped for this stage"
                    );
                    None
                }
            };

            if let Some(envelope) = &tail_snapshot {
                if let Some(runtime_ctx) = &envelope.event.runtime_context {
                    {
                        let metrics = metrics_store
                            .stage_metrics
                            .entry(*stage_id)
                            .or_insert_with(StageMetrics::default);

                        // Seed wide-event snapshot fields from the latest event
                        // Use max() for monotonic counters to handle out-of-order reads
                        metrics.latest_events_processed_total = Some(
                            metrics
                                .latest_events_processed_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_processed_total),
                        );
                        metrics.latest_events_accumulated_total = Some(
                            metrics
                                .latest_events_accumulated_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_accumulated_total),
                        );
                        metrics.latest_events_emitted_total = Some(
                            metrics
                                .latest_events_emitted_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_emitted_total),
                        );
                        metrics.latest_errors_total = Some(
                            metrics
                                .latest_errors_total
                                .unwrap_or(0)
                                .max(runtime_ctx.errors_total),
                        );
                        metrics.last_in_flight = Some(runtime_ctx.in_flight);
                        metrics.last_failures_total = Some(runtime_ctx.failures_total);
                        metrics.join_reference_since_last_stream =
                            Some(runtime_ctx.join_reference_since_last_stream);
                        metrics.event_loops_total =
                            metrics.event_loops_total.max(runtime_ctx.event_loops_total);
                        metrics.event_loops_with_work_total = metrics
                            .event_loops_with_work_total
                            .max(runtime_ctx.event_loops_with_work_total);

                        // Seed pre-computed percentiles from runtime_context
                        metrics.snapshot_p50_ms = Some(runtime_ctx.recent_p50_ms);
                        metrics.snapshot_p90_ms = Some(runtime_ctx.recent_p90_ms);
                        metrics.snapshot_p95_ms = Some(runtime_ctx.recent_p95_ms);
                        metrics.snapshot_p99_ms = Some(runtime_ctx.recent_p99_ms);
                        metrics.snapshot_p999_ms = Some(runtime_ctx.recent_p999_ms);

                        // Actual sum - never reconstructed from percentiles
                        metrics.processing_time_sum_nanos =
                            Some(runtime_ctx.processing_time_sum_nanos);
                    }

                    metrics_store
                        .update_control_metrics_from_runtime_context(*stage_id, runtime_ctx);
                }

                // Seed per-stage vector clock watermark from the last envelope
                let writer_id = *envelope.event.writer_id();
                let writer_key = writer_id.to_string();
                let seq = envelope.vector_clock.get(&writer_key);
                let entry = metrics_store
                    .stage_vector_clocks
                    .entry(*stage_id)
                    .or_insert(0);
                *entry = (*entry).max(seq);
            }

            // Determine starting position for data subscription by streaming to EOF.
            // This is O(n) in time but O(1) in memory and keeps semantics simple.
            let start_position = match journal.reader().await {
                Ok(mut reader) => {
                    let mut pos: u64 = 0;
                    loop {
                        match reader.next().await {
                            Ok(Some(_)) => {
                                pos += 1;
                            }
                            Ok(None) => break,
                            Err(e) => {
                                tracing::warn!(
                                    target: "flowip-059",
                                    owner = "metrics_aggregator",
                                    stage_id = ?stage_id,
                                    stage_name = stage_name,
                                    error = ?e,
                                    "Failed while streaming data journal to determine tail position; starting from 0"
                                );
                                pos = 0;
                                break;
                            }
                        }
                    }
                    pos
                }
                Err(e) => {
                    tracing::warn!(
                        target: "flowip-059",
                        owner = "metrics_aggregator",
                        stage_id = ?stage_id,
                        stage_name = stage_name,
                        error = ?e,
                        "Failed to create reader for data journal; starting from 0"
                    );
                    0
                }
            };

            data_start_positions.push(start_position);
        }

        tracing::info!(
            upstream_count = inputs.stage_data_journals.len(),
            upstream_stages = tracing::field::debug(
                &inputs
                    .stage_data_journals
                    .iter()
                    .map(|(id, _)| *id)
                    .collect::<Vec<_>>(),
            ),
            "MetricsAggregator creating data subscription (tail-start)"
        );
        // Create subscription for data journals starting at tail positions.
        // Readers are treated as logically at EOF for historical data
        // (baseline_at_tail = true) while still observing any new events
        // appended after subscription creation (FLOWIP-059d).
        let data_subscription =
            crate::messaging::upstream_subscription::UpstreamSubscription::new_at_tail(
                "metrics_aggregator",
                &data_with_names,
                &data_start_positions,
            )
            .await
            .map_err(|e| format!("Failed to create data subscription: {e}"))?;

        // Also seed wide-event snapshots from error journals (late error snapshots),
        // using the same tail-aware helper. This keeps StageMetrics consistent even
        // when the last wide event for a stage is written to an error journal.
        let error_with_names = with_names(&inputs.error_journals);
        for (stage_id, stage_name, journal) in &error_with_names {
            match journal.read_last_n(1).await {
                Ok(mut events) => {
                    if let Some(envelope) = events
                        .drain(..)
                        .find(|env| env.event.runtime_context.is_some())
                    {
                        if let Some(runtime_ctx) = &envelope.event.runtime_context {
                            {
                                let metrics = metrics_store
                                    .stage_metrics
                                    .entry(*stage_id)
                                    .or_insert_with(StageMetrics::default);

                                metrics.latest_events_processed_total = Some(
                                    metrics
                                        .latest_events_processed_total
                                        .unwrap_or(0)
                                        .max(runtime_ctx.events_processed_total),
                                );
                                metrics.latest_events_accumulated_total = Some(
                                    metrics
                                        .latest_events_accumulated_total
                                        .unwrap_or(0)
                                        .max(runtime_ctx.events_accumulated_total),
                                );
                                metrics.latest_events_emitted_total = Some(
                                    metrics
                                        .latest_events_emitted_total
                                        .unwrap_or(0)
                                        .max(runtime_ctx.events_emitted_total),
                                );
                                metrics.latest_errors_total = Some(
                                    metrics
                                        .latest_errors_total
                                        .unwrap_or(0)
                                        .max(runtime_ctx.errors_total),
                                );
                                metrics.last_in_flight = Some(runtime_ctx.in_flight);
                                metrics.last_failures_total = Some(runtime_ctx.failures_total);
                                metrics.join_reference_since_last_stream =
                                    Some(runtime_ctx.join_reference_since_last_stream);
                                metrics.event_loops_total =
                                    metrics.event_loops_total.max(runtime_ctx.event_loops_total);
                                metrics.event_loops_with_work_total = metrics
                                    .event_loops_with_work_total
                                    .max(runtime_ctx.event_loops_with_work_total);

                                metrics.snapshot_p50_ms = Some(runtime_ctx.recent_p50_ms);
                                metrics.snapshot_p90_ms = Some(runtime_ctx.recent_p90_ms);
                                metrics.snapshot_p95_ms = Some(runtime_ctx.recent_p95_ms);
                                metrics.snapshot_p99_ms = Some(runtime_ctx.recent_p99_ms);
                                metrics.snapshot_p999_ms = Some(runtime_ctx.recent_p999_ms);

                                // Actual sum - never reconstructed from percentiles
                                metrics.processing_time_sum_nanos =
                                    Some(runtime_ctx.processing_time_sum_nanos);
                            }

                            metrics_store.update_control_metrics_from_runtime_context(
                                *stage_id,
                                runtime_ctx,
                            );
                        }

                        let writer_id = *envelope.event.writer_id();
                        let writer_key = writer_id.to_string();
                        let seq = envelope.vector_clock.get(&writer_key);
                        let entry = metrics_store
                            .stage_vector_clocks
                            .entry(*stage_id)
                            .or_insert(0);
                        *entry = (*entry).max(seq);
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        target: "flowip-059",
                        owner = "metrics_aggregator",
                        stage_id = ?stage_id,
                        stage_name = stage_name,
                        error = ?e,
                        "Failed to tail-read error journal for snapshot; seeding skipped for this stage"
                    );
                }
            }
        }

        // Determine starting positions for error subscriptions by streaming to EOF.
        // This mirrors the data journal behavior and ensures error subscriptions
        // can start from tail while still observing any new events appended after
        // the metrics aggregator is created.
        let mut error_start_positions = Vec::with_capacity(error_with_names.len());
        for (stage_id, stage_name, journal) in &error_with_names {
            let start_position = match journal.reader().await {
                Ok(mut reader) => {
                    let mut pos: u64 = 0;
                    loop {
                        match reader.next().await {
                            Ok(Some(_)) => {
                                pos += 1;
                            }
                            Ok(None) => break,
                            Err(e) => {
                                tracing::warn!(
                                    target: "flowip-059",
                                    owner = "metrics_aggregator",
                                    stage_id = ?stage_id,
                                    stage_name = stage_name,
                                    error = ?e,
                                    "Failed while streaming error journal to determine tail position; starting from 0"
                                );
                                pos = 0;
                                break;
                            }
                        }
                    }
                    pos
                }
                Err(e) => {
                    tracing::warn!(
                        target: "flowip-059",
                        owner = "metrics_aggregator",
                        stage_id = ?stage_id,
                        stage_name = stage_name,
                        error = ?e,
                        "Failed to create reader for error journal; starting from 0"
                    );
                    0
                }
            };

            error_start_positions.push(start_position);
        }

        if !inputs.error_journals.is_empty() {
            tracing::info!(
                upstream_count = inputs.error_journals.len(),
                upstream_stages = tracing::field::debug(
                    &inputs
                        .error_journals
                        .iter()
                        .map(|(id, _)| *id)
                        .collect::<Vec<_>>(),
                ),
                "MetricsAggregator creating error subscription (tail-start)"
            );
        }

        // Create subscription for error journals (FLOWIP-082g), starting at
        // computed tail positions so they are treated as logically at EOF for
        // historical data while still observing any new events appended after
        // subscription creation.
        let error_subscription = if !inputs.error_journals.is_empty() {
            Some(
                crate::messaging::upstream_subscription::UpstreamSubscription::new_at_tail(
                    "metrics_aggregator",
                    &error_with_names,
                    &error_start_positions,
                )
                .await
                .map_err(|e| format!("Failed to create error subscription: {e}"))?,
            )
        } else {
            None
        };

        // Create reader for system journal to receive lifecycle events (FLOWIP-059b)
        let system_reader = system_journal
            .reader()
            .await
            .map_err(|e| format!("Failed to create system journal reader: {e:?}"))?;

        // Wrap in SystemSubscription for consistent polling interface
        let system_subscription = crate::messaging::system_subscription::SystemSubscription::new(
            system_reader,
            "metrics_aggregator".to_string(),
        );

        // Build maps of journals for tail-read helpers used during export.
        let stage_data_journals: HashMap<StageId, Arc<dyn Journal<ChainEvent>>> = inputs
            .stage_data_journals
            .iter()
            .map(|(id, journal)| (*id, journal.clone()))
            .collect();
        let stage_error_journals: HashMap<StageId, Arc<dyn Journal<ChainEvent>>> = inputs
            .error_journals
            .iter()
            .map(|(id, journal)| (*id, journal.clone()))
            .collect();

        let context = Self {
            system_journal,
            stage_data_journals,
            stage_error_journals,
            backpressure_registry: inputs.backpressure_registry.clone(),
            include_error_journals: true, // Default to true per FLOWIP-082g
            exporter,
            metrics_store,
            export_interval_secs,
            system_id,
            stage_metadata,
        };

        let io = MetricsAggregatorIo {
            data_subscription,
            error_subscription,
            system_subscription,
        };

        Ok((context, io))
    }
}

impl FsmContext for MetricsAggregatorContext {}

fn infer_join_reference_mode_from_fsm_state(fsm_state: &str) -> Option<&'static str> {
    match fsm_state {
        "Live" => Some("live"),
        "Hydrating" | "Enriching" => Some("finite_eof"),
        _ => None,
    }
}

impl MetricsAggregatorContext {
    /// Build an `AppMetricsSnapshot` from the current in-memory `metrics_store`.
    ///
    /// This logic was originally inlined in the `ExportMetrics` action and has
    /// been refactored for reuse. It assumes that any tail-read refresh has
    /// already been applied to `metrics_store`.
    fn build_app_metrics_snapshot(&self) -> obzenflow_core::metrics::AppMetricsSnapshot {
        let store = &self.metrics_store;
        let mut snapshot = obzenflow_core::metrics::AppMetricsSnapshot::default();

        tracing::info!(
            "Exporting metrics: {} stage entries",
            store.stage_metrics.len()
        );

        // Flow-level aggregates derived from per-stage snapshots
        let mut flow_events_in_total: u64 = 0;
        let mut flow_events_out_total: u64 = 0;
        let mut flow_errors_total_snapshot: u64 = 0;
        let mut total_events_processed_snapshot: u64 = 0;
        let mut total_event_loops: u64 = 0;
        let mut total_event_loops_with_work: u64 = 0;

        // Convert stage metrics to snapshot format
        for (stage_id, metrics) in &store.stage_metrics {
            // Prefer wide-event snapshot counters when available
            let events_count = metrics.latest_events_processed_total.unwrap_or(0);
            snapshot.event_counts.insert(*stage_id, events_count);

            let accumulated_count = metrics.latest_events_accumulated_total.unwrap_or(0);
            snapshot
                .events_accumulated_total
                .insert(*stage_id, accumulated_count);

            let emitted_count = metrics.latest_events_emitted_total.unwrap_or(0);
            snapshot
                .events_emitted_total
                .insert(*stage_id, emitted_count);

            if let Some(value) = metrics.join_reference_since_last_stream {
                if let Some(metadata) = self.stage_metadata.get(stage_id) {
                    if metadata.stage_type == StageType::Join {
                        snapshot
                            .join_reference_since_last_stream
                            .insert(*stage_id, value);
                    }
                }
            }

            // Use wide-event snapshot errors_total as authoritative.
            let stage_errors_total = metrics.latest_errors_total.unwrap_or(0);
            snapshot.error_counts.insert(*stage_id, stage_errors_total);

            if !metrics.errors_by_kind.is_empty() && stage_errors_total > 0 {
                snapshot
                    .error_counts_by_kind
                    .insert(*stage_id, metrics.errors_by_kind.clone());
            }

            // Add processing time histogram reconstructed from runtime_context percentiles.
            if events_count > 0 && metrics.snapshot_p50_ms.is_some() {
                let mut percentiles = std::collections::HashMap::new();
                if let Some(p50) = metrics.snapshot_p50_ms {
                    percentiles.insert(Percentile::P50, (p50 * 1_000_000) as f64);
                }
                if let Some(p90) = metrics.snapshot_p90_ms {
                    percentiles.insert(Percentile::P90, (p90 * 1_000_000) as f64);
                }
                if let Some(p95) = metrics.snapshot_p95_ms {
                    percentiles.insert(Percentile::P95, (p95 * 1_000_000) as f64);
                }
                if let Some(p99) = metrics.snapshot_p99_ms {
                    percentiles.insert(Percentile::P99, (p99 * 1_000_000) as f64);
                }
                if let Some(p999) = metrics.snapshot_p999_ms {
                    percentiles.insert(Percentile::P999, (p999 * 1_000_000) as f64);
                }

                // Use actual sum - never reconstructed from percentiles (FLOWIP-059a-3)
                let sum_nanos = metrics.processing_time_sum_nanos.unwrap_or(0);

                let hist_snapshot = obzenflow_core::metrics::HistogramSnapshot {
                    count: events_count,
                    sum: sum_nanos as f64,
                    min: (metrics.snapshot_p50_ms.unwrap_or(0) * 1_000_000) as f64,
                    max: (metrics.snapshot_p999_ms.unwrap_or(0) * 1_000_000) as f64,
                    percentiles,
                };

                snapshot.processing_times.insert(*stage_id, hist_snapshot);
            }

            // Add runtime context metrics if available (FLOWIP-056c)
            if let Some(in_flight) = metrics.last_in_flight {
                snapshot.in_flight.insert(*stage_id, in_flight as f64);
            }

            if let Some(failures_total) = metrics.last_failures_total {
                snapshot.failures_total.insert(*stage_id, failures_total);
            }

            // Event loop metrics are cumulative counters
            snapshot
                .event_loops_total
                .insert(*stage_id, metrics.event_loops_total);
            snapshot
                .event_loops_with_work_total
                .insert(*stage_id, metrics.event_loops_with_work_total);

            // Aggregate flow-level metrics from snapshots
            total_events_processed_snapshot =
                total_events_processed_snapshot.saturating_add(events_count);

            flow_errors_total_snapshot =
                flow_errors_total_snapshot.saturating_add(stage_errors_total);

            total_event_loops = total_event_loops.saturating_add(metrics.event_loops_total);
            total_event_loops_with_work =
                total_event_loops_with_work.saturating_add(metrics.event_loops_with_work_total);

            if let Some(metadata) = self.stage_metadata.get(stage_id) {
                match metadata.stage_type {
                    obzenflow_core::event::context::StageType::FiniteSource
                    | obzenflow_core::event::context::StageType::InfiniteSource => {
                        flow_events_in_total = flow_events_in_total.saturating_add(events_count);
                    }
                    obzenflow_core::event::context::StageType::Sink => {
                        flow_events_out_total = flow_events_out_total.saturating_add(events_count);
                    }
                    _ => {}
                }
            }

            tracing::debug!(
                "Exported metrics for {:?}: events={}, errors_total_snapshot={}",
                stage_id,
                events_count,
                stage_errors_total
            );
        }

        // Add flow-level metrics
        if let (Some(first_time), Some(last_time)) = (store.first_event_time, store.last_event_time)
        {
            let flow_duration = last_time.duration_since(first_time);
            let flow_metrics = obzenflow_core::metrics::FlowMetricsSnapshot {
                flow_duration: MetricsDuration::from(flow_duration),
                total_events_processed: total_events_processed_snapshot,
                events_in: flow_events_in_total,
                events_out: flow_events_out_total,
                errors_total: flow_errors_total_snapshot,
                event_loops_total: total_event_loops,
                event_loops_with_work_total: total_event_loops_with_work,
            };
            snapshot.flow_metrics = Some(flow_metrics);
        }

        // Add stage metadata
        snapshot.stage_metadata = self.stage_metadata.clone();

        // FLOWIP-059a: Middleware metrics
        snapshot.circuit_breaker_state = store.circuit_breaker_state.clone();
        snapshot.circuit_breaker_rejection_rate = store.circuit_breaker_rejection_rate.clone();
        snapshot.circuit_breaker_consecutive_failures =
            store.circuit_breaker_consecutive_failures.clone();
        snapshot.circuit_breaker_requests_total = store.circuit_breaker_requests_total.clone();
        snapshot.circuit_breaker_rejections_total = store.circuit_breaker_rejections_total.clone();
        snapshot.circuit_breaker_opened_total = store.circuit_breaker_opened_total.clone();
        snapshot.circuit_breaker_successes_total = store.circuit_breaker_successes_total.clone();
        snapshot.circuit_breaker_failures_total = store.circuit_breaker_failures_total.clone();
        snapshot.circuit_breaker_time_in_state_seconds_total =
            store.circuit_breaker_time_in_state_seconds_total.clone();
        snapshot.circuit_breaker_state_transitions_total =
            store.circuit_breaker_state_transitions_total.clone();

        snapshot.rate_limiter_utilization = store
            .rate_limiter_bucket_capacity
            .iter()
            .filter_map(|(stage_id, capacity)| {
                if *capacity <= 0.0 {
                    return None;
                }
                let tokens = store.rate_limiter_bucket_tokens.get(stage_id)?;
                let utilization = 1.0 - (*tokens / *capacity);
                Some((*stage_id, utilization.clamp(0.0, 1.0)))
            })
            .collect();
        snapshot.rate_limiter_events_total = store.rate_limiter_events_total.clone();
        snapshot.rate_limiter_delayed_total = store.rate_limiter_delayed_total.clone();
        snapshot.rate_limiter_tokens_consumed_total =
            store.rate_limiter_tokens_consumed_total.clone();
        snapshot.rate_limiter_delay_seconds_total = store.rate_limiter_delay_seconds_total.clone();
        snapshot.rate_limiter_bucket_tokens = store.rate_limiter_bucket_tokens.clone();
        snapshot.rate_limiter_bucket_capacity = store.rate_limiter_bucket_capacity.clone();

        // FLOWIP-086k: Backpressure metrics (registry snapshot, not event-derived).
        snapshot.backpressure_bypass_enabled =
            crate::backpressure::BackpressureWriter::is_bypass_enabled();
        if let Some(registry) = &self.backpressure_registry {
            let bp = registry.metrics_snapshot();

            snapshot.backpressure_window = bp.edge_window;
            snapshot.backpressure_in_flight = bp.edge_in_flight;
            snapshot.backpressure_credits = bp.edge_credits;

            snapshot.backpressure_blocked = bp
                .stage_blocked
                .into_iter()
                .map(|(stage_id, blocked)| (stage_id, if blocked { 1.0 } else { 0.0 }))
                .collect();
            snapshot.backpressure_min_reader_seq = bp.stage_min_reader_seq;
            snapshot.backpressure_writer_seq = bp.stage_writer_seq;
            snapshot.backpressure_wait_seconds_total = bp
                .stage_wait_nanos_total
                .into_iter()
                .map(|(stage_id, nanos)| (stage_id, nanos as f64 / 1_000_000_000.0))
                .collect();
        }

        // FLOWIP-059a: Contract metrics
        snapshot.contract_metrics = store.contract_metrics.clone();

        // FLOWIP-084d: HTTP ingestion telemetry (wide events)
        snapshot.ingestion_metrics = store.ingestion_metrics.clone();

        // FLOWIP-084e: HTTP pull telemetry (wide events)
        snapshot.http_pull_metrics = store.http_pull_metrics.clone();

        // FLOWIP-059b: Add lifecycle states
        snapshot.stage_lifecycle_states = store.stage_lifecycle_states.clone();
        snapshot.pipeline_state = store.pipeline_state.clone();

        // Add stage timestamps for rate calculation
        let now = std::time::Instant::now();
        let now_utc = chrono::Utc::now();

        for (stage_id, metrics) in &store.stage_metrics {
            if let Some(first_time) = metrics.first_event_time {
                let elapsed_since_first = now.duration_since(first_time);
                let first_datetime =
                    now_utc - chrono::Duration::from_std(elapsed_since_first).unwrap_or_default();
                snapshot
                    .stage_first_event_time
                    .insert(*stage_id, first_datetime);
            }
            if let Some(last_time) = metrics.last_event_time {
                let elapsed_since_last = now.duration_since(last_time);
                let last_datetime =
                    now_utc - chrono::Duration::from_std(elapsed_since_last).unwrap_or_default();
                snapshot
                    .stage_last_event_time
                    .insert(*stage_id, last_datetime);
            }

            if let Some(seq) = store.stage_vector_clocks.get(stage_id) {
                snapshot.stage_vector_clocks.insert(*stage_id, *seq);
            }
        }

        snapshot
    }
}

impl MetricsStore {
    /// Returns true when every known stage has reached a terminal lifecycle
    /// state (completed, failed, or cancelled) according to system.events.
    pub fn all_stages_terminal(&self, stage_metadata: &HashMap<StageId, StageMetadata>) -> bool {
        stage_metadata.keys().all(|stage_id| {
            self.stage_lifecycle_states
                .get(&(*stage_id, "completed".to_string()))
                .copied()
                .unwrap_or(false)
                || self
                    .stage_lifecycle_states
                    .get(&(*stage_id, "failed".to_string()))
                    .copied()
                    .unwrap_or(false)
                || self
                    .stage_lifecycle_states
                    .get(&(*stage_id, "cancelled".to_string()))
                    .copied()
                    .unwrap_or(false)
        })
    }

    /// Returns true when the pipeline has reached a terminal lifecycle state.
    pub fn pipeline_terminal(&self) -> bool {
        matches!(
            self.pipeline_state.as_str(),
            "completed"
                | "failed"
                | "cancelled"
                | "drained"
                | "all_stages_completed"
                | "stop_requested"
        )
    }

    fn record_circuit_breaker_transition(&mut self, stage_id: StageId, state: &str) {
        let Some(next_state) = normalize_circuit_breaker_state_label(state) else {
            return;
        };

        let from_state = self
            .circuit_breaker_last_state
            .get(&stage_id)
            .map(String::as_str)
            .unwrap_or("closed");

        if from_state != next_state {
            let transition_key = (stage_id, from_state.to_string(), next_state.to_string());
            let count = self
                .circuit_breaker_state_transitions_total
                .entry(transition_key)
                .or_insert(0);
            *count = (*count).saturating_add(1);
        }

        self.circuit_breaker_last_state
            .insert(stage_id, next_state.to_string());
    }

    fn set_circuit_breaker_last_state(&mut self, stage_id: StageId, state: &str) {
        let Some(state) = normalize_circuit_breaker_state_label(state) else {
            return;
        };

        self.circuit_breaker_last_state
            .insert(stage_id, state.to_string());
    }

    fn update_control_metrics_from_runtime_context(
        &mut self,
        stage_id: StageId,
        runtime_ctx: &obzenflow_core::event::context::RuntimeContext,
    ) {
        let cb_present = runtime_ctx.cb_requests_total > 0
            || runtime_ctx.cb_rejections_total > 0
            || runtime_ctx.cb_opened_total > 0
            || runtime_ctx.cb_successes_total > 0
            || runtime_ctx.cb_failures_total > 0
            || runtime_ctx.cb_time_closed_seconds > 0.0
            || runtime_ctx.cb_time_open_seconds > 0.0
            || runtime_ctx.cb_time_half_open_seconds > 0.0;

        if cb_present {
            let total = self
                .circuit_breaker_requests_total
                .entry(stage_id)
                .or_insert(0);
            *total = (*total).max(runtime_ctx.cb_requests_total);

            let rejected = self
                .circuit_breaker_rejections_total
                .entry(stage_id)
                .or_insert(0);
            *rejected = (*rejected).max(runtime_ctx.cb_rejections_total);

            let opened = self
                .circuit_breaker_opened_total
                .entry(stage_id)
                .or_insert(0);
            *opened = (*opened).max(runtime_ctx.cb_opened_total);

            let successes = self
                .circuit_breaker_successes_total
                .entry(stage_id)
                .or_insert(0);
            *successes = (*successes).max(runtime_ctx.cb_successes_total);

            let failures = self
                .circuit_breaker_failures_total
                .entry(stage_id)
                .or_insert(0);
            *failures = (*failures).max(runtime_ctx.cb_failures_total);

            self.circuit_breaker_state
                .insert(stage_id, runtime_ctx.cb_state);

            self.circuit_breaker_time_in_state_seconds_total
                .entry((stage_id, "closed".to_string()))
                .and_modify(|v| *v = (*v).max(runtime_ctx.cb_time_closed_seconds))
                .or_insert(runtime_ctx.cb_time_closed_seconds);
            self.circuit_breaker_time_in_state_seconds_total
                .entry((stage_id, "open".to_string()))
                .and_modify(|v| *v = (*v).max(runtime_ctx.cb_time_open_seconds))
                .or_insert(runtime_ctx.cb_time_open_seconds);
            self.circuit_breaker_time_in_state_seconds_total
                .entry((stage_id, "half_open".to_string()))
                .and_modify(|v| *v = (*v).max(runtime_ctx.cb_time_half_open_seconds))
                .or_insert(runtime_ctx.cb_time_half_open_seconds);

            let state_label = if (runtime_ctx.cb_state - 0.5).abs() < f64::EPSILON {
                "half_open"
            } else if (runtime_ctx.cb_state - 1.0).abs() < f64::EPSILON {
                "open"
            } else {
                "closed"
            };
            self.set_circuit_breaker_last_state(stage_id, state_label);
        }

        let rl_present = runtime_ctx.rl_events_total > 0
            || runtime_ctx.rl_delayed_total > 0
            || runtime_ctx.rl_tokens_consumed_total > 0.0
            || runtime_ctx.rl_delay_seconds_total > 0.0;

        if rl_present {
            let events_total = self.rate_limiter_events_total.entry(stage_id).or_insert(0);
            *events_total = (*events_total).max(runtime_ctx.rl_events_total);

            let delayed_total = self.rate_limiter_delayed_total.entry(stage_id).or_insert(0);
            *delayed_total = (*delayed_total).max(runtime_ctx.rl_delayed_total);

            let tokens_consumed = self
                .rate_limiter_tokens_consumed_total
                .entry(stage_id)
                .or_insert(0.0);
            *tokens_consumed = (*tokens_consumed).max(runtime_ctx.rl_tokens_consumed_total);

            let delay_seconds = self
                .rate_limiter_delay_seconds_total
                .entry(stage_id)
                .or_insert(0.0);
            *delay_seconds = (*delay_seconds).max(runtime_ctx.rl_delay_seconds_total);

            // Bucket state is a gauge, not a counter - just overwrite with latest value
            self.rate_limiter_bucket_tokens
                .insert(stage_id, runtime_ctx.rl_bucket_tokens);
            self.rate_limiter_bucket_capacity
                .insert(stage_id, runtime_ctx.rl_bucket_capacity);
        }
    }
}

fn normalize_circuit_breaker_state_label(state: &str) -> Option<&'static str> {
    let state = state.trim();
    if state.eq_ignore_ascii_case("closed") {
        Some("closed")
    } else if state.eq_ignore_ascii_case("open") {
        Some("open")
    } else {
        let normalized = state.to_ascii_lowercase();
        match normalized.as_str() {
            "halfopen" | "half_open" | "half-open" => Some("half_open"),
            _ => None,
        }
    }
}

#[async_trait::async_trait]
impl FsmAction for MetricsAggregatorAction {
    type Context = MetricsAggregatorContext;

    async fn execute(&self, ctx: &mut Self::Context) -> Result<(), obzenflow_fsm::FsmError> {
        match self {
            MetricsAggregatorAction::Initialize => {
                tracing::info!("Metrics aggregator initialized");
                Ok(())
            }

            MetricsAggregatorAction::ProcessSystemEvent { envelope } => {
                tracing::info!(
                    event_id = %envelope.event.id(),
                    event_type = envelope.event.event_type_name(),
                    "Metrics aggregator ProcessSystemEvent action"
                );
                // FLOWIP-059b: Process system journal events for lifecycle tracking
                let store = &mut ctx.metrics_store;

                // FLOWIP-059c: Track system-writer vector clocks so `metrics_watermark` can cover
                // system-originated metrics (pipeline + metrics writers) in addition to stage journals.
                if let Some(system_id) = envelope.event.writer_id.as_system() {
                    let writer_key = envelope.event.writer_id.to_string();
                    let seq = envelope.vector_clock.get(&writer_key);
                    let entry = store.system_vector_clocks.entry(*system_id).or_insert(0);
                    *entry = (*entry).max(seq);
                }

                match &envelope.event.event {
                    obzenflow_core::event::SystemEventType::StageLifecycle { stage_id, event } => {
                        // Track ALL states each stage has been in (never overwrite)
                        match event {
                            obzenflow_core::event::StageLifecycleEvent::Running => {
                                store
                                    .stage_lifecycle_states
                                    .insert((*stage_id, "running".to_string()), true);
                                tracing::debug!("Stage {:?} transitioned to running", stage_id);
                            }
                            obzenflow_core::event::StageLifecycleEvent::Completed { .. } => {
                                store
                                    .stage_lifecycle_states
                                    .insert((*stage_id, "completed".to_string()), true);
                                tracing::debug!("Stage {:?} transitioned to completed", stage_id);
                            }
                            obzenflow_core::event::StageLifecycleEvent::Cancelled { .. } => {
                                store
                                    .stage_lifecycle_states
                                    .insert((*stage_id, "cancelled".to_string()), true);
                                tracing::debug!("Stage {:?} transitioned to cancelled", stage_id);
                            }
                            obzenflow_core::event::StageLifecycleEvent::Failed { .. } => {
                                store
                                    .stage_lifecycle_states
                                    .insert((*stage_id, "failed".to_string()), true);
                                tracing::debug!("Stage {:?} transitioned to failed", stage_id);
                            }
                            _ => {} // Skip draining, drained for now
                        }
                    }
                    obzenflow_core::event::SystemEventType::PipelineLifecycle(event) => {
                        // Track only essential pipeline events, with monotonic semantics:
                        // - "failed" is sticky and never regresses.
                        // - "completed" never regresses to "drained".
                        // - "drained" is only used when no explicit outcome was ever observed.
                        match event {
                            obzenflow_core::event::PipelineLifecycleEvent::StopRequested {
                                ..
                            } => {
                                if store.pipeline_state.is_empty() {
                                    store.pipeline_state = "stop_requested".to_string();
                                }
                                tracing::info!("Pipeline: stop requested (metrics view)");
                            }
                            obzenflow_core::event::PipelineLifecycleEvent::AllStagesCompleted {
                                ..
                            } => {
                                if store.pipeline_state.is_empty() {
                                    store.pipeline_state = "all_stages_completed".to_string();
                                }
                                tracing::info!("Pipeline: all stages completed (metrics view)");
                            }
                            obzenflow_core::event::PipelineLifecycleEvent::Completed { .. } => {
                                if store.pipeline_state != "failed" {
                                    store.pipeline_state = "completed".to_string();
                                    tracing::info!("Pipeline: completed (metrics view)");
                                } else {
                                    tracing::info!(
                                        "Pipeline: completed event observed after failed; \
	                                         keeping failed as terminal state (metrics view)"
                                    );
                                }
                            }
                            obzenflow_core::event::PipelineLifecycleEvent::Cancelled { .. } => {
                                if store.pipeline_state != "failed" {
                                    store.pipeline_state = "cancelled".to_string();
                                    tracing::info!("Pipeline: cancelled (metrics view)");
                                } else {
                                    tracing::info!(
                                        "Pipeline: cancelled event observed after failed; \
	                                         keeping failed as terminal state (metrics view)"
                                    );
                                }
                            }
                            obzenflow_core::event::PipelineLifecycleEvent::Failed { .. } => {
                                // Failure is always terminal and sticky.
                                if store.pipeline_state != "failed" {
                                    store.pipeline_state = "failed".to_string();
                                    tracing::info!("Pipeline: failed (metrics view)");
                                }
                            }
                            obzenflow_core::event::PipelineLifecycleEvent::Drained => {
                                // Drained is a termination marker only; do not override an
                                // explicit completed/failed outcome.
                                match store.pipeline_state.as_str() {
                                    "failed" | "completed" => {
                                        tracing::info!(
                                            "Pipeline: drained event observed after terminal outcome; \
                                             keeping {} as terminal state (metrics view)",
                                            store.pipeline_state
                                        );
                                    }
                                    _ => {
                                        store.pipeline_state = "drained".to_string();
                                        tracing::info!("Pipeline: drained (metrics view)");
                                    }
                                }
                            }
                            _ => {} // Skip other pipeline events
                        }
                    }
                    obzenflow_core::event::SystemEventType::ContractResult {
                        upstream,
                        reader,
                        contract_name,
                        status,
                        cause,
                        reader_seq,
                        advertised_writer_seq,
                    } => {
                        let key = (*upstream, *reader, contract_name.clone(), status.clone());
                        let counter = store.contract_metrics.results_total.entry(key).or_insert(0);
                        *counter = (*counter).saturating_add(1);

                        if let Some(cause) = cause {
                            let key = (*upstream, *reader, contract_name.clone(), cause.clone());
                            let counter = store
                                .contract_metrics
                                .violations_total
                                .entry(key)
                                .or_insert(0);
                            *counter = (*counter).saturating_add(1);
                        }

                        let edge_key = (*upstream, *reader, contract_name.clone());
                        if let Some(seq) = reader_seq {
                            let gauge = store
                                .contract_metrics
                                .reader_seq
                                .entry(edge_key.clone())
                                .or_insert(0);
                            *gauge = (*gauge).max(seq.0);
                        }
                        if let Some(seq) = advertised_writer_seq {
                            let gauge = store
                                .contract_metrics
                                .advertised_writer_seq
                                .entry(edge_key)
                                .or_insert(0);
                            *gauge = (*gauge).max(seq.0);
                        }
                    }
                    obzenflow_core::event::SystemEventType::ContractOverrideByPolicy {
                        upstream,
                        reader,
                        contract_name,
                        policy,
                        ..
                    } => {
                        let key = (*upstream, *reader, contract_name.clone(), policy.clone());
                        let counter = store
                            .contract_metrics
                            .overrides_total
                            .entry(key)
                            .or_insert(0);
                        *counter = (*counter).saturating_add(1);
                    }
                    _ => {} // Skip MetricsCoordination and other event types
                }

                Ok(())
            }

            MetricsAggregatorAction::UpdateMetrics { envelope } => {
                tracing::trace!(
                    event_id = %envelope.event.id(),
                    event_type = envelope.event.event_type(),
                    "Metrics aggregator UpdateMetrics action"
                );
                let store = &mut ctx.metrics_store;

                // Update last event ID
                store.last_event_id = Some(envelope.event.id);

                let event = &envelope.event;

                // Update per-stage vector clock watermark (FLOWIP-059c).
                // We use the event writer_id component from the envelope's vector clock.
                let stage_id = event.flow_context.stage_id;
                let writer_id = *event.writer_id();
                let writer_key = writer_id.to_string();
                let seq = envelope.vector_clock.get(&writer_key);
                let entry = store.stage_vector_clocks.entry(stage_id).or_insert(0);
                *entry = (*entry).max(seq);

                // Capture flow_id for joinability (FLOWIP-059a) when it becomes available.
                if let Some(meta) = ctx.stage_metadata.get_mut(&stage_id) {
                    if meta.flow_id.is_none() {
                        if let Ok(flow_id) = FlowId::from_str(event.flow_context.flow_id.as_str()) {
                            meta.flow_id = Some(flow_id);
                        }
                    }

                    // Best-effort: infer join reference mode from the observed FSM state.
                    // This enables exporters to attach a `reference_mode` label for joins
                    // without requiring the pipeline to plumb join config into metrics metadata.
                    if meta.reference_mode.is_none() && meta.stage_type == StageType::Join {
                        if let Some(runtime_ctx) = &event.runtime_context {
                            if let Some(mode) =
                                infer_join_reference_mode_from_fsm_state(&runtime_ctx.fsm_state)
                            {
                                meta.reference_mode = Some(mode.to_string());
                            }
                        }
                    }
                }

                // Skip system events entirely; they are not part of per-stage wide metrics.
                if event.is_system() {
                    return Ok(());
                }

                // Consume HTTP ingestion telemetry wide events (FLOWIP-084d).
                if let ChainEventContent::Observability(ObservabilityPayload::Metrics(
                    MetricsLifecycle::Custom { name, value, .. },
                )) = &event.content
                {
                    if name == "http_ingestion.snapshot" {
                        match serde_json::from_value::<IngestionTelemetrySnapshot>(value.clone()) {
                            Ok(snapshot) => {
                                let entry = store
                                    .ingestion_metrics
                                    .entry(snapshot.base_path.clone())
                                    .or_insert_with(|| snapshot.clone());

                                // Monotonic counters: use max() to tolerate ordering.
                                entry.requests_total =
                                    entry.requests_total.max(snapshot.requests_total);
                                entry.events_accepted_total = entry
                                    .events_accepted_total
                                    .max(snapshot.events_accepted_total);
                                entry.events_rejected_auth_total = entry
                                    .events_rejected_auth_total
                                    .max(snapshot.events_rejected_auth_total);
                                entry.events_rejected_validation_total = entry
                                    .events_rejected_validation_total
                                    .max(snapshot.events_rejected_validation_total);
                                entry.events_rejected_buffer_full_total = entry
                                    .events_rejected_buffer_full_total
                                    .max(snapshot.events_rejected_buffer_full_total);
                                entry.events_rejected_not_ready_total = entry
                                    .events_rejected_not_ready_total
                                    .max(snapshot.events_rejected_not_ready_total);
                                entry.events_rejected_payload_too_large_total = entry
                                    .events_rejected_payload_too_large_total
                                    .max(snapshot.events_rejected_payload_too_large_total);
                                entry.events_rejected_invalid_json_total = entry
                                    .events_rejected_invalid_json_total
                                    .max(snapshot.events_rejected_invalid_json_total);
                                entry.events_rejected_channel_closed_total = entry
                                    .events_rejected_channel_closed_total
                                    .max(snapshot.events_rejected_channel_closed_total);

                                // Gauges: overwrite with latest.
                                entry.channel_depth = snapshot.channel_depth;
                                entry.channel_capacity =
                                    entry.channel_capacity.max(snapshot.channel_capacity);
                            }
                            Err(e) => tracing::warn!(
                                error = %e,
                                "Failed to decode http_ingestion.snapshot payload; ignoring"
                            ),
                        }
                    } else if name == "http_pull.snapshot" {
                        match serde_json::from_value::<HttpPullTelemetry>(value.clone()) {
                            Ok(snapshot) => {
                                let entry = store.http_pull_metrics.entry(stage_id).or_default();

                                // State/gauges: overwrite with latest.
                                entry.state = snapshot.state.clone();
                                entry.wait_reason = snapshot.wait_reason.clone();
                                entry.next_wake_unix_secs = snapshot.next_wake_unix_secs;

                                // Timestamps/counters: monotonic max to tolerate ordering.
                                entry.last_success_unix_secs = match (
                                    entry.last_success_unix_secs,
                                    snapshot.last_success_unix_secs,
                                ) {
                                    (Some(a), Some(b)) => Some(a.max(b)),
                                    (Some(a), None) => Some(a),
                                    (None, Some(b)) => Some(b),
                                    (None, None) => None,
                                };

                                entry.requests_total =
                                    entry.requests_total.max(snapshot.requests_total);
                                entry.responses_2xx =
                                    entry.responses_2xx.max(snapshot.responses_2xx);
                                entry.responses_4xx =
                                    entry.responses_4xx.max(snapshot.responses_4xx);
                                entry.responses_5xx =
                                    entry.responses_5xx.max(snapshot.responses_5xx);
                                entry.rate_limited_total =
                                    entry.rate_limited_total.max(snapshot.rate_limited_total);
                                entry.retries_total =
                                    entry.retries_total.max(snapshot.retries_total);
                                entry.events_decoded_total = entry
                                    .events_decoded_total
                                    .max(snapshot.events_decoded_total);

                                entry.wait_seconds_rate_limit = entry
                                    .wait_seconds_rate_limit
                                    .max(snapshot.wait_seconds_rate_limit);
                                entry.wait_seconds_poll_interval = entry
                                    .wait_seconds_poll_interval
                                    .max(snapshot.wait_seconds_poll_interval);
                                entry.wait_seconds_backoff = entry
                                    .wait_seconds_backoff
                                    .max(snapshot.wait_seconds_backoff);
                            }
                            Err(e) => tracing::warn!(
                                error = %e,
                                "Failed to decode http_pull.snapshot payload; ignoring"
                            ),
                        }
                    }
                }

                // Consume middleware observability events (FLOWIP-059a).
                if let ChainEventContent::Observability(ObservabilityPayload::Middleware(
                    MiddlewareLifecycle::CircuitBreaker(cb),
                )) = &event.content
                {
                    match cb {
                        CircuitBreakerEvent::Opened {
                            error_rate: _,
                            failure_count,
                            ..
                        } => {
                            store.record_circuit_breaker_transition(stage_id, "open");
                            store.circuit_breaker_state.insert(stage_id, 1.0);
                            store
                                .circuit_breaker_consecutive_failures
                                .insert(stage_id, *failure_count as f64);
                        }
                        CircuitBreakerEvent::Closed { .. } => {
                            store.record_circuit_breaker_transition(stage_id, "closed");
                            store.circuit_breaker_state.insert(stage_id, 0.0);
                            store
                                .circuit_breaker_consecutive_failures
                                .insert(stage_id, 0.0);
                        }
                        CircuitBreakerEvent::HalfOpen { .. } => {
                            store.record_circuit_breaker_transition(stage_id, "half_open");
                            store.circuit_breaker_state.insert(stage_id, 0.5);
                        }
                        CircuitBreakerEvent::Summary {
                            requests_processed: _,
                            requests_rejected: _,
                            state,
                            consecutive_failures,
                            rejection_rate,
                            successes_total,
                            failures_total,
                            opened_total,
                            time_in_closed_seconds,
                            time_in_open_seconds,
                            time_in_half_open_seconds,
                            ..
                        } => {
                            store.set_circuit_breaker_last_state(stage_id, state);

                            store
                                .circuit_breaker_rejection_rate
                                .insert(stage_id, *rejection_rate);
                            store
                                .circuit_breaker_consecutive_failures
                                .insert(stage_id, *consecutive_failures as f64);

                            // Cumulative breaker stats (FLOWIP-059a-2). These are emitted
                            // as monotonic totals in the Summary wide event.
                            let opened = store
                                .circuit_breaker_opened_total
                                .entry(stage_id)
                                .or_insert(0);
                            *opened = (*opened).max(*opened_total);

                            let successes = store
                                .circuit_breaker_successes_total
                                .entry(stage_id)
                                .or_insert(0);
                            *successes = (*successes).max(*successes_total);

                            let failures = store
                                .circuit_breaker_failures_total
                                .entry(stage_id)
                                .or_insert(0);
                            *failures = (*failures).max(*failures_total);

                            store
                                .circuit_breaker_time_in_state_seconds_total
                                .entry((stage_id, "closed".to_string()))
                                .and_modify(|v| *v = (*v).max(*time_in_closed_seconds))
                                .or_insert(*time_in_closed_seconds);
                            store
                                .circuit_breaker_time_in_state_seconds_total
                                .entry((stage_id, "open".to_string()))
                                .and_modify(|v| *v = (*v).max(*time_in_open_seconds))
                                .or_insert(*time_in_open_seconds);
                            store
                                .circuit_breaker_time_in_state_seconds_total
                                .entry((stage_id, "half_open".to_string()))
                                .and_modify(|v| *v = (*v).max(*time_in_half_open_seconds))
                                .or_insert(*time_in_half_open_seconds);

                            // State string is Debug-formatted in the middleware
                            // (e.g. "Closed", "Open", "HalfOpen"); be liberal in parsing.
                            let state_norm = state.to_ascii_lowercase();
                            let state_value = match state_norm.as_str() {
                                "closed" => Some(0.0),
                                "open" => Some(1.0),
                                "halfopen" | "half_open" | "half-open" => Some(0.5),
                                _ => None,
                            };
                            if let Some(val) = state_value {
                                store.circuit_breaker_state.insert(stage_id, val);
                            }
                        }
                        _ => {}
                    }
                }

                // Track flow timing for rate calculation based on any non-system event.
                let now = std::time::Instant::now();
                if store.first_event_time.is_none() {
                    store.first_event_time = Some(now);
                    store.flow_start_time = Some(now);
                }
                store.last_event_time = Some(now);

                // Only count data/delivery events towards total_events_processed; control
                // events (EOF, consumption_final, etc.) carry snapshots but must not bump
                // the processed count. This keeps totals aligned with stage-wide metrics.
                if event.is_data() || event.is_delivery() {
                    store.total_events_processed += 1;
                }

                // Process data and delivery events using runtime_context snapshots only
                let stage_id = event.flow_context.stage_id;

                // First handle stage metrics (data + control wide events)
                {
                    let metrics = store.stage_metrics.entry(stage_id).or_default();

                    // Track stage timing
                    let now = std::time::Instant::now();
                    if metrics.first_event_time.is_none() {
                        metrics.first_event_time = Some(now);
                    }
                    metrics.last_event_time = Some(now);

                    // Extract runtime context metrics if available (FLOWIP-056c / FLOWIP-059d)
                    if let Some(runtime_ctx) = &event.runtime_context {
                        tracing::trace!(
                            "Runtime context for {:?}: in_flight={}, fsm_state={}",
                            stage_id,
                            runtime_ctx.in_flight,
                            runtime_ctx.fsm_state
                        );

                        // Store latest runtime metrics for export
                        // Use max() for monotonic counters to handle out-of-order reads
                        metrics.last_in_flight = Some(runtime_ctx.in_flight);
                        metrics.last_failures_total = Some(runtime_ctx.failures_total);
                        metrics.join_reference_since_last_stream =
                            Some(runtime_ctx.join_reference_since_last_stream);
                        metrics.latest_events_processed_total = Some(
                            metrics
                                .latest_events_processed_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_processed_total),
                        );
                        metrics.latest_events_accumulated_total = Some(
                            metrics
                                .latest_events_accumulated_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_accumulated_total),
                        );
                        metrics.latest_events_emitted_total = Some(
                            metrics
                                .latest_events_emitted_total
                                .unwrap_or(0)
                                .max(runtime_ctx.events_emitted_total),
                        );
                        metrics.latest_errors_total = Some(
                            metrics
                                .latest_errors_total
                                .unwrap_or(0)
                                .max(runtime_ctx.errors_total),
                        );

                        // Update cumulative event loop counters (take max to handle resets)
                        metrics.event_loops_total =
                            metrics.event_loops_total.max(runtime_ctx.event_loops_total);
                        metrics.event_loops_with_work_total = metrics
                            .event_loops_with_work_total
                            .max(runtime_ctx.event_loops_with_work_total);

                        // Update pre-computed percentiles from runtime_context
                        metrics.snapshot_p50_ms = Some(runtime_ctx.recent_p50_ms);
                        metrics.snapshot_p90_ms = Some(runtime_ctx.recent_p90_ms);
                        metrics.snapshot_p95_ms = Some(runtime_ctx.recent_p95_ms);
                        metrics.snapshot_p99_ms = Some(runtime_ctx.recent_p99_ms);
                        metrics.snapshot_p999_ms = Some(runtime_ctx.recent_p999_ms);

                        // Actual sum - never reconstructed from percentiles
                        metrics.processing_time_sum_nanos =
                            Some(runtime_ctx.processing_time_sum_nanos);
                    }

                    // Track error kinds for breakdown (total bounded later by snapshot errors_total)
                    if let ProcessingStatus::Error { kind, .. } = &event.processing_info.status {
                        let key = kind.clone().unwrap_or(ErrorKind::Unknown);
                        *metrics.errors_by_kind.entry(key).or_insert(0) += 1;
                    }
                } // metrics reference dropped here

                if let Some(runtime_ctx) = &event.runtime_context {
                    store.update_control_metrics_from_runtime_context(stage_id, runtime_ctx);
                }
                Ok(())
            }

            MetricsAggregatorAction::ExportMetrics => {
                tracing::info!("ExportMetrics action triggered");
                // Refresh in-memory stage metrics from journal tails before export so
                // `/metrics` reflects delivery truth even if subscriptions lag.
                for (stage_id, data_journal) in &ctx.stage_data_journals {
                    let error_journal = ctx.stage_error_journals.get(stage_id);
                    if let Some(snapshot) = tail_read::read_stage_metrics_from_tail(
                        data_journal,
                        error_journal,
                        *stage_id,
                    )
                    .await
                    {
                        let metrics = ctx
                            .metrics_store
                            .stage_metrics
                            .entry(*stage_id)
                            .or_default();

                        metrics.latest_events_processed_total = Some(
                            metrics
                                .latest_events_processed_total
                                .unwrap_or(0)
                                .max(snapshot.events_processed_total),
                        );
                        metrics.latest_events_accumulated_total = Some(
                            metrics
                                .latest_events_accumulated_total
                                .unwrap_or(0)
                                .max(snapshot.events_accumulated_total),
                        );
                        metrics.latest_events_emitted_total = Some(
                            metrics
                                .latest_events_emitted_total
                                .unwrap_or(0)
                                .max(snapshot.events_emitted_total),
                        );
                        metrics.latest_errors_total = Some(
                            metrics
                                .latest_errors_total
                                .unwrap_or(0)
                                .max(snapshot.errors_total),
                        );

                        // Use tail-read error breakdown as authoritative for this stage.
                        metrics.errors_by_kind = snapshot.errors_by_kind.clone();
                        metrics.last_in_flight = Some(snapshot.in_flight);
                        metrics.snapshot_p50_ms = Some(snapshot.recent_p50_ms);
                        metrics.snapshot_p90_ms = Some(snapshot.recent_p90_ms);
                        metrics.snapshot_p95_ms = Some(snapshot.recent_p95_ms);
                        metrics.snapshot_p99_ms = Some(snapshot.recent_p99_ms);
                        metrics.snapshot_p999_ms = Some(snapshot.recent_p999_ms);

                        // Actual sum - never reconstructed from percentiles
                        metrics.processing_time_sum_nanos =
                            Some(snapshot.processing_time_sum_nanos);
                    }

                    // Refresh control middleware cumulative metrics from the latest runtime_context.
                    if let Some(runtime_ctx) =
                        tail_read::read_latest_runtime_context_for_stage(data_journal, *stage_id)
                            .await
                    {
                        if let Some(meta) = ctx.stage_metadata.get_mut(stage_id) {
                            if meta.reference_mode.is_none() && meta.stage_type == StageType::Join {
                                if let Some(mode) =
                                    infer_join_reference_mode_from_fsm_state(&runtime_ctx.fsm_state)
                                {
                                    meta.reference_mode = Some(mode.to_string());
                                }
                            }
                        }
                        if let Some(metrics) = ctx.metrics_store.stage_metrics.get_mut(stage_id) {
                            metrics.join_reference_since_last_stream =
                                Some(runtime_ctx.join_reference_since_last_stream);
                        }
                        ctx.metrics_store
                            .update_control_metrics_from_runtime_context(*stage_id, &runtime_ctx);
                    }
                    if let Some(error_journal) = error_journal {
                        if let Some(runtime_ctx) = tail_read::read_latest_runtime_context_for_stage(
                            error_journal,
                            *stage_id,
                        )
                        .await
                        {
                            if let Some(metrics) = ctx.metrics_store.stage_metrics.get_mut(stage_id)
                            {
                                metrics.join_reference_since_last_stream =
                                    Some(runtime_ctx.join_reference_since_last_stream);
                            }
                            ctx.metrics_store
                                .update_control_metrics_from_runtime_context(
                                    *stage_id,
                                    &runtime_ctx,
                                );
                        }
                    }
                }

                if let Some(exporter) = &ctx.exporter {
                    let snapshot = ctx.build_app_metrics_snapshot();
                    tracing::info!("Pushing metrics snapshot to exporter");

                    if let Err(e) = exporter.update_app_metrics(snapshot) {
                        tracing::warn!("Failed to export metrics: {}", e);
                    } else {
                        tracing::info!("Successfully exported metrics");
                    }
                }

                // FLOWIP-059c: Emit a metrics watermark event so SSE clients can "pull-on-push"
                // for `/metrics` refresh and deterministic freshness gating.
                let mut clocks: std::collections::BTreeMap<String, u64> =
                    std::collections::BTreeMap::new();
                for (stage_id, seq) in &ctx.metrics_store.stage_vector_clocks {
                    clocks.insert(WriterId::from(*stage_id).to_string(), *seq);
                }
                for (system_id, seq) in &ctx.metrics_store.system_vector_clocks {
                    clocks.insert(WriterId::from(*system_id).to_string(), *seq);
                }

                let export_event = obzenflow_core::event::SystemEvent::new(
                    WriterId::from(ctx.system_id),
                    obzenflow_core::event::SystemEventType::MetricsCoordination(
                        obzenflow_core::event::MetricsCoordinationEvent::Exported {
                            watermark: obzenflow_core::event::vector_clock::VectorClock { clocks },
                        },
                    ),
                );

                if let Err(e) = ctx.system_journal.append(export_event, None).await {
                    tracing::warn!(
                        journal_error = %e,
                        "Failed to publish metrics watermark event; continuing without system journal entry"
                    );
                }

                Ok(())
            }

            MetricsAggregatorAction::PublishDrainComplete { last_event_id } => {
                // Get writer ID from context
                let system_writer_id = WriterId::from(ctx.system_id);

                // Build the drain complete event
                let mut payload = serde_json::json!({});
                if let Some(id) = last_event_id {
                    payload["last_event_id"] = serde_json::json!(id.to_string());
                }

                // Metrics aggregator publishes SystemEvent to system journal
                let drain_event = obzenflow_core::event::SystemEvent::new(
                    system_writer_id,
                    obzenflow_core::event::SystemEventType::MetricsCoordination(
                        obzenflow_core::event::MetricsCoordinationEvent::Drained,
                    ),
                );

                // Publish to system journal
                ctx.system_journal
                    .append(drain_event, None)
                    .await
                    .map(|_| ())
                    .map_err(|e| {
                        obzenflow_fsm::FsmError::HandlerError(format!(
                            "Failed to publish drain complete event: {e}"
                        ))
                    })?;

                tracing::info!(
                    "Published metrics drain complete event (last_event_id={:?})",
                    last_event_id
                );
                Ok(())
            }
        }
    }
}

/// Type alias for the metrics FSM
pub type MetricsAggregatorFsm = StateMachine<
    MetricsAggregatorState,
    MetricsAggregatorEvent,
    MetricsAggregatorContext,
    MetricsAggregatorAction,
>;

/// Build the metrics aggregator FSM with lifecycle transitions only
pub fn build_metrics_aggregator_fsm() -> MetricsAggregatorFsm {
    fsm! {
        state:   MetricsAggregatorState;
        event:   MetricsAggregatorEvent;
        context: MetricsAggregatorContext;
        action:  MetricsAggregatorAction;
        initial: MetricsAggregatorState::Initializing;

        state MetricsAggregatorState::Initializing {
            on MetricsAggregatorEvent::StartRunning => |_state: &MetricsAggregatorState, _event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                Box::pin(async move {
                    Ok(Transition {
                        next_state: MetricsAggregatorState::Running,
                        actions: vec![MetricsAggregatorAction::Initialize],
                    })
                })
            };

            on MetricsAggregatorEvent::Error => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                let event = event.clone();
                Box::pin(async move {
                    let error = match event {
                        MetricsAggregatorEvent::Error(err) => err.clone(),
                        _ => {
                            return Err(obzenflow_fsm::FsmError::HandlerError(
                                "Invalid event for Error handler".to_string(),
                            ));
                        }
                    };
                    tracing::error!(error = %error, "Metrics aggregator encountered error");
                    Ok(Transition {
                        next_state: MetricsAggregatorState::Failed { error },
                        actions: vec![],
                    })
                })
            };
        }

            state MetricsAggregatorState::Running {
            on MetricsAggregatorEvent::StartDraining => |_state: &MetricsAggregatorState, _event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                Box::pin(async move {
                    Ok(Transition {
                        next_state: MetricsAggregatorState::Draining,
                        actions: vec![],
                    })
                })
            };

                on MetricsAggregatorEvent::ProcessSystemEvent => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, ctx: &mut MetricsAggregatorContext| {
                    let event = event.clone();
                    let last_event_id = ctx.metrics_store.last_event_id;
                    Box::pin(async move {
                        match event {
                            MetricsAggregatorEvent::ProcessSystemEvent { envelope } => {
                                let pipeline_event = match &envelope.event.event {
                                    obzenflow_core::event::SystemEventType::PipelineLifecycle(event) => {
                                        Some(event)
                                    }
                                    _ => None,
                                };

                                let should_drain = matches!(
                                    pipeline_event,
                                    Some(
                                        obzenflow_core::event::PipelineLifecycleEvent::Draining { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::AllStagesCompleted { .. }
                                    )
                                );

                                let should_finalize = matches!(
                                    pipeline_event,
                                    Some(
                                        obzenflow_core::event::PipelineLifecycleEvent::Completed { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::Failed { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::Drained
                                    )
                                );

                                if should_finalize {
                                    let publish_last_event_id = last_event_id;
                                    return Ok(Transition {
                                        next_state: MetricsAggregatorState::Drained { last_event_id },
                                        actions: vec![
                                            MetricsAggregatorAction::ProcessSystemEvent {
                                                envelope: envelope.clone(),
                                            },
                                            MetricsAggregatorAction::ExportMetrics,
                                            MetricsAggregatorAction::PublishDrainComplete {
                                                last_event_id: publish_last_event_id,
                                            },
                                        ],
                                    });
                                }

                                if should_drain {
                                    return Ok(Transition {
                                        next_state: MetricsAggregatorState::Draining,
                                        actions: vec![
                                            MetricsAggregatorAction::ProcessSystemEvent {
                                                envelope: envelope.clone(),
                                            },
                                            MetricsAggregatorAction::ExportMetrics,
                                        ],
                                    });
                                }

                                Ok(Transition {
                                    next_state: MetricsAggregatorState::Running,
                                    actions: vec![MetricsAggregatorAction::ProcessSystemEvent {
                                        envelope: envelope.clone(),
                                    }],
                                })
                            }
                            _ => Err(obzenflow_fsm::FsmError::HandlerError(
                                "Invalid event for ProcessSystemEvent handler".to_string(),
                            )),
                        }
                })
            };

            on MetricsAggregatorEvent::ProcessBatch => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                let event = event.clone();
                Box::pin(async move {
                    match event {
                        MetricsAggregatorEvent::ProcessBatch { events } => {
                            let actions = events
                                .iter()
                                .cloned()
                                .map(|envelope| MetricsAggregatorAction::UpdateMetrics {
                                    envelope: Box::new(envelope),
                                })
                                .collect::<Vec<_>>();
                            Ok(Transition {
                                next_state: MetricsAggregatorState::Running,
                                actions,
                            })
                        }
                        _ => Err(obzenflow_fsm::FsmError::HandlerError(
                            "Invalid event for ProcessBatch handler".to_string(),
                        )),
                    }
                })
            };

            on MetricsAggregatorEvent::ExportMetrics => |_state: &MetricsAggregatorState, _event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                Box::pin(async move {
                    Ok(Transition {
                        next_state: MetricsAggregatorState::Running,
                        actions: vec![MetricsAggregatorAction::ExportMetrics],
                    })
                })
            };

            on MetricsAggregatorEvent::Error => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                let event = event.clone();
                Box::pin(async move {
                    match event {
                        MetricsAggregatorEvent::Error(error) => Ok(Transition {
                            next_state: MetricsAggregatorState::Failed {
                                error: error.clone(),
                            },
                            actions: vec![],
                        }),
                        _ => Err(obzenflow_fsm::FsmError::HandlerError(
                            "Invalid event for Error handler".to_string(),
                        )),
                    }
                })
            };
        }

            state MetricsAggregatorState::Draining {
            on MetricsAggregatorEvent::FlowTerminal => |_state: &MetricsAggregatorState, _event: &MetricsAggregatorEvent, ctx: &mut MetricsAggregatorContext| {
                Box::pin(async move {
                    let last_event_id = ctx.metrics_store.last_event_id;
                    let publish_last_event_id = last_event_id;
                    Ok(Transition {
                        next_state: MetricsAggregatorState::Drained { last_event_id },
                        actions: vec![
                            MetricsAggregatorAction::ExportMetrics,
                            MetricsAggregatorAction::PublishDrainComplete {
                                last_event_id: publish_last_event_id,
                            },
                        ],
                    })
                })
            };

                on MetricsAggregatorEvent::ProcessSystemEvent => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, ctx: &mut MetricsAggregatorContext| {
                    let event = event.clone();
                    let last_event_id = ctx.metrics_store.last_event_id;
                    Box::pin(async move {
                        match event {
                            MetricsAggregatorEvent::ProcessSystemEvent { envelope } => {
                                let pipeline_event = match &envelope.event.event {
                                    obzenflow_core::event::SystemEventType::PipelineLifecycle(event) => {
                                        Some(event)
                                    }
                                    _ => None,
                                };

                                let should_export = matches!(
                                    pipeline_event,
                                    Some(
                                        obzenflow_core::event::PipelineLifecycleEvent::Draining { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::AllStagesCompleted { .. }
                                    )
                                );

                                let should_finalize = matches!(
                                    pipeline_event,
                                    Some(
                                        obzenflow_core::event::PipelineLifecycleEvent::Completed { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::Failed { .. }
                                            | obzenflow_core::event::PipelineLifecycleEvent::Drained
                                    )
                                );

                                if should_finalize {
                                    let publish_last_event_id = last_event_id;
                                    return Ok(Transition {
                                        next_state: MetricsAggregatorState::Drained { last_event_id },
                                        actions: vec![
                                            MetricsAggregatorAction::ProcessSystemEvent {
                                                envelope: envelope.clone(),
                                            },
                                            MetricsAggregatorAction::ExportMetrics,
                                            MetricsAggregatorAction::PublishDrainComplete {
                                                last_event_id: publish_last_event_id,
                                            },
                                        ],
                                    });
                                }

                                let mut actions = vec![MetricsAggregatorAction::ProcessSystemEvent {
                                    envelope: envelope.clone(),
                                }];
                                if should_export {
                                    actions.push(MetricsAggregatorAction::ExportMetrics);
                                }

                                Ok(Transition {
                                    next_state: MetricsAggregatorState::Draining,
                                    actions,
                                })
                            }
                            _ => Err(obzenflow_fsm::FsmError::HandlerError(
                                "Invalid event for ProcessSystemEvent handler in Draining".to_string(),
                            )),
                        }
                })
            };

            on MetricsAggregatorEvent::ProcessBatch => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                let event = event.clone();
                Box::pin(async move {
                    match event {
                        MetricsAggregatorEvent::ProcessBatch { events } => {
                            let actions = events
                                .iter()
                                .cloned()
                                .map(|envelope| MetricsAggregatorAction::UpdateMetrics {
                                    envelope: Box::new(envelope),
                                })
                                .collect::<Vec<_>>();
                            Ok(Transition {
                                next_state: MetricsAggregatorState::Draining,
                                actions,
                            })
                        }
                        _ => Err(obzenflow_fsm::FsmError::HandlerError(
                            "Invalid event for ProcessBatch handler in Draining".to_string(),
                        )),
                    }
                })
            };

            on MetricsAggregatorEvent::Error => |_state: &MetricsAggregatorState, event: &MetricsAggregatorEvent, _ctx: &mut MetricsAggregatorContext| {
                let event = event.clone();
                Box::pin(async move {
                    match event {
                        MetricsAggregatorEvent::Error(error) => Ok(Transition {
                            next_state: MetricsAggregatorState::Failed {
                                error: error.clone(),
                            },
                            actions: vec![],
                        }),
                        _ => Err(obzenflow_fsm::FsmError::HandlerError(
                            "Invalid event for Error handler".to_string(),
                        )),
                    }
                })
            };
        }

        // Drained state (terminal) - no transitions
        state MetricsAggregatorState::Drained { }

        // Failed state (terminal) - no transitions
        state MetricsAggregatorState::Failed { }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use obzenflow_core::event::context::StageType;
    use obzenflow_core::event::payloads::correlation_payload::CorrelationPayload;
    use obzenflow_core::event::payloads::delivery_payload::{DeliveryMethod, DeliveryPayload};
    use obzenflow_core::event::status::processing_status::ErrorKind;
    use obzenflow_core::event::ChainEventFactory;
    use obzenflow_core::event::CorrelationId;
    use obzenflow_core::event::JournalEvent;
    use obzenflow_core::journal::journal_error::JournalError;
    use obzenflow_core::journal::journal_owner::JournalOwner;
    use obzenflow_core::journal::journal_reader::JournalReader;
    use obzenflow_core::journal::Journal;
    use obzenflow_core::metrics::StageMetadata;
    use std::marker::PhantomData;

    #[tokio::test]
    async fn test_delivery_event_preserves_correlation() {
        // Create a test event with correlation
        let writer_id = WriterId::from(StageId::new());
        let correlation_id = CorrelationId::new();
        let mut event = ChainEventFactory::data_event(
            writer_id,
            "test.event",
            serde_json::json!({"data": "test"}),
        );
        event.correlation_id = Some(correlation_id);
        event.correlation_payload = Some(CorrelationPayload::new("test_source", event.id));

        // Simulate what the sink supervisor does when creating a delivery event
        let payload = DeliveryPayload::success("test_sink", DeliveryMethod::Noop, Some(1));
        let delivery_event =
            ChainEventFactory::delivery_event(writer_id, payload).with_correlation_from(&event);

        // Verify correlation is preserved
        assert_eq!(delivery_event.correlation_id, Some(correlation_id));
        assert!(delivery_event.correlation_payload.is_some());
        assert_eq!(
            delivery_event
                .correlation_payload
                .as_ref()
                .unwrap()
                .entry_stage,
            "test_source"
        );
    }

    #[test]
    fn build_app_metrics_snapshot_uses_errors_by_kind_from_store() {
        let stage_id = StageId::new();

        // Seed MetricsStore with a single stage entry.
        let mut store = MetricsStore::default();
        let errors_by_kind = HashMap::from([(ErrorKind::Domain, 2), (ErrorKind::Remote, 1)]);
        let stage_metrics = StageMetrics {
            errors_by_kind,
            latest_events_processed_total: Some(42),
            latest_errors_total: Some(3),
            event_loops_total: 10,
            event_loops_with_work_total: 7,
            ..Default::default()
        };
        store.stage_metrics.insert(stage_id, stage_metrics);

        // Minimal stage metadata so flow aggregation can classify the stage.
        let mut stage_metadata = std::collections::HashMap::new();
        stage_metadata.insert(
            stage_id,
            StageMetadata {
                name: "test_stage".to_string(),
                stage_type: StageType::Sink,
                reference_mode: None,
                flow_name: "test_flow".to_string(),
                flow_id: None,
            },
        );

        // Local NoopJournal implementation for the system_journal field; it is never used
        // by build_app_metrics_snapshot but satisfies the context type.
        struct NoopJournal<T: JournalEvent> {
            id: obzenflow_core::id::JournalId,
            owner: Option<JournalOwner>,
            _marker: PhantomData<T>,
        }

        impl<T: JournalEvent> NoopJournal<T> {
            fn new(owner: JournalOwner) -> Self {
                Self {
                    id: obzenflow_core::id::JournalId::new(),
                    owner: Some(owner),
                    _marker: PhantomData,
                }
            }
        }

        struct NoopReader;

        #[async_trait]
        impl<T: JournalEvent + 'static> Journal<T> for NoopJournal<T> {
            fn id(&self) -> &obzenflow_core::id::JournalId {
                &self.id
            }

            fn owner(&self) -> Option<&JournalOwner> {
                self.owner.as_ref()
            }

            async fn append(
                &self,
                _event: T,
                _parent: Option<&obzenflow_core::EventEnvelope<T>>,
            ) -> Result<obzenflow_core::EventEnvelope<T>, JournalError> {
                Err(JournalError::Implementation {
                    message: "noop journal".to_string(),
                    source: "noop".into(),
                })
            }

            async fn read_causally_ordered(
                &self,
            ) -> Result<Vec<obzenflow_core::EventEnvelope<T>>, JournalError> {
                Ok(Vec::new())
            }

            async fn read_causally_after(
                &self,
                _after_event_id: &obzenflow_core::EventId,
            ) -> Result<Vec<obzenflow_core::EventEnvelope<T>>, JournalError> {
                Ok(Vec::new())
            }

            async fn read_event(
                &self,
                _event_id: &obzenflow_core::EventId,
            ) -> Result<Option<obzenflow_core::EventEnvelope<T>>, JournalError> {
                Ok(None)
            }

            async fn reader(&self) -> Result<Box<dyn JournalReader<T>>, JournalError> {
                Ok(Box::new(NoopReader))
            }

            async fn reader_from(
                &self,
                _position: u64,
            ) -> Result<Box<dyn JournalReader<T>>, JournalError> {
                Ok(Box::new(NoopReader))
            }

            async fn read_last_n(
                &self,
                _count: usize,
            ) -> Result<Vec<obzenflow_core::EventEnvelope<T>>, JournalError> {
                Ok(Vec::new())
            }
        }

        #[async_trait]
        impl<T: JournalEvent + 'static> JournalReader<T> for NoopReader {
            async fn next(
                &mut self,
            ) -> Result<Option<obzenflow_core::EventEnvelope<T>>, JournalError> {
                Ok(None)
            }

            async fn skip(&mut self, _n: u64) -> Result<u64, JournalError> {
                Ok(0)
            }

            fn position(&self) -> u64 {
                0
            }

            fn is_at_end(&self) -> bool {
                true
            }
        }

        // Build a context with only the fields required by build_app_metrics_snapshot.
        let ctx = MetricsAggregatorContext {
            system_journal: Arc::new(NoopJournal::<obzenflow_core::event::SystemEvent>::new(
                JournalOwner::system(obzenflow_core::SystemId::new()),
            )),
            stage_data_journals: HashMap::new(),
            stage_error_journals: HashMap::new(),
            backpressure_registry: None,
            include_error_journals: true,
            exporter: None,
            metrics_store: store,
            export_interval_secs: 10,
            system_id: obzenflow_core::SystemId::new(),
            stage_metadata,
        };

        let snapshot = ctx.build_app_metrics_snapshot();

        // Stage-level totals should reflect the seeded store.
        assert_eq!(snapshot.error_counts.get(&stage_id), Some(&3));
        let by_kind = snapshot
            .error_counts_by_kind
            .get(&stage_id)
            .expect("per-kind breakdown should be present");
        assert_eq!(by_kind.get(&ErrorKind::Domain), Some(&2));
        assert_eq!(by_kind.get(&ErrorKind::Remote), Some(&1));

        // Stage metadata should be carried through.
        assert!(snapshot.stage_metadata.contains_key(&stage_id));
    }
}