saddle_observability/

chain.rs

use std::{error::Error, fmt, time::Instant};

use saddle_core::{CallContext, ErrorKind, SaddleError};
use serde_json::{Value, json};

use crate::{EventLevel, Observer, OutputStage, logger::LogRecord};

const MAX_IDENTITY_BYTES: usize = 256;

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ChainFieldError {
    Empty,
    TooLong,
    ControlCharacter,
    UnsafeAuthority,
    ZeroAttempt,
}

impl fmt::Display for ChainFieldError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Empty => "observability identity must not be empty",
            Self::TooLong => "observability identity exceeds 256 bytes",
            Self::ControlCharacter => "observability identity contains a control character",
            Self::UnsafeAuthority => "outbound authority must be a credential-free target label",
            Self::ZeroAttempt => "attempt must be greater than zero",
        })
    }
}

impl Error for ChainFieldError {}

macro_rules! safe_identity {
    ($name:ident) => {
        #[derive(Clone, Debug, Eq, PartialEq)]
        pub struct $name(String);

        impl $name {
            pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
                validate_identity(value.into()).map(Self)
            }

            pub fn as_str(&self) -> &str {
                &self.0
            }
        }
    };
}

safe_identity!(RequestIdentity);
safe_identity!(RouteIdentity);
safe_identity!(BottleneckIdentity);
safe_identity!(RejectReason);

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OutboundAuthority(String);

impl OutboundAuthority {
    pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
        let value = validate_identity(value.into())?;
        if value.contains(['@', '?', '#', '/', '\\']) || value.contains("://") {
            return Err(ChainFieldError::UnsafeAuthority);
        }
        Ok(Self(value))
    }
}

fn validate_identity(value: String) -> Result<String, ChainFieldError> {
    if value.is_empty() {
        return Err(ChainFieldError::Empty);
    }
    if value.len() > MAX_IDENTITY_BYTES {
        return Err(ChainFieldError::TooLong);
    }
    if value.chars().any(char::is_control) {
        return Err(ChainFieldError::ControlCharacter);
    }
    Ok(value)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Stage {
    Ingress,
    Admission,
    Handler,
    Database,
    ProfuseContract,
    Response,
    ResourceFinalization,
}

impl Stage {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Ingress => "ingress",
            Self::Admission => "admission",
            Self::Handler => "handler",
            Self::Database => "database",
            Self::ProfuseContract => "profuse_contract",
            Self::Response => "response",
            Self::ResourceFinalization => "resource_finalization",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StageOutcome {
    Success,
    Failure,
    Rejected,
    Cancelled,
}

/// DB owning-driver facts, not an inference from HTTP or connection disposal.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransactionOutcome {
    /// COMMIT acknowledgement was received, even if the client later disconnects.
    Committed,
    /// No possible commit was entered. This does not prove rollback acknowledgement.
    Rejected,
    /// COMMIT may have executed but its acknowledgement is unavailable.
    Unknown,
}

impl TransactionOutcome {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Committed => "Committed",
            Self::Rejected => "Rejected",
            Self::Unknown => "Unknown",
        }
    }
}

/// Same-request/scope carrier checked by Core and transported by Runtime.
/// It is consumed once; it is not proof of transaction outcome or log durability.
pub type TransactionTerminalObservation =
    saddle_core::DbScopeTerminalObservation<(Observer, CallContext, EventContext)>;

impl StageOutcome {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Success => "success",
            Self::Failure => "failure",
            Self::Rejected => "rejected",
            Self::Cancelled => "cancelled",
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct EventContext {
    request: RequestIdentity,
    route: RouteIdentity,
    attempt: u32,
}

impl EventContext {
    pub(crate) fn diagnostic_attempt(&self) -> u32 {
        self.attempt
    }
    pub(crate) fn diagnostic_request(&self) -> &str {
        self.request.as_str()
    }
    pub(crate) fn diagnostic_route(&self) -> &str {
        self.route.as_str()
    }
    pub fn new(
        request: RequestIdentity,
        route: RouteIdentity,
        attempt: u32,
    ) -> Result<Self, ChainFieldError> {
        if attempt == 0 {
            return Err(ChainFieldError::ZeroAttempt);
        }
        Ok(Self {
            request,
            route,
            attempt,
        })
    }
}

pub struct ActiveStage {
    observer: Observer,
    context: CallContext,
    parent_span: String,
    stage: Stage,
    request: RequestIdentity,
    route: RouteIdentity,
    attempt: u32,
    started_at: Instant,
    finished: bool,
}

impl Observer {
    /// DB owning driver calls once before returning its result/physical receipt.
    /// Uses the carrier's observer/context, with no caller-supplied scope or sink.
    /// Queue loss preserves the DB result; a missing log remains NOT_PROVEN.
    ///
    /// ```compile_fail
    /// use saddle_observability::{Observer, TransactionOutcome, TransactionTerminalObservation};
    /// fn replay(observation: TransactionTerminalObservation) {
    ///     Observer::record_transaction_terminal(observation, TransactionOutcome::Committed);
    ///     Observer::record_transaction_terminal(observation, TransactionOutcome::Rejected);
    /// }
    /// ```
    pub fn record_transaction_terminal(
        observation: TransactionTerminalObservation,
        outcome: TransactionOutcome,
    ) {
        let ((observer, context, event_context), scope) = observation.into_log_parts();
        let mut record = base_record(
            &context,
            if outcome == TransactionOutcome::Committed {
                EventLevel::Info
            } else {
                EventLevel::Error
            },
            "database_transaction_terminal",
            "database",
        );
        add_event_context(&mut record, &event_context);
        // Core's concrete Serialize-only payload has exactly one numeric field.
        // A serialization failure drops this attempt; it never changes DB outcome.
        let Ok(Value::Object(fields)) = serde_json::to_value(scope) else {
            return;
        };
        record.data.extend(fields);
        record.data.insert(
            "transaction_outcome".into(),
            Value::String(outcome.as_str().into()),
        );
        record
            .data
            .insert("outcome".into(), Value::String(outcome.as_str().into()));
        observer.emit(record);
    }

    pub fn record_lifecycle_timeout(
        &self,
        application: &str,
        stage: LifecycleTimeoutStage,
        elapsed_ms: u64,
    ) {
        let Ok((call, _)) = self.start_external_call_checked(
            application,
            "runtime",
            "lifecycle",
            stage.as_str(),
            None,
        ) else {
            return;
        };
        let mut record = base_record(
            call.context(),
            EventLevel::Error,
            "framework.lifecycle.timeout",
            "lifecycle",
        );
        record
            .data
            .insert("timeout_stage".into(), Value::String(stage.as_str().into()));
        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
        record
            .data
            .insert("outcome".into(), Value::String("timeout".into()));
        self.emit(record);
        call.fail(&SaddleError::new(
            ErrorKind::Infrastructure,
            "runtime.lifecycle_timeout",
            "managed lifecycle deadline elapsed",
        ));
    }

    pub fn start_stage(
        &self,
        parent: &CallContext,
        stage: Stage,
        event_context: EventContext,
    ) -> ActiveStage {
        let context = CallContext::new(
            parent.application().clone(),
            parent.module().clone(),
            parent.service().clone(),
            parent.operation().clone(),
            parent.trace_id(),
            self.new_span_id(),
        )
        .with_trace_correlation_id(parent.trace_correlation_id().clone())
        .with_rpc_correlation_id(parent.rpc_correlation_id().cloned());
        let active = ActiveStage {
            observer: self.clone(),
            context,
            parent_span: parent.span_id().to_string(),
            stage,
            request: event_context.request,
            route: event_context.route,
            attempt: event_context.attempt,
            started_at: Instant::now(),
            finished: false,
        };
        active.emit_started();
        active
    }

    pub fn record_capacity(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        value: CapacityObservation,
    ) {
        self.inner
            .metrics
            .capacity(value.dimension, value.used, value.reject_reason.is_some());
        let mut record = base_record(context, EventLevel::Info, "framework.capacity", "admission");
        add_event_context(&mut record, event_context);
        if let Some(dimension) = value.dimension {
            record.data.insert(
                "capacity_dimension".into(),
                Value::String(dimension.as_str().into()),
            );
        }
        record.data.insert("budget".into(), json!(value.budget));
        record.data.insert("limit".into(), json!(value.limit));
        record.data.insert("used".into(), json!(value.used));
        record
            .data
            .insert("elapsed_ms".into(), json!(value.elapsed_ms));
        record
            .data
            .insert("bottleneck".into(), Value::String(value.bottleneck.0));
        record.data.insert(
            "outcome".into(),
            Value::String(
                if value.reject_reason.is_some() {
                    "rejected"
                } else {
                    "accepted"
                }
                .into(),
            ),
        );
        if let Some(reason) = value.reject_reason {
            record
                .data
                .insert("reject_reason".into(), Value::String(reason.0));
        }
        self.emit(record);
    }

    /// Update the existing fixed-label counter without allocating or emitting a
    /// second log. The DB owning terminal path calls this exactly once with its
    /// actual physical disposition, independently of operation outcome.
    pub fn observe_database_disposition(&self, disposition: DatabaseDisposition) {
        self.inner.metrics.database(disposition);
    }

    pub fn record_database_disposition(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        disposition: DatabaseDisposition,
    ) {
        self.observe_database_disposition(disposition);
        let mut record = base_record(
            context,
            EventLevel::Info,
            "framework.database.disposition",
            "database",
        );
        add_event_context(&mut record, event_context);
        record.data.insert(
            "db_disposition".into(),
            Value::String(disposition.as_str().into()),
        );
        record
            .data
            .insert("outcome".into(), Value::String(disposition.as_str().into()));
        self.emit(record);
    }

    pub fn record_resource_finalization(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        disposition: DatabaseDisposition,
        elapsed_ms: u64,
    ) {
        let mut record = base_record(
            context,
            EventLevel::Info,
            "framework.resource.finalized",
            "resource_finalization",
        );
        add_event_context(&mut record, event_context);
        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
        record
            .data
            .insert("credit".into(), Value::String("released".into()));
        record.data.insert(
            "db_disposition".into(),
            Value::String(disposition.as_str().into()),
        );
        record
            .data
            .insert("outcome".into(), Value::String("success".into()));
        self.emit(record);
    }

    pub fn record_outbound(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        observation: OutboundObservation,
    ) {
        self.inner.metrics.outbound(observation.result);
        let mut record = base_record(
            context,
            EventLevel::Info,
            "framework.outbound",
            "profuse_contract",
        );
        add_event_context(&mut record, event_context);
        record
            .data
            .insert("zone".into(), Value::String(observation.zone.0));
        record
            .data
            .insert("authority".into(), Value::String(observation.authority.0));
        record.data.insert(
            "outbound_result".into(),
            Value::String(observation.result.as_str().into()),
        );
        record.data.insert(
            "outcome".into(),
            Value::String(observation.result.as_str().into()),
        );
        self.emit(record);
    }

    pub fn record_lifecycle(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        state: LifecycleState,
        health: Health,
    ) {
        self.inner.metrics.lifecycle(state, health);
        let mut record = base_record(
            context,
            if health == Health::Healthy {
                EventLevel::Info
            } else {
                EventLevel::Error
            },
            "framework.lifecycle",
            "resource_finalization",
        );
        add_event_context(&mut record, event_context);
        record
            .data
            .insert("lifecycle".into(), Value::String(state.as_str().into()));
        record
            .data
            .insert("health".into(), Value::String(health.as_str().into()));
        record
            .data
            .insert("outcome".into(), Value::String(health.as_str().into()));
        self.emit(record);
    }

    pub fn record_logger_health(
        &self,
        context: &CallContext,
        event_context: &EventContext,
        dropped: u64,
        failure: Option<OutputStage>,
    ) {
        self.inner.metrics.logger_output(failure.is_some());
        let mut record = base_record(
            context,
            if failure.is_some() || dropped > 0 {
                EventLevel::Error
            } else {
                EventLevel::Info
            },
            "framework.logger.health",
            "logger",
        );
        add_event_context(&mut record, event_context);
        record.data.insert("logger_dropped".into(), json!(dropped));
        record.data.insert(
            "logger_health".into(),
            Value::String(
                if failure.is_some() {
                    "output_failed"
                } else {
                    "healthy"
                }
                .into(),
            ),
        );
        record.data.insert(
            "outcome".into(),
            Value::String(
                if failure.is_some() {
                    "failure"
                } else {
                    "success"
                }
                .into(),
            ),
        );
        if let Some(stage) = failure {
            record.data.insert(
                "output_failed".into(),
                Value::String(format!("{stage:?}").to_ascii_lowercase()),
            );
        }
        self.emit(record);
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LifecycleTimeoutStage {
    ComponentStart,
    RequestDrain,
    ComponentShutdown,
    PostDriverFinalization,
}

impl LifecycleTimeoutStage {
    const fn as_str(self) -> &'static str {
        match self {
            Self::ComponentStart => "component_start",
            Self::RequestDrain => "request_drain",
            Self::ComponentShutdown => "component_shutdown",
            Self::PostDriverFinalization => "post_driver_finalization",
        }
    }
}

impl ActiveStage {
    /// Finish a non-technical result without inventing a source receipt. Only
    /// Succeeded/Rejected are accepted; all other axes return the untouched stage
    /// for the caller to finish through its actual diagnostic receipt path.
    #[allow(clippy::result_large_err)] // Returning the owner must not allocate.
    pub fn finish_nonfailure_bounded(
        mut self,
        output: Option<&crate::EmergencyDiagnosticHandle>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
    ) -> Result<crate::DiagnosticSubmission, Self> {
        if !matches!(
            axes.operation,
            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
        ) {
            return Err(self);
        }
        let elapsed = self.finish_bounded_metrics(axes);
        Ok(
            output.map_or(crate::DiagnosticSubmission::OutputUnavailable, |output| {
                output.submit_boundary_parts(
                    None,
                    axes,
                    Some((&self.context, self.request.as_str(), self.route.as_str())),
                    Some((self.stage.as_str(), elapsed)),
                )
            }),
        )
    }

    /// Ends this stage once using the source receipt's original context. Missing
    /// output changes delivery status only, never the operation or Drop outcome.
    pub fn finish_request_failure<D>(
        mut self,
        output: Option<&crate::EmergencyDiagnosticHandle>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        reference: &crate::RequestBoundaryReference<D>,
    ) -> crate::BoundaryDiagnosticDelivery {
        let elapsed = self.finish_bounded_metrics(axes);
        reference.record_stage(output, axes, self.stage.as_str(), elapsed)
    }

    fn finish_bounded_metrics(&mut self, axes: &saddle_core::DiagnosticOutcomeAxes) -> u64 {
        self.finished = true;
        let elapsed = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        use saddle_core::OperationOutcome;
        let metric = match axes.operation {
            OperationOutcome::Succeeded => Some(StageOutcome::Success),
            OperationOutcome::Rejected => Some(StageOutcome::Rejected),
            OperationOutcome::Failed | OperationOutcome::Panicked => Some(StageOutcome::Failure),
            OperationOutcome::Cancelled | OperationOutcome::TimedOut => {
                Some(StageOutcome::Cancelled)
            }
            OperationOutcome::Unknown => None,
        };
        if let Some(metric) = metric {
            self.observer
                .inner
                .metrics
                .stage_finished(self.stage, metric, elapsed);
        }
        elapsed
    }
    /// Finish from independent source-owned axes without constructing SaddleError
    /// or allocating a LogRecord. Delivery failure never retries the stage.
    pub fn finish_bounded(
        self,
        output: &crate::EmergencyDiagnosticHandle,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        diagnostic: Option<&saddle_core::BoundedDiagnostic>,
    ) -> crate::DiagnosticSubmission {
        self.finish_bounded_reference(output, axes, diagnostic.map(|d| d.occurrence()))
    }
    /// Finish with correlation only; never repeats the source detail.
    pub fn finish_bounded_reference(
        mut self,
        output: &crate::EmergencyDiagnosticHandle,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        occurrence: Option<saddle_core::DiagnosticOccurrence>,
    ) -> crate::DiagnosticSubmission {
        let elapsed = self.finish_bounded_metrics(axes);
        output.submit_boundary_parts(
            occurrence,
            axes,
            Some((&self.context, self.request.as_str(), self.route.as_str())),
            Some((self.stage.as_str(), elapsed)),
        )
    }
    pub fn context(&self) -> &CallContext {
        &self.context
    }
    pub fn succeed(mut self) {
        self.finish(StageOutcome::Success, None);
    }
    pub fn reject(mut self, error: &SaddleError) {
        self.finish(StageOutcome::Rejected, Some(error));
    }
    pub fn fail(mut self, error: &SaddleError) {
        self.finish(StageOutcome::Failure, Some(error));
    }

    fn emit_started(&self) {
        let mut record = self.record(EventLevel::Info, "framework.stage.started");
        record
            .data
            .insert("outcome".into(), Value::String("started".into()));
        self.observer.emit(record);
    }

    fn finish(&mut self, outcome: StageOutcome, error: Option<&SaddleError>) {
        if self.finished {
            return;
        }
        let mut record = self.record(
            if outcome == StageOutcome::Success {
                EventLevel::Info
            } else {
                EventLevel::Error
            },
            "framework.stage.finished",
        );
        record
            .data
            .insert("outcome".into(), Value::String(outcome.as_str().into()));
        let elapsed_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
        if let Some(error) = error {
            record
                .data
                .insert("error_code".into(), Value::String(error.code().to_owned()));
            record.data.insert(
                "error_kind".into(),
                Value::String(error_kind(error.kind()).into()),
            );
        }
        self.observer
            .inner
            .metrics
            .stage_finished(self.stage, outcome, elapsed_ms);
        self.observer.emit(record);
        self.finished = true;
    }

    fn record(&self, level: EventLevel, event: &'static str) -> LogRecord {
        let mut record = base_record(&self.context, level, event, self.stage.as_str());
        record.parent = Some(self.parent_span.clone());
        record.parent_span_id = Some(self.parent_span.clone());
        record.data.insert(
            "request_identity".into(),
            Value::String(self.request.0.clone()),
        );
        record
            .data
            .insert("route".into(), Value::String(self.route.0.clone()));
        record.data.insert("attempt".into(), json!(self.attempt));
        record.data.insert("elapsed_ms".into(), json!(0));
        record
            .data
            .insert("error_code".into(), Value::String("none".into()));
        record
    }
}

impl Drop for ActiveStage {
    fn drop(&mut self) {
        self.finish(StageOutcome::Cancelled, None);
    }
}

pub struct CapacityObservation {
    dimension: Option<CapacityDimension>,
    budget: u64,
    limit: u64,
    used: u64,
    bottleneck: BottleneckIdentity,
    reject_reason: Option<RejectReason>,
    elapsed_ms: u64,
}

impl CapacityObservation {
    pub fn accepted(budget: u64, limit: u64, used: u64, bottleneck: BottleneckIdentity) -> Self {
        Self {
            dimension: None,
            budget,
            limit,
            used,
            bottleneck,
            reject_reason: None,
            elapsed_ms: 0,
        }
    }
    pub fn rejected(
        budget: u64,
        limit: u64,
        used: u64,
        bottleneck: BottleneckIdentity,
        reason: RejectReason,
    ) -> Self {
        Self {
            dimension: None,
            budget,
            limit,
            used,
            bottleneck,
            reject_reason: Some(reason),
            elapsed_ms: 0,
        }
    }

    pub fn accepted_dimension(
        dimension: CapacityDimension,
        budget: u64,
        limit: u64,
        used: u64,
        bottleneck: BottleneckIdentity,
        elapsed_ms: u64,
    ) -> Self {
        Self {
            dimension: Some(dimension),
            budget,
            limit,
            used,
            bottleneck,
            reject_reason: None,
            elapsed_ms,
        }
    }

    pub fn rejected_dimension(
        dimension: CapacityDimension,
        budget: u64,
        limit: u64,
        used: u64,
        bottleneck: BottleneckIdentity,
        reason: RejectReason,
        elapsed_ms: u64,
    ) -> Self {
        Self {
            dimension: Some(dimension),
            budget,
            limit,
            used,
            bottleneck,
            reject_reason: Some(reason),
            elapsed_ms,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CapacityDimension {
    Cpu,
    Memory,
    Database,
    ProfuseContract,
}

impl CapacityDimension {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Cpu => "cpu",
            Self::Memory => "memory",
            Self::Database => "database",
            Self::ProfuseContract => "profuse_contract",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DatabaseDisposition {
    NotUsed,
    Returned,
    Discarded,
}
impl DatabaseDisposition {
    const fn as_str(self) -> &'static str {
        match self {
            Self::NotUsed => "not_used",
            Self::Returned => "returned",
            Self::Discarded => "discarded",
        }
    }
}

pub struct OutboundObservation {
    zone: RouteIdentity,
    authority: OutboundAuthority,
    result: OutboundResult,
}
impl OutboundObservation {
    pub fn new(zone: RouteIdentity, authority: OutboundAuthority, result: OutboundResult) -> Self {
        Self {
            zone,
            authority,
            result,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OutboundResult {
    Success,
    Failure,
    Rejected,
    Timeout,
}
impl OutboundResult {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Success => "success",
            Self::Failure => "failure",
            Self::Rejected => "rejected",
            Self::Timeout => "timeout",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LifecycleState {
    Starting,
    Running,
    Draining,
    Stopped,
}
impl LifecycleState {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Starting => "starting",
            Self::Running => "running",
            Self::Draining => "draining",
            Self::Stopped => "stopped",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Health {
    Healthy,
    Degraded,
    Failed,
}
impl Health {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Healthy => "healthy",
            Self::Degraded => "degraded",
            Self::Failed => "failed",
        }
    }
}

fn base_record(
    context: &CallContext,
    level: EventLevel,
    event: &'static str,
    stage: &'static str,
) -> LogRecord {
    let mut record = LogRecord::new(level, event);
    record.trace_id = Some(context.trace_correlation_id().to_string());
    record.span = Some(stage.into());
    record.span_id = Some(context.span_id().to_string());
    record
        .data
        .insert("timestamp".into(), json!(record.timestamp_unix_ms));
    record
        .data
        .insert("stage".into(), Value::String(stage.into()));
    record.data.insert(
        "rpc_id".into(),
        context
            .rpc_correlation_id()
            .map(|rpc| Value::String(rpc.as_str().into()))
            .unwrap_or(Value::Null),
    );
    record.data.insert(
        "request".into(),
        Value::String(context.operation().to_string()),
    );
    record
}

fn add_event_context(record: &mut LogRecord, context: &EventContext) {
    record.data.insert(
        "request_identity".into(),
        Value::String(context.request.0.clone()),
    );
    record
        .data
        .insert("route".into(), Value::String(context.route.0.clone()));
    record.data.insert("attempt".into(), json!(context.attempt));
    record.data.insert("elapsed_ms".into(), json!(0));
    record
        .data
        .insert("error_code".into(), Value::String("none".into()));
}

const fn error_kind(kind: ErrorKind) -> &'static str {
    match kind {
        ErrorKind::InvalidArgument => "invalid_argument",
        ErrorKind::NotFound => "not_found",
        ErrorKind::Conflict => "conflict",
        ErrorKind::Business => "business",
        ErrorKind::Unavailable => "unavailable",
        ErrorKind::Infrastructure => "infrastructure",
        ErrorKind::Internal => "internal",
        _ => "unknown",
    }
}

#[cfg(test)]
mod tests {
    use std::{
        future::Future,
        io,
        sync::{Arc, Mutex},
        task::{Context, Poll, Wake, Waker},
        thread,
    };

    use super::*;
    use crate::ObserverConfig;

    #[derive(Clone, Default)]
    struct Capture(Arc<Mutex<Vec<u8>>>);

    impl io::Write for Capture {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(bytes);
            Ok(bytes.len())
        }
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    struct ThreadWaker(thread::Thread);
    impl Wake for ThreadWaker {
        fn wake(self: Arc<Self>) {
            self.0.unpark();
        }
    }

    fn block_on<T>(future: impl Future<Output = T>) -> T {
        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
        let mut context = Context::from_waker(&waker);
        let mut future = std::pin::pin!(future);
        loop {
            match future.as_mut().poll(&mut context) {
                Poll::Ready(output) => return output,
                Poll::Pending => thread::park(),
            }
        }
    }

    fn event_context(attempt: u32) -> EventContext {
        EventContext::new(
            RequestIdentity::new("request-7").unwrap(),
            RouteIdentity::new("orders.create").unwrap(),
            attempt,
        )
        .unwrap()
    }

    #[test]
    fn transaction_terminal_serial_scope_schema_and_foreign_restore() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        let (root, _) = observer
            .start_external_call_with_rpc(
                "shop",
                "entry",
                "orders",
                "create",
                Some("safe-trace"),
                saddle_core::RpcCorrelationId::new("0").unwrap(),
            )
            .unwrap();
        let (startup, issuer) =
            saddle_core::DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
        let process = startup.into_process_capability();
        let (mut request, mut execution) = issuer.issue_request().unwrap();
        let (_, foreign) = issuer.issue_request().unwrap();
        for outcome in [
            TransactionOutcome::Committed,
            TransactionOutcome::Rejected,
            TransactionOutcome::Unknown,
        ] {
            let observation = request
                .take_scope_observation(
                    &execution,
                    (observer.clone(), root.context().clone(), event_context(1)),
                )
                .ok()
                .unwrap();
            let observation = observation.bind_terminal(&foreign).err().unwrap();
            let observation = observation.bind_terminal(&execution).ok().unwrap();
            Observer::record_transaction_terminal(observation, outcome);
            assert!(request.take_scope_observation(&execution, ()).is_err());
            let physical = process.connection_returned(execution, ()).ok().unwrap();
            let (next, _, _) = saddle_core::pair_db_physical_disposition(physical, request)
                .ok()
                .unwrap()
                .into_scope_continuation();
            (request, execution) = next.into_next_scope().ok().unwrap();
        }
        block_on(observer.flush()).unwrap();
        let rows: Vec<_> = records(&capture)
            .into_iter()
            .filter(|r| r["event"] == "database_transaction_terminal")
            .collect();
        assert_eq!(rows.len(), 3);
        for (index, outcome) in ["Committed", "Rejected", "Unknown"].into_iter().enumerate() {
            let row = &rows[index];
            assert_eq!(row["transaction_scope"], index);
            assert_eq!(row["transaction_outcome"], outcome);
            assert_eq!(row["trace_id"], "safe-trace");
            assert_eq!(row["rpc_id"], "0");
            assert_eq!(row["span_id"], root.context().span_id().to_string());
            assert_ne!(row["rpc_id"], row["span_id"]);
            assert_eq!(row["request_identity"], "request-7");
            assert_eq!(row["route"], "orders.create");
            let mut keys: Vec<_> = row
                .as_object()
                .unwrap()
                .keys()
                .map(String::as_str)
                .collect();
            keys.sort_unstable();
            let mut expected = vec![
                "timestamp_unix_ms",
                "timestamp",
                "level",
                "event",
                "stage",
                "trace_id",
                "rpc_id",
                "span",
                "span_id",
                "request",
                "request_identity",
                "route",
                "attempt",
                "elapsed_ms",
                "error_code",
                "outcome",
                "transaction_scope",
                "transaction_outcome",
            ];
            expected.sort_unstable();
            assert_eq!(keys, expected); // No SQL/parameters/business error or arbitrary fields.
        }
        root.succeed();
        block_on(observer.shutdown()).unwrap();
    }

    #[test]
    fn transaction_terminal_full_queue_never_waits_for_writer() {
        use std::sync::mpsc;
        use std::time::Duration;
        struct PausedWriter {
            entered: Option<mpsc::Sender<()>>,
            release: mpsc::Receiver<()>,
        }
        impl io::Write for PausedWriter {
            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
                if let Some(entered) = self.entered.take() {
                    entered.send(()).unwrap();
                    self.release.recv().unwrap();
                }
                Ok(bytes.len())
            }
            fn flush(&mut self) -> io::Result<()> {
                Ok(())
            }
        }
        let (entered, reached) = mpsc::channel();
        let (release, gate) = mpsc::channel();
        let observer = Observer::with_writer(
            ObserverConfig { queue_capacity: 1 },
            PausedWriter {
                entered: Some(entered),
                release: gate,
            },
        )
        .unwrap();
        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
        reached.recv_timeout(Duration::from_secs(5)).unwrap();
        observer.emit(LogRecord::new(EventLevel::Info, "fill"));
        let (_, issuer) =
            saddle_core::DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
        let (mut request, execution) = issuer.issue_request().unwrap();
        let observation = request
            .take_scope_observation(
                &execution,
                (observer.clone(), root.context().clone(), event_context(1)),
            )
            .ok()
            .unwrap()
            .bind_terminal(&execution)
            .ok()
            .unwrap();
        let (done, completion) = mpsc::channel();
        let emitter = thread::spawn(move || {
            Observer::record_transaction_terminal(observation, TransactionOutcome::Unknown);
            done.send(()).unwrap();
        });
        let result = completion.recv_timeout(Duration::from_secs(2));
        let dropped = observer.dropped_events();
        release.send(()).unwrap(); // Release even if a regression blocked the emitter.
        emitter.join().unwrap();
        assert!(result.is_ok());
        assert_eq!(dropped, 1);
        let _ = block_on(observer.flush());
        root.succeed();
        let _ = block_on(observer.shutdown());
    }

    fn records(capture: &Capture) -> Vec<Value> {
        String::from_utf8(capture.0.lock().unwrap().clone())
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect()
    }

    #[test]
    fn lifecycle_timeout_is_typed_and_contains_no_application_payload() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        observer.record_lifecycle_timeout(
            "profusegw",
            LifecycleTimeoutStage::ComponentShutdown,
            30_000,
        );
        block_on(observer.flush()).unwrap();

        let records = records(&capture);
        let timeout = records
            .iter()
            .find(|record| record["event"] == "framework.lifecycle.timeout")
            .unwrap();
        assert_eq!(timeout["timeout_stage"], "component_shutdown");
        assert_eq!(timeout["elapsed_ms"], 30_000);
        assert_eq!(timeout["outcome"], "timeout");
        assert!(timeout.get("payload").is_none());
    }

    #[test]
    fn stage_chain_is_correlated_ordered_and_has_one_terminal() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        let (root, _) = observer.start_external_call(
            "shop",
            "entry",
            "orders",
            "create",
            Some("00112233445566778899aabbccddeeff"),
        );
        let stages = [
            Stage::Ingress,
            Stage::Admission,
            Stage::Handler,
            Stage::Database,
            Stage::ProfuseContract,
            Stage::Response,
            Stage::ResourceFinalization,
        ];
        for (index, stage) in stages.into_iter().enumerate() {
            observer
                .start_stage(root.context(), stage, event_context((index + 1) as u32))
                .succeed();
        }
        drop(observer.start_stage(root.context(), Stage::Handler, event_context(8)));
        root.succeed();
        block_on(observer.flush()).unwrap();

        let records = records(&capture);
        let stage_records: Vec<_> = records
            .iter()
            .filter(|value| {
                value["event"]
                    .as_str()
                    .unwrap()
                    .starts_with("framework.stage.")
            })
            .collect();
        assert_eq!(stage_records.len(), 16);
        for pair in stage_records.chunks_exact(2) {
            assert_eq!(pair[0]["event"], "framework.stage.started");
            assert_eq!(pair[1]["event"], "framework.stage.finished");
            assert_eq!(pair[0]["rpc_id"], pair[1]["rpc_id"]);
            assert_eq!(pair[0]["trace_id"], "00112233445566778899aabbccddeeff");
            for field in [
                "timestamp_unix_ms",
                "timestamp",
                "level",
                "event",
                "stage",
                "trace_id",
                "rpc_id",
                "request_identity",
                "route",
                "attempt",
                "outcome",
                "error_code",
            ] {
                assert!(pair[0].get(field).is_some(), "missing {field}");
            }
            assert!(pair[1].get("elapsed_ms").is_some());
        }
        assert_eq!(stage_records.last().unwrap()["outcome"], "cancelled");
    }

    #[test]
    fn closed_observations_have_common_fields_and_no_sensitive_payload() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
        let common = event_context(1);
        observer.record_capacity(
            root.context(),
            &common,
            CapacityObservation::rejected(
                100,
                80,
                80,
                BottleneckIdentity::new("request_slots").unwrap(),
                RejectReason::new("limit_reached").unwrap(),
            ),
        );
        observer.record_database_disposition(
            root.context(),
            &common,
            DatabaseDisposition::Returned,
        );
        observer.record_outbound(
            root.context(),
            &common,
            OutboundObservation::new(
                RouteIdentity::new("cn-hz-a").unwrap(),
                OutboundAuthority::new("inventory-service").unwrap(),
                OutboundResult::Success,
            ),
        );
        observer.record_lifecycle(
            root.context(),
            &common,
            LifecycleState::Draining,
            Health::Healthy,
        );
        observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
        root.succeed();
        let _ = block_on(observer.flush());

        let records: Vec<_> = records(&capture)
            .into_iter()
            .filter(|value| {
                matches!(
                    value["event"].as_str(),
                    Some(
                        "framework.capacity"
                            | "framework.database.disposition"
                            | "framework.outbound"
                            | "framework.lifecycle"
                            | "framework.logger.health"
                    )
                )
            })
            .collect();
        assert_eq!(records.len(), 5);
        for record in records {
            for field in [
                "timestamp_unix_ms",
                "timestamp",
                "level",
                "event",
                "stage",
                "trace_id",
                "rpc_id",
                "request_identity",
                "route",
                "attempt",
                "elapsed_ms",
                "outcome",
                "error_code",
            ] {
                assert!(record.get(field).is_some(), "missing {field}");
            }
            let encoded = serde_json::to_string(&record).unwrap();
            for forbidden in [
                "request_body",
                "response_body",
                "cookie",
                "session",
                "token",
                "password",
                "connection_string",
                "db_value",
            ] {
                assert!(!encoded.contains(forbidden));
            }
        }
    }

    #[test]
    fn fixed_metrics_snapshot_tracks_closed_low_cardinality_events() {
        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
        let common = event_context(1);
        observer
            .start_stage(root.context(), Stage::Response, event_context(1))
            .succeed();
        observer.record_capacity(
            root.context(),
            &common,
            CapacityObservation::rejected_dimension(
                CapacityDimension::Database,
                100,
                80,
                80,
                BottleneckIdentity::new("database_slots").unwrap(),
                RejectReason::new("limit_reached").unwrap(),
                2,
            ),
        );
        observer.record_database_disposition(
            root.context(),
            &common,
            DatabaseDisposition::Discarded,
        );
        observer.record_outbound(
            root.context(),
            &common,
            OutboundObservation::new(
                RouteIdentity::new("cn-hz-a").unwrap(),
                OutboundAuthority::new("inventory-service").unwrap(),
                OutboundResult::Timeout,
            ),
        );
        observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
        observer.record_lifecycle(
            root.context(),
            &common,
            LifecycleState::Running,
            Health::Healthy,
        );

        let snapshot = observer.metrics_snapshot();
        assert_eq!(snapshot.requests(StageOutcome::Success), 1);
        assert_eq!(
            snapshot.stage_latency(Stage::Response).iter().sum::<u64>(),
            1
        );
        assert_eq!(
            snapshot.capacity_rejected(Some(CapacityDimension::Database)),
            1
        );
        assert_eq!(
            snapshot.capacity_used(Some(CapacityDimension::Database)),
            80
        );
        assert_eq!(snapshot.database(DatabaseDisposition::Discarded), 1);
        assert_eq!(snapshot.outbound(OutboundResult::Timeout), 1);
        assert_eq!(snapshot.logger_dropped(), 0);
        assert_eq!(snapshot.logger_output_failed(), 1);
        assert_eq!(snapshot.lifecycle(), LifecycleState::Running);
        assert_eq!(snapshot.health(), Health::Healthy);
        assert!(snapshot.ready());
        root.succeed();
    }

    #[test]
    fn typed_capacity_and_resource_terminal_preserve_closed_schema() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
        let common = event_context(1);
        observer.record_capacity(
            root.context(),
            &common,
            CapacityObservation::rejected_dimension(
                CapacityDimension::Database,
                4,
                2,
                2,
                BottleneckIdentity::new("database").unwrap(),
                RejectReason::new("at_limit").unwrap(),
                17,
            ),
        );
        observer.record_resource_finalization(
            root.context(),
            &common,
            DatabaseDisposition::NotUsed,
            23,
        );
        root.succeed();
        block_on(observer.flush()).unwrap();

        let records = records(&capture);
        let capacity = records
            .iter()
            .find(|record| record["event"] == "framework.capacity")
            .unwrap();
        assert_eq!(capacity["capacity_dimension"], "database");
        assert_eq!(capacity["budget"], 4);
        assert_eq!(capacity["limit"], 2);
        assert_eq!(capacity["used"], 2);
        assert_eq!(capacity["bottleneck"], "database");
        assert_eq!(capacity["reject_reason"], "at_limit");
        assert_eq!(capacity["elapsed_ms"], 17);

        let terminal = records
            .iter()
            .find(|record| record["event"] == "framework.resource.finalized")
            .unwrap();
        assert_eq!(terminal["stage"], "resource_finalization");
        assert_eq!(terminal["credit"], "released");
        assert_eq!(terminal["db_disposition"], "not_used");
        assert_eq!(terminal["elapsed_ms"], 23);
        assert_eq!(terminal["outcome"], "success");
        assert_eq!(capacity["trace_id"], terminal["trace_id"]);
        assert_eq!(capacity["rpc_id"], terminal["rpc_id"]);
        assert_eq!(capacity["request_identity"], terminal["request_identity"]);
        assert_eq!(capacity["route"], terminal["route"]);
        assert_eq!(capacity["attempt"], terminal["attempt"]);
    }

    #[test]
    fn unsafe_or_unbounded_identifiers_and_zero_attempt_are_rejected() {
        assert_eq!(RequestIdentity::new(""), Err(ChainFieldError::Empty));
        assert_eq!(
            RouteIdentity::new("x".repeat(257)),
            Err(ChainFieldError::TooLong)
        );
        assert_eq!(
            RejectReason::new("bad\nreason"),
            Err(ChainFieldError::ControlCharacter)
        );
        for unsafe_value in ["https://user:secret@host/path", "host/path", "host?token=x"] {
            assert_eq!(
                OutboundAuthority::new(unsafe_value),
                Err(ChainFieldError::UnsafeAuthority)
            );
        }
        assert_eq!(
            EventContext::new(
                RequestIdentity::new("r").unwrap(),
                RouteIdentity::new("route").unwrap(),
                0
            ),
            Err(ChainFieldError::ZeroAttempt),
        );
    }
}