ferrum_cli/

observability_product.rs

use chrono::{Duration, Utc};
use clap::ValueEnum;
use ferrum_types::{
    FerrumError, FerrumObservabilityConfig, FerrumProfileEvent, MemorySnapshot,
    ObservabilityProfileDetail, ProfileEntrypoint, ProfileError, ProfileEventKind, ProfileStatus,
    ReplayReference, ResourceAction, ResourceTraceEvent, Result, SamplingParams,
    DEFAULT_OBSERVABILITY_PROFILE_SAMPLE_RATE, OBSERVABILITY_PROFILE_SCHEMA_VERSION,
};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fs;
use std::ops::Deref;
use std::path::{Path, PathBuf};
use uuid::Uuid;

const SYNTHETIC_MODEL: &str = "synthetic/no-weight";
const SYNTHETIC_BACKEND: &str = "synthetic";

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)]
pub enum ProfileDetailArg {
    #[default]
    Off,
    Basic,
    Resource,
    Latency,
    Kernel,
    Debug,
    Replay,
    Verify,
    Full,
}

impl ProfileDetailArg {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Off => "off",
            Self::Basic => "basic",
            Self::Resource => "resource",
            Self::Latency => "latency",
            Self::Kernel => "kernel",
            Self::Debug => "debug",
            Self::Replay => "replay",
            Self::Verify => "verify",
            Self::Full => "full",
        }
    }
}

impl From<ProfileDetailArg> for ObservabilityProfileDetail {
    fn from(value: ProfileDetailArg) -> Self {
        match value {
            ProfileDetailArg::Off => Self::Off,
            ProfileDetailArg::Basic => Self::Basic,
            ProfileDetailArg::Resource => Self::Resource,
            ProfileDetailArg::Latency => Self::Latency,
            ProfileDetailArg::Kernel => Self::Kernel,
            ProfileDetailArg::Debug => Self::Debug,
            ProfileDetailArg::Replay => Self::Replay,
            ProfileDetailArg::Verify => Self::Verify,
            ProfileDetailArg::Full => Self::Full,
        }
    }
}

#[derive(Clone, Debug)]
pub struct ProductObservabilityConfig {
    pub core: FerrumObservabilityConfig,
}

impl Deref for ProductObservabilityConfig {
    type Target = FerrumObservabilityConfig;

    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

pub struct ActualRunObservation {
    pub request_id: String,
    pub duration_us: u64,
    pub sampling_params: SamplingParams,
    pub prompt_token_ids: Option<Vec<u32>>,
    pub prompt_token_count: Option<usize>,
    pub output_tokens: usize,
    pub output_token_ids: Vec<u32>,
    pub chunk_count: usize,
    pub finish_reason: Option<String>,
    pub prompt_chars: usize,
    pub response_chars: usize,
    pub response_text: String,
    pub execution_evidence: Option<ferrum_types::InferenceExecutionEvidence>,
    pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    pub memory_stages: Vec<ActualMemoryStageObservation>,
}

pub struct ActualRunFailureObservation {
    pub request_id: String,
    pub duration_us: u64,
    pub sampling_params: SamplingParams,
    pub prompt_token_ids: Option<Vec<u32>>,
    pub prompt_token_count: Option<usize>,
    pub prompt_chars: usize,
    pub failure_kind: String,
    pub error_kind: String,
    pub error_message: String,
    pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    pub memory_stages: Vec<ActualMemoryStageObservation>,
}

#[derive(Clone, Debug)]
pub struct ActualMemoryStageObservation {
    pub phase: String,
    pub stage: String,
    pub duration_us: Option<u64>,
    pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    pub attributes: BTreeMap<String, Value>,
}

impl ActualMemoryStageObservation {
    pub fn new(
        phase: impl Into<String>,
        stage: impl Into<String>,
        duration_us: Option<u64>,
        memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    ) -> Self {
        Self {
            phase: phase.into(),
            stage: stage.into(),
            duration_us,
            memory,
            attributes: BTreeMap::new(),
        }
    }

    pub fn with_attribute(mut self, key: impl Into<String>, value: Value) -> Self {
        self.attributes.insert(key.into(), value);
        self
    }

    pub fn with_profile_run_status(
        mut self,
        executed: bool,
        status: impl Into<String>,
        source: impl Into<String>,
    ) -> Self {
        self.attributes.extend([
            ("profile_run_executed".to_string(), json!(executed)),
            ("profile_run_status".to_string(), json!(status.into())),
            ("profile_run_source".to_string(), json!(source.into())),
        ]);
        self
    }

    pub fn with_engine_cache_status(mut self, status: &ferrum_types::EngineStatus) -> Self {
        let memory = &status.memory_usage;
        let dynamic_capacity_bytes = memory.cache_memory_bytes.saturating_add(memory.free_bytes);
        self.attributes.extend([
            (
                "kv_cache_total_bytes".to_string(),
                json!(dynamic_capacity_bytes),
            ),
            (
                "kv_cache_used_bytes".to_string(),
                json!(memory.cache_memory_bytes),
            ),
            ("kv_cache_free_bytes".to_string(), json!(memory.free_bytes)),
            (
                "cache_memory_bytes".to_string(),
                json!(memory.cache_memory_bytes),
            ),
            (
                "available_kv_or_state_bytes".to_string(),
                json!(memory.free_bytes),
            ),
            (
                "resource_total_bytes".to_string(),
                json!(memory.total_bytes),
            ),
            ("resource_used_bytes".to_string(), json!(memory.used_bytes)),
            ("resource_free_bytes".to_string(), json!(memory.free_bytes)),
        ]);
        self
    }
}

impl ProductObservabilityConfig {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        entrypoint: ProfileEntrypoint,
        model: impl Into<String>,
        profile_jsonl: Option<&PathBuf>,
        profile_detail: ProfileDetailArg,
        memory_profile_jsonl: Option<&PathBuf>,
        scheduler_trace_jsonl: Option<&PathBuf>,
        request_dump_dir: Option<&PathBuf>,
        profile_sample_rate: f64,
    ) -> Self {
        Self {
            core: FerrumObservabilityConfig::new(
                entrypoint,
                model,
                profile_jsonl.cloned(),
                profile_detail.into(),
                memory_profile_jsonl.cloned(),
                scheduler_trace_jsonl.cloned(),
                request_dump_dir.cloned(),
                profile_sample_rate,
            ),
        }
    }

    pub fn enabled(&self) -> bool {
        self.core.enabled()
    }

    pub fn synthetic_no_weight_enabled(&self) -> bool {
        self.core.synthetic_no_weight_enabled()
    }

    pub fn unified_product_profile_enabled(&self) -> bool {
        self.core.unified_product_profile_enabled()
    }

    fn validate(&self) -> Result<()> {
        self.core.validate().map_err(FerrumError::invalid_parameter)
    }
}

pub fn default_profile_sample_rate() -> f64 {
    DEFAULT_OBSERVABILITY_PROFILE_SAMPLE_RATE
}

pub fn write_synthetic_product_observability(
    config: &ProductObservabilityConfig,
) -> Result<Vec<PathBuf>> {
    config.validate()?;
    let request_id = format!(
        "product-obs-{}-{}",
        entrypoint_label(config.entrypoint),
        Uuid::new_v4().simple()
    );
    let replay_command = replay_command(config);
    let events = product_events(config, &request_id, &replay_command);
    let mut written = write_profile_outputs(
        config,
        &events,
        ferrum_bench_core::JsonlJournalOpenMode::Truncate,
        true,
        true,
        true,
    )?;
    if let Some(dir) = &config.request_dump_dir {
        fs_create_dir_all(dir)?;
        written.extend(write_replay_bundle(
            dir,
            config,
            &request_id,
            &replay_command,
            ReplayBundleData {
                request: request_dump(config, &request_id, &replay_command),
                prompt_token_ids: Some(vec![101, 202, 303, 404]),
                prompt_token_count: Some(4),
                prompt_token_unavailable_reason: None,
                sampling_params: Some(SamplingParams::greedy()),
                backend: SYNTHETIC_BACKEND,
                actual_model_smoke: false,
                output_token_ids: Some(vec![909, 808]),
                output_text: Some("synthetic ok"),
                finish_reason: Some("stop"),
                failure_kind: None,
                failure_diagnostics: None,
            },
        )?);
    }
    Ok(written)
}

pub fn write_actual_run_observability(
    config: &ProductObservabilityConfig,
    observation: &ActualRunObservation,
) -> Result<Vec<PathBuf>> {
    if !config.enabled() {
        return Ok(Vec::new());
    }
    config.validate()?;
    if let Some(timing) = observation
        .execution_evidence
        .as_ref()
        .and_then(|evidence| evidence.engine_token_timing.as_ref())
    {
        timing
            .validate(observation.output_tokens)
            .map_err(FerrumError::invalid_parameter)?;
    }
    let replay_command = replay_command(config);
    let events = actual_run_events(config, observation, &replay_command);
    write_actual_run_artifacts(config, &events, observation, &replay_command)
}

pub fn write_actual_run_failure_observability(
    config: &ProductObservabilityConfig,
    observation: &ActualRunFailureObservation,
) -> Result<Vec<PathBuf>> {
    if !config.enabled() {
        return Ok(Vec::new());
    }
    config.validate()?;
    let replay_command = replay_command(config);
    let events = actual_run_failure_events(config, observation, &replay_command);
    write_actual_run_failure_artifacts(config, &events, observation, &replay_command)
}

pub fn write_actual_serve_startup_observability(
    config: &ProductObservabilityConfig,
    startup_duration_us: u64,
    startup_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    memory_stages: Vec<ActualMemoryStageObservation>,
) -> Result<Vec<PathBuf>> {
    if !config.unified_product_profile_enabled() {
        return Ok(Vec::new());
    }
    config.validate()?;
    let request_id = format!("serve-startup-{}", Uuid::new_v4().simple());
    let replay_command = replay_command(config);
    let events = actual_serve_startup_events(
        config,
        &request_id,
        startup_duration_us,
        startup_memory.as_ref(),
        &memory_stages,
        &replay_command,
    );
    write_actual_artifacts(config, &events, &request_id, &replay_command)
}

pub fn append_actual_serve_memory_stage_observability(
    config: &ProductObservabilityConfig,
    stage: ActualMemoryStageObservation,
) -> Result<Vec<PathBuf>> {
    if !config.unified_product_profile_enabled() {
        return Ok(Vec::new());
    }
    config.validate()?;
    let request_id = format!("serve-memory-{}", Uuid::new_v4().simple());
    let events = actual_memory_stage_events(config, &request_id, &[stage], Utc::now());
    write_profile_outputs(
        config,
        &events,
        ferrum_bench_core::JsonlJournalOpenMode::Append,
        true,
        true,
        false,
    )
}

fn write_actual_run_artifacts(
    config: &ProductObservabilityConfig,
    events: &[FerrumProfileEvent],
    observation: &ActualRunObservation,
    replay_command: &str,
) -> Result<Vec<PathBuf>> {
    // Successful actual inference must prove its scheduler lifecycle at the engine boundary.
    // Adding a product fallback here would duplicate real events or hide missing engine evidence.
    let mut written = write_actual_artifacts_with_scheduler(
        config,
        events,
        &observation.request_id,
        replay_command,
        false,
    )?;
    if let Some(dir) = &config.request_dump_dir {
        written.extend(write_replay_bundle(
            dir,
            config,
            &observation.request_id,
            replay_command,
            ReplayBundleData {
                request: actual_request_dump(config, &observation.request_id, replay_command),
                prompt_token_ids: observation.prompt_token_ids.clone(),
                prompt_token_count: observation.prompt_token_count,
                prompt_token_unavailable_reason: observation.prompt_token_ids.is_none().then_some(
                    "rendered prompt token ids were unavailable for run one-shot observability",
                ),
                sampling_params: Some(observation.sampling_params.clone()),
                backend: "actual",
                actual_model_smoke: true,
                output_token_ids: Some(observation.output_token_ids.clone()),
                output_text: Some(&observation.response_text),
                finish_reason: observation.finish_reason.as_deref(),
                failure_kind: None,
                failure_diagnostics: None,
            },
        )?);
    }
    Ok(written)
}

fn write_actual_run_failure_artifacts(
    config: &ProductObservabilityConfig,
    events: &[FerrumProfileEvent],
    observation: &ActualRunFailureObservation,
    replay_command: &str,
) -> Result<Vec<PathBuf>> {
    let mut written =
        write_actual_artifacts(config, events, &observation.request_id, replay_command)?;
    if let Some(dir) = &config.request_dump_dir {
        written.extend(write_replay_bundle(
            dir,
            config,
            &observation.request_id,
            replay_command,
            ReplayBundleData {
                request: actual_request_dump(config, &observation.request_id, replay_command),
                prompt_token_ids: observation.prompt_token_ids.clone(),
                prompt_token_count: observation.prompt_token_count,
                prompt_token_unavailable_reason: observation.prompt_token_ids.is_none().then_some(
                    "rendered prompt token ids were unavailable for run failure observability",
                ),
                sampling_params: Some(observation.sampling_params.clone()),
                backend: "actual",
                actual_model_smoke: true,
                output_token_ids: Some(Vec::new()),
                output_text: Some(""),
                finish_reason: Some("error"),
                failure_kind: Some(observation.failure_kind.as_str()),
                failure_diagnostics: Some(actual_run_failure_diagnostics(observation)),
            },
        )?);
    }
    Ok(written)
}

fn write_actual_artifacts(
    config: &ProductObservabilityConfig,
    events: &[FerrumProfileEvent],
    request_id: &str,
    replay_command: &str,
) -> Result<Vec<PathBuf>> {
    write_actual_artifacts_with_scheduler(config, events, request_id, replay_command, true)
}

fn write_actual_artifacts_with_scheduler(
    config: &ProductObservabilityConfig,
    events: &[FerrumProfileEvent],
    request_id: &str,
    replay_command: &str,
    include_scheduler: bool,
) -> Result<Vec<PathBuf>> {
    let mut written = write_profile_outputs(
        config,
        events,
        ferrum_bench_core::JsonlJournalOpenMode::Append,
        true,
        true,
        include_scheduler,
    )?;
    if let Some(dir) = &config.request_dump_dir {
        fs_create_dir_all(dir)?;
        written.extend(write_replay_bundle(
            dir,
            config,
            request_id,
            replay_command,
            ReplayBundleData {
                request: actual_request_dump(config, request_id, replay_command),
                prompt_token_ids: None,
                prompt_token_count: None,
                prompt_token_unavailable_reason: Some(
                    "startup or non-run request has no rendered prompt token dump in WP9 L0",
                ),
                sampling_params: None,
                backend: "actual",
                actual_model_smoke: true,
                output_token_ids: Some(Vec::new()),
                output_text: None,
                finish_reason: None,
                failure_kind: None,
                failure_diagnostics: None,
            },
        )?);
    }
    Ok(written)
}

#[derive(Clone, Copy, Default)]
struct ProfileOutputRoles {
    profile: bool,
    memory: bool,
    scheduler: bool,
}

fn add_profile_output_target(
    targets: &mut BTreeMap<PathBuf, ProfileOutputRoles>,
    path: Option<&PathBuf>,
    role: fn(&mut ProfileOutputRoles) -> &mut bool,
) -> Result<()> {
    let Some(path) = path else {
        return Ok(());
    };
    let normalized = ferrum_bench_core::normalize_jsonl_path(path).map_err(|error| {
        FerrumError::io(format!(
            "normalize observability JSONL path {}: {error}",
            path.display()
        ))
    })?;
    *role(targets.entry(normalized).or_default()) = true;
    Ok(())
}

fn write_profile_outputs(
    config: &ProductObservabilityConfig,
    events: &[FerrumProfileEvent],
    mode: ferrum_bench_core::JsonlJournalOpenMode,
    include_profile: bool,
    include_memory: bool,
    include_scheduler: bool,
) -> Result<Vec<PathBuf>> {
    let mut targets = BTreeMap::<PathBuf, ProfileOutputRoles>::new();
    if include_profile {
        add_profile_output_target(&mut targets, config.profile_jsonl.as_ref(), |roles| {
            &mut roles.profile
        })?;
    }
    if include_memory {
        add_profile_output_target(
            &mut targets,
            config.memory_profile_jsonl.as_ref(),
            |roles| &mut roles.memory,
        )?;
    }
    if include_scheduler {
        add_profile_output_target(
            &mut targets,
            config.scheduler_trace_jsonl.as_ref(),
            |roles| &mut roles.scheduler,
        )?;
    }

    let mut written = Vec::with_capacity(targets.len());
    for (path, roles) in targets {
        let selected = events
            .iter()
            .filter(|event| {
                roles.profile
                    || (roles.memory && event.memory.is_some())
                    || (roles.scheduler && event.resource.is_some())
            })
            .cloned()
            .collect::<Vec<_>>();
        if selected.is_empty() {
            continue;
        }
        write_profile_events(&path, mode, &selected)?;
        written.push(path);
    }
    Ok(written)
}

fn product_events(
    config: &ProductObservabilityConfig,
    request_id: &str,
    replay_command: &str,
) -> Vec<FerrumProfileEvent> {
    let base = Utc::now();
    let open = resource_event(
        config,
        request_id,
        "request",
        request_id,
        "request_slot",
        "request_open",
        ResourceAction::RequestOpen,
        base,
        None,
        None,
        None,
        Some(1),
        None,
    );
    let reserve = resource_event(
        config,
        request_id,
        "request",
        request_id,
        "request_slot",
        "request_slot_reserve",
        ResourceAction::Reserve,
        base + Duration::microseconds(10),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );
    let commit = resource_event(
        config,
        request_id,
        "request",
        request_id,
        "request_slot",
        "request_slot_commit",
        ResourceAction::Commit,
        base + Duration::microseconds(20),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );

    let mut prefill = base_event(
        config,
        request_id,
        "synthetic_prefill",
        ProfileEventKind::TimedSpan,
        base + Duration::microseconds(30),
    );
    prefill.duration_us = Some(160);
    prefill.memory = Some(MemorySnapshot {
        scope: "process".to_string(),
        backend: Some(SYNTHETIC_BACKEND.to_string()),
        before_bytes: Some(2048),
        after_bytes: Some(2304),
        current_bytes: Some(2304),
        high_water_bytes: Some(2304),
        available_bytes: Some(1024 * 1024),
    });
    prefill.attributes.extend(common_attrs(config));
    prefill
        .attributes
        .insert("input_tokens".to_string(), json!(8));

    let release = resource_event(
        config,
        request_id,
        "request",
        request_id,
        "request_slot",
        "request_slot_release",
        ResourceAction::Release,
        base + Duration::microseconds(180),
        Some(1),
        Some(1),
        Some(0),
        Some(1),
        None,
    );

    let mut close = resource_event(
        config,
        request_id,
        "request",
        request_id,
        "request_slot",
        "request_close",
        ResourceAction::RequestClose,
        base + Duration::microseconds(190),
        None,
        None,
        None,
        Some(1),
        None,
    );
    close.status = if config.profile_detail.diagnostic_only() {
        ProfileStatus::DiagnosticOnly
    } else {
        ProfileStatus::Ok
    };
    close.replay = Some(ReplayReference {
        command: replay_command.to_string(),
        bundle_dir: config
            .request_dump_dir
            .as_ref()
            .map(|path| path.to_string_lossy().to_string()),
    });
    close.attributes.extend(common_attrs(config));
    close
        .attributes
        .insert("response_text".to_string(), json!("synthetic ok"));
    vec![open, reserve, commit, prefill, release, close]
}

fn actual_run_events(
    config: &ProductObservabilityConfig,
    observation: &ActualRunObservation,
    replay_command: &str,
) -> Vec<FerrumProfileEvent> {
    let base = Utc::now();
    let shutdown_stage_index = observation
        .memory_stages
        .iter()
        .position(|stage| stage.stage == "shutdown")
        .unwrap_or(observation.memory_stages.len());
    let mut events = actual_memory_stage_events(
        config,
        &observation.request_id,
        &observation.memory_stages[..shutdown_stage_index],
        base,
    );
    let open = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_open",
        ResourceAction::RequestOpen,
        base,
        None,
        None,
        None,
        Some(1),
        None,
    );
    let reserve = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_reserve",
        ResourceAction::Reserve,
        base + Duration::microseconds(5),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );
    let commit = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_commit",
        ResourceAction::Commit,
        base + Duration::microseconds(10),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );

    let mut generation = actual_base_event(
        config,
        &observation.request_id,
        "actual_run_generation",
        ProfileEventKind::TimedSpan,
        base + Duration::microseconds(20),
    );
    generation.duration_us = Some(observation.duration_us);
    attach_process_memory(
        &mut generation,
        observation.memory.as_ref(),
        "first_request_done",
    );
    generation.attributes.insert(
        "output_tokens".to_string(),
        json!(observation.output_tokens),
    );
    generation.attributes.insert(
        "output_token_count".to_string(),
        json!(observation.output_tokens),
    );
    generation.attributes.insert(
        "completion_token_count".to_string(),
        json!(observation.output_tokens),
    );
    generation.attributes.insert(
        "e2e_duration_us".to_string(),
        json!(observation.duration_us),
    );
    if let Some(prompt_token_count) = observation.prompt_token_count {
        generation
            .attributes
            .insert("prompt_token_count".to_string(), json!(prompt_token_count));
        generation.attributes.insert(
            "total_token_count".to_string(),
            json!(prompt_token_count.saturating_add(observation.output_tokens)),
        );
        generation.attributes.insert(
            "token_count_source".to_string(),
            json!("rendered_prompt_and_generated_tokens"),
        );
    } else {
        generation.attributes.insert(
            "total_token_count".to_string(),
            json!(observation.output_tokens),
        );
        generation
            .attributes
            .insert("token_count_source".to_string(), json!("generated_tokens"));
        generation.attributes.insert(
            "prompt_token_unavailable_reason".to_string(),
            json!("run rendered prompt token count was unavailable"),
        );
    }
    generation
        .attributes
        .insert("chunk_count".to_string(), json!(observation.chunk_count));
    generation.attributes.insert(
        "finish_reason".to_string(),
        json!(observation.finish_reason.as_deref().unwrap_or("unknown")),
    );
    if let Some(timing) = observation
        .execution_evidence
        .as_ref()
        .and_then(|evidence| evidence.engine_token_timing.as_ref())
    {
        generation
            .attributes
            .extend(ferrum_types::engine_token_timing_profile_attributes(timing));
    }

    let release = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_release",
        ResourceAction::Release,
        base + Duration::microseconds(30),
        Some(1),
        Some(1),
        Some(0),
        Some(1),
        None,
    );

    let mut close = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_close",
        ResourceAction::RequestClose,
        base + Duration::microseconds(40),
        None,
        None,
        None,
        Some(1),
        None,
    );
    close.replay = Some(ReplayReference {
        command: replay_command.to_string(),
        bundle_dir: config
            .request_dump_dir
            .as_ref()
            .map(|path| path.to_string_lossy().to_string()),
    });
    close
        .attributes
        .insert("prompt_chars".to_string(), json!(observation.prompt_chars));
    close.attributes.insert(
        "response_chars".to_string(),
        json!(observation.response_chars),
    );
    events.extend([open, reserve, commit, generation, release, close]);
    events.extend(actual_memory_stage_events(
        config,
        &observation.request_id,
        &observation.memory_stages[shutdown_stage_index..],
        base + Duration::microseconds(50),
    ));
    events
}

fn actual_run_failure_events(
    config: &ProductObservabilityConfig,
    observation: &ActualRunFailureObservation,
    replay_command: &str,
) -> Vec<FerrumProfileEvent> {
    let base = Utc::now();
    let shutdown_stage_index = observation
        .memory_stages
        .iter()
        .position(|stage| stage.stage == "shutdown")
        .unwrap_or(observation.memory_stages.len());
    let mut events = actual_memory_stage_events(
        config,
        &observation.request_id,
        &observation.memory_stages[..shutdown_stage_index],
        base,
    );
    let open = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_open",
        ResourceAction::RequestOpen,
        base,
        None,
        None,
        None,
        Some(1),
        None,
    );
    let reserve = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_reserve",
        ResourceAction::Reserve,
        base + Duration::microseconds(5),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );
    let commit = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_commit",
        ResourceAction::Commit,
        base + Duration::microseconds(10),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );

    let mut failure = actual_base_event(
        config,
        &observation.request_id,
        "actual_run_generation_failed",
        ProfileEventKind::TimedSpan,
        base + Duration::microseconds(20),
    );
    failure.status = ProfileStatus::Failure;
    failure.duration_us = Some(observation.duration_us);
    failure.error = Some(ProfileError {
        kind: observation.error_kind.clone(),
        message: observation.error_message.clone(),
        blocking: false,
    });
    failure.replay = Some(ReplayReference {
        command: replay_command.to_string(),
        bundle_dir: config
            .request_dump_dir
            .as_ref()
            .map(|path| path.to_string_lossy().to_string()),
    });
    attach_process_memory(
        &mut failure,
        observation.memory.as_ref(),
        "first_request_failed",
    );
    failure
        .attributes
        .insert("terminal_failure_event".to_string(), json!(true));
    failure
        .attributes
        .insert("prompt_chars".to_string(), json!(observation.prompt_chars));

    let release = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_slot_release",
        ResourceAction::Release,
        base + Duration::microseconds(30),
        Some(1),
        Some(1),
        Some(0),
        Some(1),
        None,
    );

    let mut close = actual_resource_event(
        config,
        &observation.request_id,
        "request",
        &observation.request_id,
        "request_slot",
        "request_close",
        ResourceAction::RequestClose,
        base + Duration::microseconds(40),
        None,
        None,
        None,
        Some(1),
        None,
    );
    close.replay = Some(ReplayReference {
        command: replay_command.to_string(),
        bundle_dir: config
            .request_dump_dir
            .as_ref()
            .map(|path| path.to_string_lossy().to_string()),
    });
    events.extend([open, reserve, commit, failure, release, close]);
    events.extend(actual_memory_stage_events(
        config,
        &observation.request_id,
        &observation.memory_stages[shutdown_stage_index..],
        base + Duration::microseconds(50),
    ));
    events
}

fn actual_memory_stage_events(
    config: &ProductObservabilityConfig,
    request_id: &str,
    stages: &[ActualMemoryStageObservation],
    base: chrono::DateTime<Utc>,
) -> Vec<FerrumProfileEvent> {
    stages
        .iter()
        .enumerate()
        .map(|(index, stage)| {
            let mut event = actual_base_event(
                config,
                request_id,
                &stage.phase,
                ProfileEventKind::Memory,
                base + Duration::microseconds(index as i64),
            );
            event.duration_us = stage.duration_us;
            attach_process_memory(&mut event, stage.memory.as_ref(), &stage.stage);
            event.attributes.extend(stage.attributes.clone());
            event
        })
        .collect()
}

fn actual_serve_startup_events(
    config: &ProductObservabilityConfig,
    request_id: &str,
    startup_duration_us: u64,
    startup_memory: Option<&crate::memory_profile::ProcessMemoryObservation>,
    memory_stages: &[ActualMemoryStageObservation],
    replay_command: &str,
) -> Vec<FerrumProfileEvent> {
    let base = Utc::now();
    let post_model_loaded_index = memory_stages
        .iter()
        .position(|stage| matches!(stage.stage.as_str(), "profile_run_done" | "cache_allocated"))
        .unwrap_or(memory_stages.len());
    let mut events = actual_memory_stage_events(
        config,
        request_id,
        &memory_stages[..post_model_loaded_index],
        base,
    );
    let open = actual_resource_event(
        config,
        request_id,
        "server",
        request_id,
        "startup_slot",
        "server_startup_open",
        ResourceAction::RequestOpen,
        base,
        None,
        None,
        None,
        Some(1),
        None,
    );
    let reserve = actual_resource_event(
        config,
        request_id,
        "server",
        request_id,
        "startup_slot",
        "server_startup_reserve",
        ResourceAction::Reserve,
        base + Duration::microseconds(5),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );
    let commit = actual_resource_event(
        config,
        request_id,
        "server",
        request_id,
        "startup_slot",
        "server_startup_commit",
        ResourceAction::Commit,
        base + Duration::microseconds(10),
        Some(1),
        Some(0),
        Some(1),
        Some(1),
        None,
    );
    let mut startup = actual_base_event(
        config,
        request_id,
        "actual_serve_startup",
        ProfileEventKind::TimedSpan,
        base + Duration::microseconds(20),
    );
    startup.duration_us = Some(startup_duration_us);
    attach_process_memory(&mut startup, startup_memory, "model_loaded");
    events.extend([open, reserve, commit, startup]);
    events.extend(actual_memory_stage_events(
        config,
        request_id,
        &memory_stages[post_model_loaded_index..],
        base + Duration::microseconds(21),
    ));

    let release = actual_resource_event(
        config,
        request_id,
        "server",
        request_id,
        "startup_slot",
        "server_startup_release",
        ResourceAction::Release,
        base + Duration::microseconds(30),
        Some(1),
        Some(1),
        Some(0),
        Some(1),
        None,
    );
    let mut ready = actual_resource_event(
        config,
        request_id,
        "server",
        request_id,
        "startup_slot",
        "server_ready_for_requests",
        ResourceAction::RequestClose,
        base + Duration::microseconds(40),
        None,
        None,
        None,
        Some(1),
        None,
    );
    ready.replay = Some(ReplayReference {
        command: replay_command.to_string(),
        bundle_dir: config
            .request_dump_dir
            .as_ref()
            .map(|path| path.to_string_lossy().to_string()),
    });
    events.extend([release, ready]);
    events
}

fn base_event(
    config: &ProductObservabilityConfig,
    request_id: &str,
    phase: &str,
    event_kind: ProfileEventKind,
    timestamp: chrono::DateTime<Utc>,
) -> FerrumProfileEvent {
    FerrumProfileEvent {
        schema_version: OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        ts_unix_nanos: timestamp
            .timestamp_nanos_opt()
            .unwrap_or_else(|| timestamp.timestamp_micros() * 1_000),
        event_id: format!(
            "evt-product-{}-{phase}",
            entrypoint_label(config.entrypoint)
        ),
        request_id: request_id.to_string(),
        correlation_id: Some(format!(
            "corr-product-{}",
            entrypoint_label(config.entrypoint)
        )),
        entrypoint: config.entrypoint,
        backend: SYNTHETIC_BACKEND.to_string(),
        runtime_preset_hash: runtime_preset_hash(config),
        phase: phase.to_string(),
        event_kind,
        timestamp,
        status: ProfileStatus::Ok,
        model: Some(config.model.clone()),
        duration_us: None,
        memory: None,
        resource: None,
        error: None,
        replay: None,
        shape: default_event_shape(),
        backend_detail: None,
        attributes: common_attrs(config),
    }
}

fn actual_base_event(
    config: &ProductObservabilityConfig,
    request_id: &str,
    phase: &str,
    event_kind: ProfileEventKind,
    timestamp: chrono::DateTime<Utc>,
) -> FerrumProfileEvent {
    let mut event = base_event(config, request_id, phase, event_kind, timestamp);
    event.backend = "actual".to_string();
    event.attributes = actual_attrs(config);
    event.attributes.insert(
        "execution_request_id".to_string(),
        json!(format!("request.product.{request_id}")),
    );
    event
}

fn runtime_preset_hash(config: &ProductObservabilityConfig) -> String {
    let mut hasher = Sha256::new();
    hasher.update(config.entrypoint.as_str().as_bytes());
    hasher.update(b"\0");
    hasher.update(config.model.as_bytes());
    hasher.update(b"\0");
    hasher.update(config.profile_detail.as_str().as_bytes());
    hasher.update(b"\0");
    hasher.update(config.profile_sample_rate.to_string().as_bytes());
    format!("sha256:{:x}", hasher.finalize())
}

fn default_event_shape() -> BTreeMap<String, Value> {
    BTreeMap::from([("batch_size".to_string(), json!(1))])
}

#[allow(clippy::too_many_arguments)]
fn resource_event(
    config: &ProductObservabilityConfig,
    request_id: &str,
    owner_kind: &str,
    owner_id: &str,
    resource_kind: &str,
    phase: &str,
    action: ResourceAction,
    timestamp: chrono::DateTime<Utc>,
    amount: Option<i64>,
    before: Option<i64>,
    after: Option<i64>,
    capacity: Option<i64>,
    reason: Option<&str>,
) -> FerrumProfileEvent {
    let mut event = base_event(
        config,
        request_id,
        phase,
        ProfileEventKind::Resource,
        timestamp,
    );
    event.resource = Some(ResourceTraceEvent {
        owner_kind: owner_kind.to_string(),
        owner_id: owner_id.to_string(),
        resource_kind: resource_kind.to_string(),
        action,
        amount,
        before,
        after,
        capacity,
        underflow_amount: match (action, amount, before) {
            (ResourceAction::Release | ResourceAction::Rollback, Some(amount), Some(before))
                if amount > before =>
            {
                Some(amount.saturating_sub(before))
            }
            _ => None,
        },
        reason: reason.map(str::to_string),
        error_kind: None,
        message: None,
        resource_error_kind: None,
    });
    event
}

#[allow(clippy::too_many_arguments)]
fn actual_resource_event(
    config: &ProductObservabilityConfig,
    request_id: &str,
    owner_kind: &str,
    owner_id: &str,
    resource_kind: &str,
    phase: &str,
    action: ResourceAction,
    timestamp: chrono::DateTime<Utc>,
    amount: Option<i64>,
    before: Option<i64>,
    after: Option<i64>,
    capacity: Option<i64>,
    reason: Option<&str>,
) -> FerrumProfileEvent {
    let mut event = resource_event(
        config,
        request_id,
        owner_kind,
        owner_id,
        resource_kind,
        phase,
        action,
        timestamp,
        amount,
        before,
        after,
        capacity,
        reason,
    );
    event.backend = "actual".to_string();
    event.attributes = actual_attrs(config);
    event
}

fn attach_process_memory(
    event: &mut FerrumProfileEvent,
    observation: Option<&crate::memory_profile::ProcessMemoryObservation>,
    stage: &str,
) {
    if let Some(observation) = observation {
        event.memory = Some(observation.to_snapshot("process", Some("actual")));
        event
            .attributes
            .insert("memory_measurement".to_string(), json!("process_rss"));
        event
            .attributes
            .insert("memory_stage".to_string(), json!(stage));
        event.attributes.insert(
            "process_memory_source".to_string(),
            json!(observation.source),
        );
    } else {
        event.memory = Some(MemorySnapshot {
            scope: "process".to_string(),
            backend: Some("actual".to_string()),
            before_bytes: Some(0),
            after_bytes: Some(0),
            current_bytes: Some(0),
            high_water_bytes: Some(0),
            available_bytes: None,
        });
        event
            .attributes
            .insert("memory_measurement".to_string(), json!("not_collected"));
        event
            .attributes
            .insert("memory_stage".to_string(), json!(stage));
    }
}

fn common_attrs(config: &ProductObservabilityConfig) -> BTreeMap<String, Value> {
    BTreeMap::from([
        (
            "profile_detail".to_string(),
            json!(config.profile_detail.as_str()),
        ),
        (
            "profile_sample_rate".to_string(),
            json!(config.profile_sample_rate),
        ),
        (
            "diagnostic_only".to_string(),
            json!(config.profile_detail.diagnostic_only()),
        ),
        ("l0_only".to_string(), json!(true)),
    ])
}

fn actual_attrs(config: &ProductObservabilityConfig) -> BTreeMap<String, Value> {
    BTreeMap::from([
        (
            "profile_detail".to_string(),
            json!(config.profile_detail.as_str()),
        ),
        (
            "profile_sample_rate".to_string(),
            json!(config.profile_sample_rate),
        ),
        (
            "diagnostic_only".to_string(),
            json!(config.profile_detail.diagnostic_only()),
        ),
        ("l0_only".to_string(), json!(false)),
        ("actual_model_smoke".to_string(), json!(true)),
    ])
}

struct ReplayBundleData<'a> {
    request: serde_json::Value,
    prompt_token_ids: Option<Vec<u32>>,
    prompt_token_count: Option<usize>,
    prompt_token_unavailable_reason: Option<&'a str>,
    sampling_params: Option<SamplingParams>,
    backend: &'a str,
    actual_model_smoke: bool,
    output_token_ids: Option<Vec<u32>>,
    output_text: Option<&'a str>,
    finish_reason: Option<&'a str>,
    failure_kind: Option<&'a str>,
    failure_diagnostics: Option<serde_json::Value>,
}

fn write_replay_bundle(
    root: &Path,
    config: &ProductObservabilityConfig,
    request_id: &str,
    replay_command: &str,
    data: ReplayBundleData<'_>,
) -> Result<Vec<PathBuf>> {
    fs_create_dir_all(root)?;
    let bundle_dir = root.join(request_id);
    fs_create_dir_all(&bundle_dir)?;
    let mut written = Vec::new();

    let request_path = root.join("request.json");
    let replay_path = root.join("replay_command.txt");
    write_json(&request_path, &data.request)?;
    fs_write(&replay_path, format!("{replay_command}\n"))?;
    written.push(request_path);
    written.push(replay_path);

    let prompt_token_count = data
        .prompt_token_count
        .or_else(|| data.prompt_token_ids.as_ref().map(Vec::len));
    let output_token_count = data.output_token_ids.as_ref().map(Vec::len).unwrap_or(0);
    let prompt_tokens = json!({
        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        "request_id": request_id,
        "model": config.model,
        "tokenizer_or_model": config.model,
        "token_ids": data.prompt_token_ids,
        "token_count": prompt_token_count,
        "unavailable_reason": data.prompt_token_unavailable_reason,
        "sanitized": true
    });
    let output_text = data.output_text.unwrap_or("");
    let sampling_unavailable_reason = if data.sampling_params.is_some() {
        Value::Null
    } else {
        json!("sampling params unavailable for this replay bundle kind in WP9 L0")
    };
    let output_text_body = if data.actual_model_smoke && config.model != SYNTHETIC_MODEL {
        format!(
            "[redacted actual output]\nsha256={}\nchars={}\n",
            sha256_hex(output_text.as_bytes()),
            output_text.chars().count()
        )
    } else {
        format!("{output_text}\n")
    };
    let output_scan = bad_output_scan(
        request_id,
        output_text,
        data.failure_kind,
        output_text_body.as_bytes(),
    );
    let engine_replay_args = engine_replay_command_args(&bundle_dir);
    let engine_replay_command = replay_command_from_args(&engine_replay_args);
    let files = [
        ("request.json", data.request),
        ("prompt_token_ids.json", prompt_tokens),
        (
            "sampling_params.json",
            json!({
                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
                "request_id": request_id,
                "sampling_params": data.sampling_params,
                "unavailable_reason": sampling_unavailable_reason
            }),
        ),
        (
            "runtime_effective_config.json",
            json!({
                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
                "request_id": request_id,
                "entrypoint": entrypoint_label(config.entrypoint),
                "profile_detail": config.profile_detail.as_str(),
                "profile_sample_rate": config.profile_sample_rate,
                "profile_jsonl": config.profile_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
                "memory_profile_jsonl": config.memory_profile_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
                "scheduler_trace_jsonl": config.scheduler_trace_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
                "request_dump_dir": Some(root.to_string_lossy().to_string()),
                "sanitized": true
            }),
        ),
        (
            "backend_selection.json",
            json!({
                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
                "request_id": request_id,
                "backend": data.backend,
                "model": config.model,
                "actual_model_smoke": data.actual_model_smoke
            }),
        ),
        (
            "output_token_ids.json",
            json!({
                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
                "request_id": request_id,
                "token_ids": data.output_token_ids.unwrap_or_default(),
                "token_count": output_token_count,
                "finish_reason": data.finish_reason
            }),
        ),
        ("bad_output_scan.json", output_scan),
        (
            "replay.command.json",
            json!({
                "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
                "request_id": request_id,
                "entrypoint": entrypoint_label(config.entrypoint),
                "command": replay_command,
                "argv": replay_command_args(config),
                "bundle_dir": bundle_dir.to_string_lossy(),
                "engine_replay": {
                    "mode": "bundle_offline",
                    "requires_http_server": false,
                    "command": engine_replay_command,
                    "argv": engine_replay_args
                },
                "sanitized": true
            }),
        ),
    ];
    for (name, value) in files {
        let path = bundle_dir.join(name);
        write_json(&path, &value)?;
        written.push(path);
    }
    if let Some(diagnostics) = data.failure_diagnostics {
        let path = bundle_dir.join("failure_diagnostics.json");
        write_json(&path, &diagnostics)?;
        written.push(path);
    }
    let output_text_path = bundle_dir.join("output_text.txt");
    fs_write(&output_text_path, output_text_body)?;
    written.push(output_text_path);
    Ok(written)
}

fn actual_run_failure_diagnostics(observation: &ActualRunFailureObservation) -> serde_json::Value {
    if resource_failure_kind(&observation.failure_kind) {
        return actual_run_resource_failure_diagnostics(observation);
    }
    json!({
        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        "request_id": observation.request_id,
        "failure_kind": observation.failure_kind,
        "first_failure_event": {
            "phase": "actual_run_generation_failed",
            "error_kind": observation.error_kind,
            "message": observation.error_message
        },
        "nearest_request_id": observation.request_id,
        "log_excerpt": observation.error_message
    })
}

fn actual_run_resource_failure_diagnostics(
    observation: &ActualRunFailureObservation,
) -> serde_json::Value {
    let memory_current = observation
        .memory
        .as_ref()
        .map(|memory| memory.current_bytes as i64)
        .unwrap_or(0)
        .max(0);
    let memory_high_water = observation
        .memory
        .as_ref()
        .map(|memory| memory.high_water_bytes as i64)
        .unwrap_or(memory_current);
    let resource_kind = resource_kind_for_failure(&observation.failure_kind);
    let needed = if resource_kind == "device_memory" {
        memory_current.saturating_add(1).max(1)
    } else {
        observation
            .prompt_token_count
            .and_then(|tokens| i64::try_from(tokens).ok())
            .unwrap_or(1)
            .max(1)
    };
    let capacity = if resource_kind == "device_memory" {
        memory_high_water.max(memory_current)
    } else {
        0
    };
    json!({
        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        "request_id": observation.request_id,
        "failure_kind": observation.failure_kind,
        "first_failure_event": {
            "phase": "actual_run_generation_failed",
            "error_kind": observation.error_kind,
            "message": observation.error_message
        },
        "nearest_request_id": observation.request_id,
        "log_excerpt": observation.error_message,
        "capacity": {
            "resource_kind": resource_kind,
            "needed": needed,
            "available": 0,
            "capacity": capacity,
            "reason": observation.error_message
        },
        "nearest_resource_event": {
            "owner_kind": "request",
            "owner_id": observation.request_id,
            "resource_kind": resource_kind,
            "action": "reject",
            "amount": needed,
            "before": 0,
            "after": 0,
            "capacity": capacity,
            "reason": observation.error_message
        },
        "nearest_memory_snapshot": {
            "scope": "actual_run_failure",
            "backend": "process",
            "current_bytes": memory_current,
            "high_water_bytes": memory_high_water.max(memory_current),
            "source": observation.memory.as_ref().map(|memory| memory.source).unwrap_or("not_collected")
        }
    })
}

fn resource_failure_kind(failure_kind: &str) -> bool {
    matches!(
        failure_kind,
        "oom" | "prevented_oom" | "admission" | "admission_reject" | "oom_admission"
    )
}

fn resource_kind_for_failure(failure_kind: &str) -> &'static str {
    match failure_kind {
        "oom" | "prevented_oom" => "device_memory",
        _ => "admission_capacity",
    }
}

fn bad_output_scan(
    request_id: &str,
    text: &str,
    failure_kind: Option<&str>,
    output_artifact_bytes: &[u8],
) -> serde_json::Value {
    let mut reasons = Vec::new();
    let mut first_span: Option<serde_json::Value> = None;
    for (needle, reason) in [
        ("<unk>", "reserved_token"),
        ("[PAD", "reserved_token"),
        ("<pad>", "reserved_token"),
        ("<|endoftext|>", "reserved_token"),
        ("<|im_start|>", "reserved_token"),
        ("<|im_end|>", "reserved_token"),
        ("<|reserved_special_token", "reserved_token"),
    ] {
        if let Some(index) = text.find(needle) {
            reasons.push(reason);
            first_span.get_or_insert_with(|| {
                json!({
                    "byte_start": index,
                    "byte_end": index + needle.len(),
                    "text": needle,
                    "reason": reason
                })
            });
        }
    }
    if let Some(index) = first_mojibake_index(text) {
        reasons.push("mojibake");
        first_span.get_or_insert_with(|| {
            json!({
                "byte_start": index,
                "byte_end": index + 1,
                "reason": "mojibake"
            })
        });
    }
    reasons.sort_unstable();
    reasons.dedup();
    let bad_output = !reasons.is_empty();
    json!({
        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        "request_id": request_id,
        "bad_output": bad_output,
        "bad_text_count": if bad_output { 1 } else { 0 },
        "reasons": reasons,
        "first_bad_text_span": first_span,
        "failure_kind": failure_kind,
        "output_chars": text.chars().count(),
        "classified_output_sha256": sha256_hex(text.as_bytes()),
        "output_sha256": sha256_hex(output_artifact_bytes)
    })
}

fn first_mojibake_index(text: &str) -> Option<usize> {
    ["\u{00c3}\u{00a9}", "\u{00c2}\u{00a9}", "\u{00e2}\u{20ac}"]
        .iter()
        .filter_map(|needle| text.find(needle))
        .min()
}

fn sha256_hex(bytes: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    format!("{:x}", hasher.finalize())
}

fn request_dump(
    config: &ProductObservabilityConfig,
    request_id: &str,
    replay_command: &str,
) -> serde_json::Value {
    let entrypoint = entrypoint_label(config.entrypoint);
    match config.entrypoint {
        ProfileEntrypoint::Run => json!({
            "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
            "entrypoint": entrypoint,
            "request_id": request_id,
            "model": config.model,
            "backend": SYNTHETIC_BACKEND,
            "profile_detail": config.profile_detail.as_str(),
            "profile_sample_rate": config.profile_sample_rate,
            "l0_only": true,
            "sanitized": true,
            "prompt": "product observability wiring",
            "replay_command": replay_command
        }),
        ProfileEntrypoint::Serve => json!({
            "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
            "entrypoint": entrypoint,
            "request_id": request_id,
            "model": config.model,
            "backend": SYNTHETIC_BACKEND,
            "profile_detail": config.profile_detail.as_str(),
            "profile_sample_rate": config.profile_sample_rate,
            "l0_only": true,
            "sanitized": true,
            "http": {
                "method": "POST",
                "path": "/v1/chat/completions",
                "body": {
                    "model": config.model,
                    "messages": [{"role": "user", "content": "product observability wiring"}],
                    "stream": false
                }
            },
            "replay_command": replay_command
        }),
        other => json!({
            "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
            "entrypoint": entrypoint_label(other),
            "request_id": request_id,
            "model": config.model,
            "backend": SYNTHETIC_BACKEND,
            "profile_detail": config.profile_detail.as_str(),
            "profile_sample_rate": config.profile_sample_rate,
            "l0_only": true,
            "sanitized": true,
            "replay_command": replay_command
        }),
    }
}

fn actual_request_dump(
    config: &ProductObservabilityConfig,
    request_id: &str,
    replay_command: &str,
) -> serde_json::Value {
    json!({
        "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
        "entrypoint": entrypoint_label(config.entrypoint),
        "request_id": request_id,
        "model": config.model,
        "backend": "actual",
        "profile_detail": config.profile_detail.as_str(),
        "profile_sample_rate": config.profile_sample_rate,
        "l0_only": false,
        "actual_model_smoke": true,
        "sanitized": true,
        "replay_command": replay_command
    })
}

fn replay_command(config: &ProductObservabilityConfig) -> String {
    replay_command_from_args(&replay_command_args(config))
}

fn replay_command_from_args(args: &[String]) -> String {
    args.iter()
        .map(|part| shell_quote(part))
        .collect::<Vec<_>>()
        .join(" ")
}

fn replay_command_args(config: &ProductObservabilityConfig) -> Vec<String> {
    let mut parts = vec![
        "cargo".to_string(),
        "run".to_string(),
        "-p".to_string(),
        "ferrum-cli".to_string(),
        "--".to_string(),
        entrypoint_label(config.entrypoint).to_string(),
        SYNTHETIC_MODEL.to_string(),
        "--profile-detail".to_string(),
        config.profile_detail.as_str().to_string(),
        "--profile-sample-rate".to_string(),
        config.profile_sample_rate.to_string(),
    ];
    push_path_arg(&mut parts, "--profile-jsonl", config.profile_jsonl.as_ref());
    push_path_arg(
        &mut parts,
        "--memory-profile-jsonl",
        config.memory_profile_jsonl.as_ref(),
    );
    push_path_arg(
        &mut parts,
        "--scheduler-trace-jsonl",
        config.scheduler_trace_jsonl.as_ref(),
    );
    push_path_arg(
        &mut parts,
        "--request-dump-dir",
        config.request_dump_dir.as_ref(),
    );
    parts
}

fn engine_replay_command_args(bundle_dir: &Path) -> Vec<String> {
    vec![
        "cargo".to_string(),
        "run".to_string(),
        "-p".to_string(),
        "ferrum-cli".to_string(),
        "--".to_string(),
        "replay-bundle".to_string(),
        bundle_dir.to_string_lossy().to_string(),
        "--out".to_string(),
        bundle_dir
            .join("engine_replay")
            .to_string_lossy()
            .to_string(),
        "--json".to_string(),
    ]
}

fn push_path_arg(parts: &mut Vec<String>, flag: &str, path: Option<&PathBuf>) {
    if let Some(path) = path {
        parts.push(flag.to_string());
        parts.push(path.to_string_lossy().to_string());
    }
}

#[cfg(test)]
fn write_profile_jsonl(path: &Path, events: &[FerrumProfileEvent]) -> Result<()> {
    write_profile_events(
        path,
        ferrum_bench_core::JsonlJournalOpenMode::Truncate,
        events,
    )
}

#[cfg(test)]
fn append_profile_jsonl(path: &Path, events: &[FerrumProfileEvent]) -> Result<()> {
    write_profile_events(
        path,
        ferrum_bench_core::JsonlJournalOpenMode::Append,
        events,
    )
}

fn write_profile_events(
    path: &Path,
    mode: ferrum_bench_core::JsonlJournalOpenMode,
    events: &[FerrumProfileEvent],
) -> Result<()> {
    if events.is_empty() {
        return Err(FerrumError::internal("profile event set must be non-empty"));
    }
    for event in events {
        event.validate().map_err(|err| {
            FerrumError::internal(format!("invalid product observability event: {err}"))
        })?;
    }
    ferrum_bench_core::write_jsonl_records(path, mode, events)
        .map_err(|error| FerrumError::io(error.to_string()))
}

fn write_json(path: &Path, value: &serde_json::Value) -> Result<()> {
    let body = serde_json::to_string_pretty(value)
        .map_err(|err| FerrumError::serialization(format!("failed to serialize JSON: {err}")))?;
    fs_write(path, format!("{body}\n"))
}

fn fs_create_dir_all(path: &Path) -> Result<()> {
    fs::create_dir_all(path)
        .map_err(|err| FerrumError::io(format!("failed to create {}: {err}", path.display())))
}

fn fs_write(path: &Path, content: impl AsRef<[u8]>) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs_create_dir_all(parent)?;
    }
    fs::write(path, content)
        .map_err(|err| FerrumError::io(format!("failed to write {}: {err}", path.display())))
}

fn entrypoint_label(entrypoint: ProfileEntrypoint) -> &'static str {
    match entrypoint {
        ProfileEntrypoint::Run => "run",
        ProfileEntrypoint::Serve => "serve",
        ProfileEntrypoint::BenchServe => "bench_serve",
        ProfileEntrypoint::Synthetic => "synthetic",
    }
}

fn shell_quote(value: &str) -> String {
    if value
        .chars()
        .all(|ch| ch.is_ascii_alphanumeric() || "-_./:".contains(ch))
    {
        return value.to_string();
    }
    format!("'{}'", value.replace('\'', "'\\''"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn engine_cache_observation_does_not_report_static_weights_as_kv() {
        let status = ferrum_types::EngineStatus {
            is_ready: true,
            loaded_models: vec![ferrum_types::ModelId::from("test-model")],
            active_requests: 0,
            queued_requests: 0,
            memory_usage: ferrum_types::MemoryUsage {
                total_bytes: 900,
                used_bytes: 500,
                free_bytes: 400,
                gpu_memory_bytes: Some(500),
                cpu_memory_bytes: None,
                cache_memory_bytes: 100,
                utilization_percent: 500.0 / 9.0,
            },
            uptime_seconds: 0,
            last_heartbeat: chrono::Utc::now(),
            version: "test".to_string(),
        };

        let observation = ActualMemoryStageObservation::new("test", "cache_allocated", None, None)
            .with_engine_cache_status(&status);

        assert_eq!(observation.attributes["kv_cache_total_bytes"], 500);
        assert_eq!(observation.attributes["kv_cache_used_bytes"], 100);
        assert_eq!(observation.attributes["resource_total_bytes"], 900);
        assert_eq!(observation.attributes["resource_used_bytes"], 500);
    }

    #[test]
    fn synthetic_product_observability_writes_profile_paths() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-product-observability-{}",
            Uuid::new_v4().simple()
        ));
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Run,
            SYNTHETIC_MODEL,
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Basic,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let written = write_synthetic_product_observability(&config).unwrap();
        assert!(written.len() >= 14);
        assert!(root.join("profile.jsonl").is_file());
        assert!(root.join("memory.jsonl").is_file());
        assert!(root.join("scheduler.jsonl").is_file());
        assert!(root.join("request_dump/request.json").is_file());
        assert!(root.join("request_dump/replay_command.txt").is_file());
        let request_dump_root = root.join("request_dump");
        let bundle_dir = fs::read_dir(&request_dump_root)
            .unwrap()
            .flatten()
            .find_map(|entry| entry.path().is_dir().then_some(entry.path()))
            .expect("request-id replay bundle directory should exist");
        assert!(bundle_dir.join("prompt_token_ids.json").is_file());
        assert!(bundle_dir.join("sampling_params.json").is_file());
        assert!(bundle_dir.join("runtime_effective_config.json").is_file());
        assert!(bundle_dir.join("backend_selection.json").is_file());
        assert!(bundle_dir.join("output_token_ids.json").is_file());
        assert!(bundle_dir.join("output_text.txt").is_file());
        assert!(bundle_dir.join("bad_output_scan.json").is_file());
        assert!(bundle_dir.join("replay.command.json").is_file());
        let replay: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("replay.command.json")).unwrap(),
        )
        .unwrap();
        let engine_argv = replay["engine_replay"]["argv"].as_array().unwrap();
        assert!(engine_argv.iter().any(|item| item == "replay-bundle"));
        assert_eq!(replay["engine_replay"]["requires_http_server"], false);
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn aliased_profile_roles_write_each_event_once() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-product-observability-alias-{}",
            Uuid::new_v4().simple()
        ));
        let combined = root.join("combined.jsonl");
        let memory_alias = root.join(".").join("combined.jsonl");
        let scheduler_alias = root.join("not-created").join("..").join("combined.jsonl");
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Run,
            SYNTHETIC_MODEL,
            Some(&combined),
            ProfileDetailArg::Resource,
            Some(&memory_alias),
            Some(&scheduler_alias),
            None,
            1.0,
        );

        let written = write_synthetic_product_observability(&config).unwrap();
        assert_eq!(written.len(), 1);
        let events = fs::read_to_string(&combined)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(
            events.len(),
            6,
            "aliased roles duplicated events: {events:#?}"
        );
        assert_eq!(
            events
                .iter()
                .filter_map(|event| event["event_id"].as_str())
                .collect::<std::collections::BTreeSet<_>>()
                .len(),
            6
        );
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn actual_serve_memory_stage_observability_appends_shutdown() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-serve-memory-observability-{}",
            Uuid::new_v4().simple()
        ));
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Serve,
            "Qwen/Qwen3-0.6B",
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Basic,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let memory = crate::memory_profile::ProcessMemoryObservation {
            before_bytes: 100,
            after_bytes: 200,
            current_bytes: 200,
            high_water_bytes: 240,
            source: "test",
        };
        write_actual_serve_startup_observability(
            &config,
            42,
            Some(memory.clone()),
            vec![
                ActualMemoryStageObservation::new(
                    "actual_serve_process_start",
                    "process_start",
                    None,
                    Some(memory.clone()),
                ),
                ActualMemoryStageObservation::new(
                    "actual_serve_backend_initialized",
                    "backend_initialized",
                    None,
                    Some(memory.clone()),
                ),
            ],
        )
        .unwrap();
        append_actual_serve_memory_stage_observability(
            &config,
            ActualMemoryStageObservation::new(
                "actual_serve_shutdown",
                "shutdown",
                None,
                Some(memory),
            ),
        )
        .unwrap();
        let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
        assert!(profile.contains("\"phase\":\"actual_serve_startup\""));
        assert!(profile.contains("\"phase\":\"actual_serve_shutdown\""));
        let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
        let memory_stages = memory_profile
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .filter_map(|event| {
                event["attributes"]["memory_stage"]
                    .as_str()
                    .map(str::to_string)
            })
            .collect::<Vec<_>>();
        assert!(memory_stages.contains(&"process_start".to_string()));
        assert!(memory_stages.contains(&"backend_initialized".to_string()));
        assert!(memory_stages.contains(&"model_loaded".to_string()));
        assert!(memory_stages.contains(&"shutdown".to_string()));
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn profile_jsonl_append_is_parseable_under_concurrent_writers() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-profile-concurrent-append-{}",
            Uuid::new_v4().simple()
        ));
        let path = root.join("profile.jsonl");
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Serve,
            "Qwen/Qwen3-0.6B",
            Some(&path),
            ProfileDetailArg::Basic,
            None,
            None,
            Some(&root.join("request_dump")),
            1.0,
        );

        let mut handles = Vec::new();
        for writer in 0..8 {
            let path = path.clone();
            let config = config.clone();
            handles.push(std::thread::spawn(move || {
                for seq in 0..32 {
                    let request_id = format!("req-{writer}-{seq}");
                    let mut event = actual_base_event(
                        &config,
                        &request_id,
                        "chat_completions_sync_complete",
                        ProfileEventKind::TimedSpan,
                        Utc::now(),
                    );
                    event.event_id = format!("evt-{writer}-{seq}");
                    event.correlation_id = Some(format!("corr-{writer}-{seq}"));
                    event.duration_us = Some(1);
                    event.attributes.insert("writer".to_string(), json!(writer));
                    event.attributes.insert("seq".to_string(), json!(seq));
                    append_profile_jsonl(&path, &[event]).unwrap();
                }
            }));
        }

        for handle in handles {
            handle.join().unwrap();
        }

        let profile = fs::read_to_string(&path).unwrap();
        let mut parsed = 0usize;
        for line in profile.lines().filter(|line| !line.trim().is_empty()) {
            serde_json::from_str::<serde_json::Value>(line).unwrap();
            parsed += 1;
        }
        assert_eq!(parsed, 8 * 32);
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn actual_serve_startup_memory_orders_model_profile_and_cache_stages() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-serve-memory-order-{}",
            Uuid::new_v4().simple()
        ));
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Serve,
            "Qwen/Qwen3-0.6B",
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Basic,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let memory = crate::memory_profile::ProcessMemoryObservation {
            before_bytes: 100,
            after_bytes: 200,
            current_bytes: 200,
            high_water_bytes: 240,
            source: "test",
        };
        write_actual_serve_startup_observability(
            &config,
            42,
            Some(memory.clone()),
            vec![
                ActualMemoryStageObservation::new(
                    "actual_serve_process_start",
                    "process_start",
                    None,
                    Some(memory.clone()),
                ),
                ActualMemoryStageObservation::new(
                    "actual_serve_backend_initialized",
                    "backend_initialized",
                    None,
                    Some(memory.clone()),
                ),
                ActualMemoryStageObservation::new(
                    "actual_serve_profile_run_done",
                    "profile_run_done",
                    None,
                    Some(memory.clone()),
                )
                .with_profile_run_status(false, "not_configured", "test"),
                ActualMemoryStageObservation::new(
                    "actual_serve_cache_allocated",
                    "cache_allocated",
                    None,
                    Some(memory),
                )
                .with_attribute("available_kv_or_state_bytes", json!(0)),
            ],
        )
        .unwrap();
        let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
        let memory_stages = memory_profile
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .filter_map(|event| {
                event["attributes"]["memory_stage"]
                    .as_str()
                    .map(str::to_string)
            })
            .collect::<Vec<_>>();
        assert_eq!(
            memory_stages,
            vec![
                "process_start",
                "backend_initialized",
                "model_loaded",
                "profile_run_done",
                "cache_allocated"
            ]
        );
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn actual_run_failure_observability_writes_diagnostics_bundle() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-run-failure-observability-{}",
            Uuid::new_v4().simple()
        ));
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Run,
            "Qwen/Qwen3-0.6B",
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Basic,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let request_id = "req-failure-test".to_string();
        write_actual_run_failure_observability(
            &config,
            &ActualRunFailureObservation {
                request_id: request_id.clone(),
                duration_us: 42,
                sampling_params: SamplingParams::greedy(),
                prompt_token_ids: Some(vec![11, 22, 33]),
                prompt_token_count: Some(3),
                prompt_chars: 12,
                failure_kind: "error".to_string(),
                error_kind: "error".to_string(),
                error_message: "synthetic failure".to_string(),
                memory: None,
                memory_stages: Vec::new(),
            },
        )
        .unwrap();
        let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
        let failure = profile
            .lines()
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .find(|event| event["phase"] == "actual_run_generation_failed")
            .expect("run failure profile event");
        assert_eq!(failure["event_kind"], "timed_span");
        assert_eq!(failure["status"], "failure");
        assert_eq!(failure["duration_us"], 42);
        assert_eq!(failure["attributes"]["terminal_failure_event"], true);
        assert!(failure["attributes"]["first_failure_event"].is_null());
        let bundle_dir = root.join("request_dump").join(&request_id);
        assert!(bundle_dir.join("failure_diagnostics.json").is_file());
        let scan: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(scan["failure_kind"], "error");
        let diagnostics: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("failure_diagnostics.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(diagnostics["failure_kind"], "error");
        assert_eq!(
            diagnostics["first_failure_event"]["phase"],
            "actual_run_generation_failed"
        );
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn actual_run_observability_writes_prompt_token_ids() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-run-observability-{}",
            Uuid::new_v4().simple()
        ));
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Run,
            "Qwen/Qwen3-0.6B",
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Latency,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let request_id = "req-run-test".to_string();
        let native_event = actual_base_event(
            &config,
            &request_id,
            "vnext.request_accepted",
            ProfileEventKind::Instant,
            Utc::now(),
        );
        write_profile_jsonl(&root.join("profile.jsonl"), &[native_event]).unwrap();
        write_actual_run_observability(
            &config,
            &ActualRunObservation {
                request_id: request_id.clone(),
                duration_us: 42,
                sampling_params: SamplingParams::greedy(),
                prompt_token_ids: Some(vec![7, 8, 9]),
                prompt_token_count: Some(3),
                output_tokens: 2,
                output_token_ids: vec![10, 11],
                chunk_count: 1,
                finish_reason: Some("stop".to_string()),
                prompt_chars: 12,
                response_chars: 2,
                response_text: "OK".to_string(),
                execution_evidence: Some(ferrum_types::InferenceExecutionEvidence {
                    prompt_token_ids: vec![
                        ferrum_types::TokenId::new(7),
                        ferrum_types::TokenId::new(8),
                        ferrum_types::TokenId::new(9),
                    ],
                    output_token_ids: vec![
                        ferrum_types::TokenId::new(10),
                        ferrum_types::TokenId::new(11),
                    ],
                    engine_token_timing: Some(ferrum_types::EngineTokenTimingEvidence {
                        clock_source: "rust_std_instant".to_string(),
                        wall_anchor_unix_nanos: 1_700_000_000_000_000_000,
                        wall_anchor_max_error_nanos: 500,
                        decode_ready_nanos_since_request_start: Some(1_000_000),
                        token_commit_nanos_since_request_start: vec![2_000_000, 4_000_000],
                        decode_stage_intervals: Vec::new(),
                    }),
                }),
                memory: None,
                memory_stages: vec![
                    ActualMemoryStageObservation::new(
                        "actual_run_process_start",
                        "process_start",
                        None,
                        Some(crate::memory_profile::ProcessMemoryObservation {
                            before_bytes: 100,
                            after_bytes: 100,
                            current_bytes: 100,
                            high_water_bytes: 120,
                            source: "test",
                        }),
                    ),
                    ActualMemoryStageObservation::new(
                        "actual_run_shutdown",
                        "shutdown",
                        None,
                        Some(crate::memory_profile::ProcessMemoryObservation {
                            before_bytes: 200,
                            after_bytes: 220,
                            current_bytes: 220,
                            high_water_bytes: 240,
                            source: "test",
                        }),
                    ),
                ],
            },
        )
        .unwrap();
        let bundle_dir = root.join("request_dump").join(&request_id);
        let prompt_tokens: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("prompt_token_ids.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(prompt_tokens["token_ids"], serde_json::json!([7, 8, 9]));
        assert_eq!(prompt_tokens["token_count"], 3);
        assert!(prompt_tokens["unavailable_reason"].is_null());
        let output_text_bytes = fs::read(bundle_dir.join("output_text.txt")).unwrap();
        let output_text = String::from_utf8(output_text_bytes.clone()).unwrap();
        assert!(output_text.starts_with("[redacted actual output]\n"));
        let scan: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(
            scan["output_sha256"],
            sha256_hex(output_text_bytes.as_slice())
        );
        assert_eq!(scan["classified_output_sha256"], sha256_hex(b"OK"));
        let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
        assert!(profile.lines().any(|line| {
            serde_json::from_str::<serde_json::Value>(line)
                .is_ok_and(|event| event["phase"] == "vnext.request_accepted")
        }));
        let generation = profile
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .find(|event| event["phase"] == "actual_run_generation")
            .expect("actual run generation event");
        assert_eq!(generation["attributes"]["prompt_token_count"], 3);
        assert_eq!(generation["attributes"]["completion_token_count"], 2);
        assert_eq!(generation["attributes"]["output_token_count"], 2);
        assert_eq!(generation["attributes"]["total_token_count"], 5);
        assert_eq!(
            generation["attributes"]["token_count_source"],
            "rendered_prompt_and_generated_tokens"
        );
        assert_eq!(generation["attributes"]["e2e_duration_us"], 42);
        assert_eq!(generation["attributes"]["profile_detail"], "latency");
        assert_eq!(
            generation["attributes"]["engine_token_commit_nanos_since_request_start"],
            serde_json::json!([2_000_000, 4_000_000])
        );
        assert_eq!(generation["attributes"]["engine_token_commit_count"], 2);
        assert_eq!(
            generation["attributes"]["itl_source"],
            "engine_token_commit"
        );
        assert_eq!(
            generation["attributes"]["itl_nanos"],
            serde_json::json!([2_000_000])
        );
        assert_eq!(generation["attributes"]["ttft_us"], 2_000);
        assert_eq!(generation["attributes"]["itl_us_avg"], 2_000);
        assert_eq!(
            generation["attributes"]["engine_decode_ready_nanos_since_request_start"],
            1_000_000
        );
        assert_eq!(
            generation["attributes"]["engine_decode_wall_nanos"],
            3_000_000
        );
        assert_eq!(generation["attributes"]["clock_conversion_error_ppm"], 166);
        assert_eq!(
            generation["attributes"]["decode_wall_timing_eligible"],
            true
        );
        let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
        let memory_stages = memory_profile
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
            .filter_map(|event| {
                event["attributes"]["memory_stage"]
                    .as_str()
                    .map(str::to_string)
            })
            .collect::<Vec<_>>();
        assert!(memory_stages.contains(&"process_start".to_string()));
        assert!(memory_stages.contains(&"first_request_done".to_string()));
        let first_request_index = memory_stages
            .iter()
            .position(|stage| stage == "first_request_done")
            .expect("first_request_done stage");
        let shutdown_index = memory_stages
            .iter()
            .position(|stage| stage == "shutdown")
            .expect("shutdown stage");
        assert!(first_request_index < shutdown_index);
        assert!(
            !root.join("scheduler.jsonl").exists(),
            "the engine owns successful actual-run scheduler lifecycles"
        );
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn actual_run_resource_failure_observability_writes_resource_diagnostics() {
        let root = std::env::temp_dir().join(format!(
            "ferrum-observability-resource-failure-{}",
            uuid::Uuid::new_v4()
        ));
        fs::create_dir_all(&root).unwrap();
        let config = ProductObservabilityConfig::new(
            ProfileEntrypoint::Run,
            "Qwen/Qwen3-0.6B",
            Some(&root.join("profile.jsonl")),
            ProfileDetailArg::Basic,
            Some(&root.join("memory.jsonl")),
            Some(&root.join("scheduler.jsonl")),
            Some(&root.join("request_dump")),
            1.0,
        );
        let request_id = "req-resource-failure-test".to_string();
        write_actual_run_failure_observability(
            &config,
            &ActualRunFailureObservation {
                request_id: request_id.clone(),
                duration_us: 42,
                sampling_params: SamplingParams::greedy(),
                prompt_token_ids: Some(vec![1, 2, 3, 4]),
                prompt_token_count: Some(4),
                prompt_chars: 128,
                failure_kind: "oom_admission".to_string(),
                error_kind: "resource_exhausted".to_string(),
                error_message: "Resource exhausted: recurrent state capacity exhausted".to_string(),
                memory: Some(crate::memory_profile::ProcessMemoryObservation {
                    before_bytes: 1024,
                    after_bytes: 2048,
                    current_bytes: 2048,
                    high_water_bytes: 4096,
                    source: "test",
                }),
                memory_stages: Vec::new(),
            },
        )
        .unwrap();
        let bundle_dir = root.join("request_dump").join(&request_id);
        let scan: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(scan["failure_kind"], "oom_admission");
        let diagnostics: serde_json::Value = serde_json::from_str(
            &fs::read_to_string(bundle_dir.join("failure_diagnostics.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(diagnostics["failure_kind"], "oom_admission");
        assert_eq!(
            diagnostics["capacity"]["resource_kind"],
            "admission_capacity"
        );
        assert_eq!(diagnostics["nearest_resource_event"]["action"], "reject");
        assert_eq!(
            diagnostics["nearest_memory_snapshot"]["current_bytes"],
            2048
        );
        fs::remove_dir_all(root).ok();
    }
}