ferrum-cli 0.12.1

//! Serve command - Start the HTTP inference server

use crate::config::CliConfig;
use crate::runtime_env::runtime_snapshot_value;
use clap::Args;
use colored::*;
use ferrum_bench_core::{ProfileMetadata, ProfileSinkConfig};
use ferrum_models::source::ModelFormat;
use ferrum_server::{AxumServer, HttpServer, ServedModelKind, ServedModelRegistry, ServerConfig};
use ferrum_types::{
    CompiledKernelFeatures, CompiledNativeOperatorArtifact, FerrumError, HardwareCapabilities,
    ModelCapabilities, ResolvedFerrumConfig, Result, RuntimeConfigEntry, RuntimeConfigSnapshot,
    RuntimeConfigSource, WorkloadProfile, M3_QWEN3_30B_A3B_INT4_PRESET,
    QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET,
};
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use std::sync::Arc;
use std::time::Duration;
use tokio::signal;

#[derive(Args)]
pub struct ServeCommand {
    /// Model alias, owner/repository[:QUANT], or local path (default: from config)
    #[arg(value_name = "MODEL")]
    pub model: Option<String>,

    /// Model alias, owner/repository[:QUANT], or local path (default: from config)
    #[arg(
        short = 'm',
        long = "model",
        value_name = "MODEL",
        conflicts_with = "model"
    )]
    pub model_option: Option<String>,

    #[command(flatten)]
    pub product_sources: crate::source_resolver::ProductSourceArgs,

    /// Public OpenAI-compatible model names. The first name is primary and
    /// additional names are aliases for the same loaded model.
    #[arg(
        long = "served-model-name",
        value_name = "NAME",
        value_delimiter = ',',
        action = clap::ArgAction::Append
    )]
    pub served_model_name: Vec<String>,

    /// Enable model reasoning by default unless a request supplies standard
    /// reasoning effort or `chat_template_kwargs.enable_thinking`.
    #[arg(long, conflicts_with = "disable_thinking")]
    pub enable_thinking: bool,

    /// Disable model reasoning by default unless a request supplies standard
    /// reasoning effort or `chat_template_kwargs.enable_thinking`.
    #[arg(long, conflicts_with = "enable_thinking")]
    pub disable_thinking: bool,

    /// Host to bind to
    #[arg(long)]
    pub host: Option<String>,

    /// Port to listen on
    #[arg(short, long)]
    pub port: Option<u16>,

    /// Number of TTS concurrent slots (default: 2)
    #[arg(long, default_value = "2")]
    pub tts_slots: usize,

    /// Backend: auto, cpu, metal, cuda.
    #[arg(long, default_value = "auto")]
    pub backend: String,

    /// Numerical execution profile: auto or an exact family profile ID.
    #[arg(long, value_name = "PROFILE")]
    pub numerical_profile: Option<ferrum_types::NumericalExecutionPolicy>,

    /// CUDA GPU ids to use, comma-separated. Multi-GPU requests select
    /// layer-split for supported Llama-family safetensors models.
    #[arg(long, value_name = "IDS")]
    pub gpu_devices: Option<String>,

    /// Layer-split decode pipeline mode for multi-GPU CUDA serving.
    #[arg(long, value_enum)]
    pub layer_split_pipeline_mode: Option<crate::layer_split_pipeline::LayerSplitPipelineModeArg>,

    /// Speculative decoding: draft model id (same family as target).
    /// Example: `--spec-draft qwen3:0.6b` when serving `qwen3:4b`.
    /// The draft model must share the tokenizer + vocabulary.
    #[arg(long, value_name = "MODEL")]
    pub spec_draft: Option<String>,

    /// Number of speculative tokens per draft forward pass (default: 4).
    /// Only active when --spec-draft is set.
    #[arg(long, default_value = "4")]
    pub spec_tokens: usize,

    /// Fraction of currently available device memory used by native startup
    /// planning (default: 0.9). Legacy CUDA models use this fraction of total
    /// GPU memory for their KV pool estimate.
    #[arg(long, default_value = "0.9")]
    pub gpu_memory_utilization: f32,

    /// Exact device-wide memory budget available to runtime weights and
    /// dynamic resources. This is the same typed ceiling used by `run`.
    #[arg(long, value_name = "BYTES")]
    pub runtime_memory_budget_bytes: Option<std::num::NonZeroUsize>,

    /// vLLM-compatible alias for `FERRUM_MAX_MODEL_LEN`.
    #[arg(long, value_name = "N")]
    pub max_model_len: Option<usize>,

    /// vLLM-compatible alias for `FERRUM_PAGED_MAX_SEQS`.
    #[arg(long, value_name = "N")]
    pub max_num_seqs: Option<usize>,

    /// vLLM-compatible alias for `FERRUM_MAX_BATCHED_TOKENS`.
    #[arg(long, value_name = "N")]
    pub max_num_batched_tokens: Option<usize>,

    /// Sequence fit gate used before prefill admission.
    #[arg(long, value_enum)]
    pub sequence_fit_policy: Option<crate::commands::SequenceFitPolicyArg>,

    /// Prefer prefilling until this many requests are active before early decodes.
    #[arg(long, value_name = "N")]
    pub scheduler_prefill_first_until_active: Option<usize>,

    /// Cap per-request scheduler prefill chunks before they enter the engine.
    #[arg(long, value_name = "N")]
    pub scheduler_prefill_step_chunk: Option<usize>,

    /// Cap prefill chunks while decode requests are active.
    #[arg(long, value_name = "N")]
    pub scheduler_active_decode_prefill_chunk: Option<usize>,

    /// Enable prefix caching (`FERRUM_PREFIX_CACHE=1`).
    #[arg(
        long,
        conflicts_with_all = ["no_enable_prefix_caching", "disable_prefix_cache"]
    )]
    pub enable_prefix_caching: bool,

    /// Disable prefix caching (`FERRUM_PREFIX_CACHE=0`).
    #[arg(long, conflicts_with = "enable_prefix_cache")]
    pub no_enable_prefix_caching: bool,

    /// Enable prefix cache (`FERRUM_PREFIX_CACHE=1`).
    #[arg(
        long,
        conflicts_with_all = ["no_enable_prefix_caching", "disable_prefix_cache"]
    )]
    pub enable_prefix_cache: bool,

    /// Disable prefix cache (`FERRUM_PREFIX_CACHE=0`).
    #[arg(long, conflicts_with_all = ["enable_prefix_caching", "enable_prefix_cache"])]
    pub disable_prefix_cache: bool,

    /// Session cache mode (`off` or `memory`).
    #[arg(long, value_name = "MODE", value_parser = ["off", "memory"])]
    pub session_cache: Option<String>,

    /// Maximum in-memory session cache entries.
    #[arg(long, value_name = "N")]
    pub session_cache_max_entries: Option<usize>,

    /// Approximate maximum tokens retained per session.
    #[arg(long, value_name = "N")]
    pub session_cache_max_tokens: Option<usize>,

    /// KV cache storage: `fp16` (default) or `int8` on supported model/backend
    /// combinations. INT8 uses separate per-token/head K/V scales; unsupported
    /// combinations fail during planning. Also configurable as runtime.kv_dtype.
    #[arg(long, value_name = "DTYPE")]
    pub kv_dtype: Option<String>,

    /// Per-sequence KV token capacity (`FERRUM_KV_CAPACITY`).
    #[arg(long, value_name = "N")]
    pub kv_capacity: Option<usize>,

    /// Global KV block budget (`FERRUM_KV_MAX_BLOCKS`).
    #[arg(long, value_name = "N")]
    pub kv_max_blocks: Option<usize>,

    /// Use GPU argmax for greedy decoding (`FERRUM_GREEDY_ARGMAX=1`).
    #[arg(long, conflicts_with = "disable_greedy_argmax")]
    pub greedy_argmax: bool,

    /// Disable GPU argmax for greedy decoding (`FERRUM_GREEDY_ARGMAX=0`).
    #[arg(long, conflicts_with = "greedy_argmax")]
    pub disable_greedy_argmax: bool,

    /// Enable legacy Llama/Gemma batched decode CUDA graph replay.
    #[arg(long, conflicts_with = "disable_batched_graph")]
    pub batched_graph: bool,

    /// Disable legacy Llama/Gemma batched decode CUDA graph replay.
    #[arg(long, conflicts_with = "batched_graph")]
    pub disable_batched_graph: bool,

    /// Enable vNext reusable device-program preparation.
    #[arg(long, conflicts_with = "disable_reusable_execution")]
    pub reusable_execution: bool,

    /// Disable vNext reusable device-program preparation.
    #[arg(long, conflicts_with = "reusable_execution")]
    pub disable_reusable_execution: bool,

    /// Enable Llama/Gemma unified decode CUDA graph replay.
    #[arg(long, conflicts_with = "disable_unified_graph")]
    pub unified_graph: bool,

    /// Disable Llama/Gemma unified decode CUDA graph replay.
    #[arg(long, conflicts_with = "unified_graph")]
    pub disable_unified_graph: bool,

    /// Capture only Llama/Gemma unified transformer layers in CUDA graph replay.
    #[arg(long, conflicts_with = "disable_unified_graph_layers_only")]
    pub unified_graph_layers_only: bool,

    /// Disable layers-only unified CUDA graph capture scope.
    #[arg(long, conflicts_with = "unified_graph_layers_only")]
    pub disable_unified_graph_layers_only: bool,

    /// Capture unified layers plus final packing; leave lm_head eager.
    #[arg(long, conflicts_with = "disable_unified_graph_lm_head_eager")]
    pub unified_graph_lm_head_eager: bool,

    /// Disable lm-head-eager unified CUDA graph capture scope.
    #[arg(long, conflicts_with = "unified_graph_lm_head_eager")]
    pub disable_unified_graph_lm_head_eager: bool,

    /// Named startup/runtime preset, for example
    /// `m3_qwen3_30b_a3b_int4`.
    #[arg(long, value_name = "PRESET")]
    pub runtime_preset: Option<String>,

    /// Write the startup effective runtime config JSON artifact.
    #[arg(long, value_name = "PATH")]
    pub effective_config_json: Option<PathBuf>,

    /// Write the startup auto-config decision trace JSONL artifact.
    #[arg(long, value_name = "PATH")]
    pub decision_trace_jsonl: Option<PathBuf>,

    /// Generate a synthetic/no-weight observability vertical-slice artifact and exit.
    #[arg(long, value_name = "DIR")]
    pub observability_vertical_slice_out: Option<PathBuf>,

    #[command(flatten)]
    pub vnext_checkpoint: crate::commands::vnext_checkpoint::VNextCheckpointArgs,

    /// Write native structured profile events to this JSONL path.
    #[arg(long, value_name = "PATH")]
    pub profile_jsonl: Option<PathBuf>,

    /// Product observability detail level.
    #[arg(long, value_enum, default_value_t = crate::observability_product::ProfileDetailArg::Off)]
    pub profile_detail: crate::observability_product::ProfileDetailArg,

    /// Inject one typed vNext diagnostic fault. Requires a latency profile.
    #[arg(long, value_enum)]
    pub vnext_diagnostic_fault: Option<crate::commands::VNextDiagnosticFaultArg>,

    /// Write product memory profile events to this JSONL path.
    #[arg(long, value_name = "PATH")]
    pub memory_profile_jsonl: Option<PathBuf>,

    /// Write scheduler iteration trace events to this JSONL path.
    #[arg(long, value_name = "PATH")]
    pub scheduler_trace_jsonl: Option<PathBuf>,

    /// Write a sanitized request/replay bundle to this directory.
    #[arg(long, value_name = "DIR")]
    pub request_dump_dir: Option<PathBuf>,

    /// Product observability sampling rate for resource lifecycle events.
    #[arg(long, default_value_t = crate::observability_product::default_profile_sample_rate())]
    pub profile_sample_rate: f64,

    /// Git commit stamped into native structured profile events.
    #[arg(long, value_name = "SHA")]
    pub profile_commit_sha: Option<String>,

    /// Runtime environment hash stamped into native structured profile events.
    #[arg(long, value_name = "SHA256")]
    pub profile_env_hash: Option<String>,

    /// Model label stamped into native structured profile events.
    #[arg(long, value_name = "MODEL")]
    pub profile_model: Option<String>,

    /// Concurrency stamped into native structured profile events.
    #[arg(long, value_name = "N")]
    pub profile_concurrency: Option<u32>,

    /// Runtime flags/config JSON object embedded in native profile events.
    #[arg(long, value_name = "JSON")]
    pub profile_runtime_flags_json: Option<String>,

    /// Startup-loaded LoRA adapter, formatted as NAME=PATH. May be repeated.
    #[arg(long = "lora", value_name = "NAME=PATH")]
    pub lora: Vec<String>,

    /// Public model id template for LoRA adapters. Supports <base> and <name>.
    #[arg(long, value_name = "TEMPLATE", default_value = "<base>:<name>")]
    pub lora_model_id_template: String,
}

/// Product CLI wrapper for serve-only compatibility controls. Keeping these
/// options outside [`ServeCommand`] avoids breaking existing library callers
/// that construct that public command with a struct literal.
#[derive(Args)]
pub struct ServeCliCommand {
    #[command(flatten)]
    command: ServeCommand,
    /// Wait at most this long for another request's exact in-flight prefix.
    #[arg(long, value_name = "MS")]
    prefix_rendezvous_max_wait_ms: Option<std::num::NonZeroU64>,

    /// Enable the compatibility retry that coalesces non-leading system
    /// messages when a model-owned chat template rejects their original order.
    #[arg(long, conflicts_with = "disable_interleaved_system_coalescing")]
    enable_interleaved_system_coalescing: bool,

    /// Disable the compatibility retry for non-leading system messages and
    /// return the model-owned template's original rendering error instead.
    #[arg(long, conflicts_with = "enable_interleaved_system_coalescing")]
    disable_interleaved_system_coalescing: bool,
}

fn resolve_interleaved_system_coalescing(
    enable: bool,
    disable: bool,
    configured_default: bool,
) -> bool {
    if enable {
        true
    } else if disable {
        false
    } else {
        configured_default
    }
}

pub async fn execute(cmd: ServeCommand, config: CliConfig) -> Result<()> {
    execute_with_compatibility(cmd, config, false, false, true, None).await
}

pub async fn execute_cli(
    cmd: ServeCliCommand,
    config: CliConfig,
    configured_interleaved_system_coalescing: bool,
) -> Result<()> {
    execute_with_compatibility(
        cmd.command,
        config,
        cmd.enable_interleaved_system_coalescing,
        cmd.disable_interleaved_system_coalescing,
        configured_interleaved_system_coalescing,
        cmd.prefix_rendezvous_max_wait_ms,
    )
    .await
}

async fn execute_with_compatibility(
    cmd: ServeCommand,
    config: CliConfig,
    enable_interleaved_system_coalescing: bool,
    disable_interleaved_system_coalescing: bool,
    configured_interleaved_system_coalescing: bool,
    prefix_rendezvous_max_wait_ms: Option<std::num::NonZeroU64>,
) -> Result<()> {
    let ServeCommand {
        model,
        model_option,
        product_sources,
        served_model_name,
        enable_thinking,
        disable_thinking,
        host,
        port,
        tts_slots,
        backend,
        numerical_profile,
        gpu_devices,
        layer_split_pipeline_mode,
        spec_draft,
        spec_tokens,
        gpu_memory_utilization,
        runtime_memory_budget_bytes,
        max_model_len,
        max_num_seqs,
        max_num_batched_tokens,
        sequence_fit_policy,
        scheduler_prefill_first_until_active,
        scheduler_prefill_step_chunk,
        scheduler_active_decode_prefill_chunk,
        enable_prefix_caching,
        no_enable_prefix_caching,
        enable_prefix_cache,
        disable_prefix_cache,
        session_cache,
        session_cache_max_entries,
        session_cache_max_tokens,
        kv_dtype,
        kv_capacity,
        kv_max_blocks,
        greedy_argmax,
        disable_greedy_argmax,
        batched_graph,
        disable_batched_graph,
        reusable_execution,
        disable_reusable_execution,
        unified_graph,
        disable_unified_graph,
        unified_graph_layers_only,
        disable_unified_graph_layers_only,
        unified_graph_lm_head_eager,
        disable_unified_graph_lm_head_eager,
        runtime_preset,
        effective_config_json,
        decision_trace_jsonl,
        observability_vertical_slice_out,
        vnext_checkpoint,
        profile_jsonl,
        profile_detail,
        vnext_diagnostic_fault,
        memory_profile_jsonl,
        scheduler_trace_jsonl,
        request_dump_dir,
        profile_sample_rate,
        profile_commit_sha,
        profile_env_hash,
        profile_model,
        profile_concurrency,
        profile_runtime_flags_json,
        lora,
        lora_model_id_template,
    } = cmd;

    let default_enable_thinking = if enable_thinking {
        Some(true)
    } else if disable_thinking {
        Some(false)
    } else {
        None
    };
    let interleaved_system_coalescing = resolve_interleaved_system_coalescing(
        enable_interleaved_system_coalescing,
        disable_interleaved_system_coalescing,
        configured_interleaved_system_coalescing,
    );

    // Reject malformed adapter specs before selecting a device or resolving,
    // downloading, and preparing the base model.
    let lora_specs = parse_lora_specs(&lora)?;

    if let Some(out_dir) = observability_vertical_slice_out.as_ref() {
        crate::observability_vertical_slice::write_observability_vertical_slice(
            ferrum_types::ProfileEntrypoint::Serve,
            out_dir,
        )?;
        println!(
            "OBSERVABILITY VERTICAL SLICE ARTIFACT: {}",
            out_dir.display()
        );
        return Ok(());
    }

    // Resolve model
    let model_name = model
        .or(model_option)
        .or_else(|| {
            config
                .models
                .default_model
                .clone()
                .filter(|model| !model.trim().is_empty())
        })
        .ok_or_else(|| {
            FerrumError::config(crate::source_resolver::first_success_model_help(
                "serve --model",
            ))
        })?;
    let serve_start = std::time::Instant::now();
    let product_observability = crate::observability_product::ProductObservabilityConfig::new(
        ferrum_types::ProfileEntrypoint::Serve,
        &model_name,
        profile_jsonl.as_ref(),
        profile_detail,
        memory_profile_jsonl.as_ref(),
        scheduler_trace_jsonl.as_ref(),
        request_dump_dir.as_ref(),
        profile_sample_rate,
    );
    let memory_sampler = crate::memory_profile::ProcessMemorySampler;
    let product_memory_enabled = product_observability.enabled();
    let process_start_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let process_start_memory = process_start_sample
        .clone()
        .map(crate::memory_profile::ProcessMemoryObservation::from_sample);
    if product_observability.synthetic_no_weight_enabled() {
        let written = crate::observability_product::write_synthetic_product_observability(
            &product_observability,
        )?;
        println!(
            "OBSERVABILITY PRODUCT ARTIFACTS: {}",
            written
                .iter()
                .map(|path| path.display().to_string())
                .collect::<Vec<_>>()
                .join(",")
        );
        return Ok(());
    }

    // Keep the original user environment independent of inferred defaults.
    let user_environment = RuntimeConfigSnapshot::capture_current();

    // Select the requested device before model/cache resolution. Explicit
    // backend requests must fail closed instead of doing model work and then
    // silently running on CPU.
    let mut device = super::run::select_device(&backend)?;
    let mut gpu_selection =
        crate::gpu_devices::resolve_cuda_gpu_devices(gpu_devices.as_deref(), &device)?;
    if let Some(selection) = &gpu_selection {
        device = selection.primary_device();
        println!(
            "{} {} ({})",
            "CUDA GPUs:".dimmed(),
            selection.selected_csv(),
            selection.selected_distributed_strategy
        );
    }
    let backend_initialized_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let backend_initialized_memory = serve_process_memory_observation_between(
        process_start_sample.clone(),
        backend_initialized_sample.clone(),
    );

    // Print banner
    print_banner();

    // `run` and `serve` share one source decision. This covers exact GGUF
    // aliases/files, local model directories, HF cache hits, and resumable HF
    // download without an entrypoint-specific fallback chain.
    let cache_dir = crate::source_resolver::hf_cache_dir(&config);
    let resolved = crate::source_resolver::resolve_model_source_with_product_sources(
        &model_name,
        &cache_dir,
        crate::source_resolver::DownloadPolicy::AutoDownload,
        None,
        &product_sources,
    )
    .await?;
    let product_input = resolved.into_product_engine_input();
    let requested_model = product_input.requested_model.clone();
    let model_id = product_input.public_model_id.clone();
    let source = product_input.source;
    let mut product_engine_config = product_input.engine_config;
    product_engine_config.numerical_execution =
        config.resolve_numerical_execution(numerical_profile.as_ref());
    let config_runtime_entries = config.runtime.runtime_config_entries();
    let configured_runtime_preset = runtime_preset
        .as_deref()
        .map(|preset| (preset, RuntimeConfigSource::Cli))
        .or_else(|| {
            config
                .runtime
                .preset
                .as_deref()
                .map(|preset| (preset, RuntimeConfigSource::ConfigFile))
        });
    let selected_runtime_preset_name =
        configured_runtime_preset.map(|(preset, _source)| preset.to_string());
    let preset_runtime_entries = match configured_runtime_preset {
        Some((preset, source)) => runtime_preset_entries(preset, source)?,
        None => Vec::new(),
    };
    let mut non_env_runtime_entries = preset_runtime_entries;
    non_env_runtime_entries.extend(config_runtime_entries);
    let mut non_env_runtime_entries =
        RuntimeConfigSnapshot::from_entries(non_env_runtime_entries).entries;
    let kv_runtime_snapshot = merge_runtime_config_sources(
        non_env_runtime_entries.clone(),
        user_environment.clone(),
        Vec::new(),
    );
    let effective_kv_dtype = resolve_effective_kv_dtype(
        kv_dtype.as_deref(),
        runtime_snapshot_value(&kv_runtime_snapshot, "FERRUM_KV_DTYPE"),
        config.runtime.kv_dtype.as_deref(),
    );
    super::run::apply_kv_dtype_override(&mut product_engine_config, effective_kv_dtype)?;
    let model_sources = product_input.model_sources;
    let defined_model = crate::source_resolver::define_registered_product_model(
        model_sources.as_ref(),
        &product_engine_config.numerical_execution,
        product_engine_config.kv_cache.dtype,
    )?;
    let vnext_plan_owns_context_capacity = defined_model.is_some();
    let model_chat_template = match defined_model.as_deref() {
        Some(prepared) => Some(crate::source_resolver::load_defined_product_chat_template(
            prepared,
        )?),
        None => match model_sources.as_deref() {
            Some(sources) => crate::source_resolver::load_product_chat_template(sources),
            None => crate::source_resolver::load_model_chat_template(&source.local_path),
        },
    };
    let product_source_identity = crate::source_resolver::product_source_identity(
        defined_model.as_deref(),
        model_sources.as_deref(),
        &requested_model,
        &model_id,
        model_chat_template.as_ref(),
    )?;
    let requested_public_model_name = matches!(
        product_engine_config.model.source.as_ref(),
        Some(ferrum_types::ModelSource::HuggingFace { .. })
    )
    .then_some(model_name.as_str());
    let served_model_names =
        effective_served_model_names(&model_id, requested_public_model_name, served_model_name)?;
    let primary_served_model_name = served_model_names
        .first()
        .expect("effective served model names are non-empty")
        .clone();
    let gguf_path = (source.format == ModelFormat::GGUF).then(|| source.local_path.clone());
    println!("{} {}", "Model:".dimmed(), model_id.cyan());
    println!("{} {}", "Path:".dimmed(), source.local_path.display());

    let startup_lora_adapters = if lora_specs.is_empty() {
        Vec::new()
    } else {
        ferrum_models::load_startup_lora_adapters(
            &primary_served_model_name,
            Some(&lora_model_id_template),
            &lora_specs,
        )?
    };
    for adapter in &startup_lora_adapters {
        println!(
            "{} {} -> {} ({})",
            "LoRA:".dimmed(),
            adapter.name.cyan(),
            adapter.public_model_id.cyan(),
            adapter.path.display()
        );
    }

    let host = host.unwrap_or_else(|| config.server.host.clone());
    let port = port.unwrap_or(config.server.port);
    let engine_model_path = source.local_path.to_string_lossy().to_string();

    // Speculative decoding draft model: resolve the draft path and pass it
    // through EngineConfig backend options. Validates that the draft model is
    // actually cached before the target load kicks in.
    let mut engine_spec_draft_path = None;
    if let Some(ref draft_name) = spec_draft {
        if gguf_path.is_some() {
            return Err(ferrum_types::FerrumError::unsupported(
                "Speculative decoding is not yet wired through the GGUF path",
            ));
        }
        let draft_id = crate::source_resolver::resolve_model_alias(draft_name);
        println!("{} {}", "Draft model:".dimmed(), draft_id.cyan());
        let cache_dir = crate::source_resolver::hf_cache_dir(&config);
        let draft_source = crate::source_resolver::find_cached_model(&cache_dir, &draft_id)
            .ok_or_else(|| {
                eprintln!(
                    "{} Draft model '{}' not in HF cache. Run: ferrum pull {}",
                    "Error:".red().bold(),
                    draft_id,
                    draft_name
                );
                ferrum_types::FerrumError::model("Draft model not found")
            })?;
        engine_spec_draft_path = Some(draft_source.local_path.to_string_lossy().to_string());
        println!(
            "{} {} tokens / verify pass",
            "Speculative decoding:".dimmed(),
            spec_tokens
        );
    }

    println!("{} {:?}", "Device:".dimmed(), device);
    let current_runtime = merge_runtime_config_sources(
        non_env_runtime_entries.clone(),
        user_environment.clone(),
        Vec::new(),
    );
    let serve_profile_entries = if vnext_plan_owns_context_capacity {
        Vec::new()
    } else {
        crate::source_resolver::serve_profile_runtime_entries(
            &source.local_path,
            &device,
            false,
            &current_runtime,
            RuntimeConfigSource::Default,
        )
    };
    if !serve_profile_entries.is_empty() {
        non_env_runtime_entries.extend(serve_profile_entries.clone());
        non_env_runtime_entries =
            RuntimeConfigSnapshot::from_entries(non_env_runtime_entries).entries;
    }
    let metal_moe_entries = if vnext_plan_owns_context_capacity {
        Vec::new()
    } else {
        crate::source_resolver::metal_gguf_moe_correctness_entries(
            &source.local_path,
            &device,
            &current_runtime,
            RuntimeConfigSource::Default,
        )
    };
    if !metal_moe_entries.is_empty() {
        non_env_runtime_entries.extend(metal_moe_entries.clone());
        non_env_runtime_entries =
            RuntimeConfigSnapshot::from_entries(non_env_runtime_entries).entries;
    }

    // Detect architecture to choose engine type. For GGUF we skip
    // ConfigManager::load_from_path (which expects HF safetensors layout)
    // and route directly to the continuous-batching LLM engine — the
    // engine's LlmExecutorFactory uses WeightFormat::detect() to route GGUF.
    println!();
    let model_definition: Option<ferrum_models::ModelDefinition> = if defined_model.is_some() {
        None
    } else if let Some(sources) = model_sources.as_deref() {
        let mut config_manager = ferrum_models::ConfigManager::new();
        Some(config_manager.load_from_bytes(sources.config_json())?)
    } else if gguf_path.is_some() {
        None
    } else {
        let mut config_manager = ferrum_models::ConfigManager::new();
        Some(config_manager.load_from_path(&source.local_path).await?)
    };
    let arch_for_dispatch = model_definition
        .as_ref()
        .map(|model_def| model_def.architecture);
    // Materialize the multi-GPU layer-split plan. The safetensors path
    // gets the layer count from ModelDefinition; the GGUF path reads it
    // from the file header — without this the placeholder plan
    // (`layers=auto`) reaches the engine and is rejected.
    let model_layer_count = if let Some(prepared) = defined_model.as_ref() {
        Some(prepared.descriptor().layer_count())
    } else if let Some(definition) = model_definition.as_ref() {
        Some(definition.num_hidden_layers)
    } else if let (Some(selection), Some(p)) = (gpu_selection.as_ref(), gguf_path.as_ref()) {
        if selection.selected_layer_split_plan.is_some() {
            Some(ferrum_models::gguf_config::gguf_num_layers(p)?)
        } else {
            None
        }
    } else {
        None
    };
    if let (Some(selection), Some(layer_count)) = (gpu_selection.as_mut(), model_layer_count) {
        if selection.apply_model_layer_count(layer_count)? {
            if let Some(plan) = selection.selected_layer_split_plan.as_deref() {
                println!("{}", format!("CUDA layer split plan: {plan}").dimmed());
            }
        }
    }
    let mut selected_runtime_preset_name = selected_runtime_preset_name;
    if selected_runtime_preset_name.is_none() {
        let inferred_preset = infer_runtime_preset_for_startup(
            arch_for_dispatch,
            model_definition.as_ref(),
            gpu_selection.as_ref(),
        );
        if let Some(preset) = inferred_preset {
            selected_runtime_preset_name = Some(preset.to_string());
            let mut inferred_entries =
                runtime_preset_entries(preset, RuntimeConfigSource::Default)?;
            inferred_entries.extend(non_env_runtime_entries);
            non_env_runtime_entries = RuntimeConfigSnapshot::from_entries(inferred_entries).entries;
        }
    }

    // Preserve the historical non-preset serve default for Qwen3-MoE, but
    // route it through the typed startup snapshot instead of a hidden
    // process-wide env mutation. M3 explicit or model-inferred presets already
    // carry the same graph-clean defaults in the typed snapshot.
    if selected_runtime_preset_name.is_none()
        && arch_for_dispatch == Some(ferrum_models::Architecture::Qwen3Moe)
    {
        let current_runtime = merge_runtime_config_sources(
            non_env_runtime_entries.clone(),
            user_environment.clone(),
            Vec::new(),
        );
        let mut legacy_entries = crate::runtime_env::moe_graph_default_entries(
            &current_runtime,
            RuntimeConfigSource::Default,
        );
        legacy_entries.extend(non_env_runtime_entries);
        non_env_runtime_entries = RuntimeConfigSnapshot::from_entries(legacy_entries).entries;
    }

    let mut startup_cli_runtime_entries = serve_cli_runtime_entries(
        kv_dtype.as_deref(),
        kv_capacity,
        kv_max_blocks,
        max_model_len,
        max_num_seqs,
        max_num_batched_tokens,
        runtime_memory_budget_bytes.map(std::num::NonZeroUsize::get),
        scheduler_prefill_first_until_active,
        scheduler_prefill_step_chunk,
        scheduler_active_decode_prefill_chunk,
        greedy_argmax_cli_override(greedy_argmax, disable_greedy_argmax),
        prefix_cache_cli_override(
            enable_prefix_caching,
            no_enable_prefix_caching,
            enable_prefix_cache,
            disable_prefix_cache,
        ),
        session_cache.as_deref(),
        session_cache_max_entries,
        session_cache_max_tokens,
        profile_jsonl.as_ref(),
        scheduler_trace_jsonl.as_ref(),
        profile_commit_sha.as_deref(),
        profile_env_hash.as_deref(),
        profile_model.as_deref(),
        profile_concurrency,
        profile_runtime_flags_json.as_deref(),
        layer_split_pipeline_mode,
    );
    startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
        "FERRUM_PROFILE_DETAIL",
        profile_detail.as_str(),
        RuntimeConfigSource::Cli,
    ));
    if let Some(wait) = prefix_rendezvous_max_wait_ms {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_PREFIX_RENDEZVOUS_MAX_WAIT_MS",
            wait.to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    push_cli_runtime_entry(
        &mut startup_cli_runtime_entries,
        "FERRUM_VNEXT_DIAGNOSTIC_FAULT",
        vnext_diagnostic_fault.map(crate::commands::VNextDiagnosticFaultArg::as_runtime_value),
    );
    push_sequence_fit_policy_cli_entry(&mut startup_cli_runtime_entries, sequence_fit_policy);
    if let Some(enabled) = batched_graph_cli_override(batched_graph, disable_batched_graph) {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_BATCHED_GRAPH",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) =
        batched_graph_cli_override(reusable_execution, disable_reusable_execution)
    {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_REUSABLE_EXECUTION",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = batched_graph_cli_override(unified_graph, disable_unified_graph) {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) =
        batched_graph_cli_override(unified_graph_layers_only, disable_unified_graph_layers_only)
    {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH_LAYERS_ONLY",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = batched_graph_cli_override(
        unified_graph_lm_head_eager,
        disable_unified_graph_lm_head_eager,
    ) {
        startup_cli_runtime_entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH_LM_HEAD_EAGER",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(selection) = &gpu_selection {
        startup_cli_runtime_entries.extend(selection.runtime_config_entries());
    }
    let execution_resource_authority = if defined_model.is_some() {
        ferrum_types::ExecutionResourceAuthority::PlanRuntime
    } else {
        ferrum_types::ExecutionResourceAuthority::LegacyEngine
    };
    let mut runtime_config = merge_runtime_config_sources(
        non_env_runtime_entries,
        user_environment,
        startup_cli_runtime_entries,
    );
    if execution_resource_authority == ferrum_types::ExecutionResourceAuthority::LegacyEngine {
        let entries = crate::gpu_mem_autosize::auto_size_runtime_entries(
            &source.local_path,
            gpu_memory_utilization,
            crate::gpu_mem_autosize::AutoSizeProfile::Server,
            &runtime_config,
        );
        for entry in entries {
            runtime_config.upsert_entry(entry);
        }
    }
    let startup_memory_request = crate::startup::memory_request(
        &device,
        execution_resource_authority,
        gpu_memory_utilization,
        &runtime_config,
    )?;
    let hardware = crate::startup::hardware_for_request(&device, startup_memory_request.as_ref());
    let typed_model_capabilities = defined_model
        .as_ref()
        .map(|defined| {
            defined.model_capabilities(
                &product_engine_config.numerical_execution,
                ferrum_types::KvStorageFormat::try_from(product_engine_config.kv_cache.dtype)
                    .map_err(ferrum_types::FerrumError::config)?,
            )
        })
        .transpose()?;
    let mut startup_auto_config = startup_auto_config(
        hardware,
        typed_model_capabilities,
        execution_resource_authority,
        arch_for_dispatch,
        model_definition.as_ref(),
        model_weight_bytes_from_path(&source.local_path),
        selected_runtime_preset_name.as_deref(),
        runtime_config,
    )?;
    crate::runtime_env::materialize_runtime_env_effective(&startup_auto_config.runtime_config);
    let lora_server_models: Vec<ferrum_server::LoraAdapterModel> = startup_lora_adapters
        .iter()
        .map(|adapter| {
            ferrum_server::LoraAdapterModel::new(
                adapter.name.clone(),
                adapter.public_model_id.clone(),
                adapter.path.display().to_string(),
            )
        })
        .collect();
    let served_model_kind = match arch_for_dispatch {
        Some(ferrum_models::Architecture::Clip) => ServedModelKind::Embedding,
        Some(ferrum_models::Architecture::Whisper) => ServedModelKind::Transcription,
        Some(ferrum_models::Architecture::Qwen3TTS) => ServedModelKind::Speech,
        _ => ServedModelKind::Llm,
    };
    validate_served_model_kv_dtype(served_model_kind, product_engine_config.kv_cache.dtype)?;
    if vnext_checkpoint.teacher_token_file.is_some() {
        return Err(FerrumError::unsupported(
            "vNext checkpoint teacher forcing is supported only by one-shot ferrum run",
        ));
    }
    let vnext_checkpoint_capture = vnext_checkpoint.to_config()?;
    if vnext_checkpoint_capture.is_some() && served_model_kind != ServedModelKind::Llm {
        return Err(FerrumError::unsupported(
            "vNext checkpoint capture is only supported for causal language models",
        ));
    }
    let served_model_registry = ServedModelRegistry::try_new(
        model_id.clone(),
        served_model_kind,
        served_model_names,
        lora_server_models,
    )
    .map_err(|error| FerrumError::config(error.to_string()))?;

    let numerical_execution = product_engine_config.numerical_execution.clone();
    let mut resolved_execution_metrics = None;
    let mut cache_allocated_status = None;
    let server_result: Result<_> = async {
        Ok(match arch_for_dispatch {
            Some(ferrum_models::Architecture::Clip) => {
                println!("{}", "Initializing CLIP embedding engine...".dimmed());
                let candle_device = candle_core::Device::Cpu;
                let executor = ferrum_models::ClipModelExecutor::from_path(
                    &source.local_path.to_string_lossy(),
                    candle_device,
                    candle_core::DType::F32,
                )?;
                let tokenizer = crate::commands::embed::load_tokenizer(&source.local_path)?;
                let mut engine_config = product_engine_config;
                engine_config.sampling.default_params =
                    ferrum_server::default_chat_sampling_params();
                engine_config.backend.device = device;
                if let Some(selection) = &gpu_selection {
                    selection.insert_backend_options(&mut engine_config.backend.backend_options);
                }
                let engine: Arc<dyn ferrum_engine::EmbedEngine + Send + Sync> = Arc::new(
                    ferrum_engine::embedding_engine::EmbeddingEngine::new(executor, engine_config)
                        .with_tokenizer(tokenizer),
                );
                AxumServer::from_embed(engine)
            }
            Some(ferrum_models::Architecture::Whisper) => {
                println!("{}", "Initializing Whisper ASR engine...".dimmed());
                let candle_device = to_candle_device(&device)?;
                let executor = ferrum_models::WhisperModelExecutor::from_path(
                    &source.local_path.to_string_lossy(),
                    candle_device,
                    candle_core::DType::F32,
                )?;
                let mut engine_config = product_engine_config;
                engine_config.backend.device = device;
                if let Some(selection) = &gpu_selection {
                    selection.insert_backend_options(&mut engine_config.backend.backend_options);
                }
                let engine: Arc<dyn ferrum_engine::TranscribeEngine + Send + Sync> = Arc::new(
                    ferrum_engine::transcription_engine::TranscriptionEngine::new(
                        executor,
                        engine_config,
                    ),
                );
                AxumServer::from_transcribe(engine)
            }
            Some(ferrum_models::Architecture::Qwen3TTS) => {
                let n_slots = tts_slots.max(1);
                println!(
                    "{} ({} slot{})",
                    "Initializing Qwen3-TTS engine...".dimmed(),
                    n_slots,
                    if n_slots > 1 { "s" } else { "" }
                );
                let model_path = source.local_path.to_string_lossy().to_string();
                let mut executors = Vec::with_capacity(n_slots);
                for i in 0..n_slots {
                    let candle_device = to_candle_device(&device)?;
                    let executor = ferrum_models::TtsModelExecutor::from_path(
                        &model_path,
                        candle_device,
                        candle_core::DType::F32,
                    )?;
                    if i == 0 {
                        println!("  Slot 0 loaded");
                    } else {
                        println!("  Slot {} loaded", i);
                    }
                    executors.push(executor);
                }
                let engine: Arc<dyn ferrum_engine::TtsEngine + Send + Sync> =
                    Arc::new(ferrum_engine::tts_engine::TtsService::new_multi(
                        executors,
                        ferrum_types::ModelId(model_id.clone()),
                    ));
                AxumServer::from_tts(engine)
            }
            _ => {
                println!(
                    "{}",
                    "Initializing engine (continuous batching)...".dimmed()
                );
                let mut engine_config = product_engine_config;
                engine_config.kv_cache.cache_type = serve_kv_cache_type_for_device(&device);
                engine_config.backend.device = device;
                engine_config.scheduler.policy = ferrum_types::SchedulingPolicy::ContinuousBatch;
                engine_config
                    .apply_runtime_config_snapshot(&startup_auto_config.runtime_config)
                    .map_err(ferrum_types::FerrumError::config)?;
                engine_config.runtime.vnext_checkpoint_capture = vnext_checkpoint_capture;
                engine_config.runtime.startup_memory_request = startup_memory_request;
                engine_config.backend.backend_options.insert(
                    "model_path".to_string(),
                    serde_json::Value::String(engine_model_path.clone()),
                );
                if let Some(selection) = &gpu_selection {
                    selection.insert_backend_options(&mut engine_config.backend.backend_options);
                }
                crate::layer_split_pipeline::insert_backend_option_from_runtime(
                    &startup_auto_config.runtime_config,
                    &mut engine_config.backend.backend_options,
                )?;
                if let Some(draft_path) = engine_spec_draft_path.as_ref() {
                    engine_config.backend.backend_options.insert(
                        "spec_draft".to_string(),
                        serde_json::Value::String(draft_path.clone()),
                    );
                    engine_config.backend.backend_options.insert(
                        "spec_n".to_string(),
                        serde_json::Value::Number(serde_json::Number::from(spec_tokens)),
                    );
                }
                super::run::apply_kv_dtype_override(&mut engine_config, effective_kv_dtype)?;
                let engine: Arc<dyn ferrum_engine::LlmInferenceEngine + Send + Sync> =
                    Arc::from(match (defined_model, model_sources) {
                        (Some(prepared), _) => {
                            ferrum_engine::create_defined_product_engine(engine_config, prepared)
                                .await?
                        }
                        (None, Some(sources)) => {
                            ferrum_engine::create_product_engine(engine_config, sources).await?
                        }
                        (None, None) => ferrum_engine::create_default_engine(engine_config).await?,
                    });
                crate::startup::apply_engine_plan(&mut startup_auto_config, engine.config());
                resolved_execution_metrics = engine.cache_metrics_snapshot();
                if product_memory_enabled {
                    cache_allocated_status = Some(engine.status().await);
                }
                AxumServer::from_llm(engine).with_prompt_template(model_chat_template)
            }
        })
    }
    .await;
    let server = match server_result {
        Ok(server) => server,
        Err(error) => {
            write_failed_startup_config_artifacts(
                &startup_auto_config,
                product_source_identity.as_ref(),
                &numerical_execution,
                effective_config_json.as_deref(),
                decision_trace_jsonl.as_deref(),
                &error,
            );
            return Err(error);
        }
    };
    write_startup_config_artifacts(
        &startup_auto_config,
        product_source_identity.as_ref(),
        &numerical_execution,
        effective_config_json.as_deref(),
        decision_trace_jsonl.as_deref(),
    )?;
    let native_profile_jsonl = if product_observability.unified_product_profile_enabled() {
        None
    } else {
        profile_jsonl.clone()
    };
    configure_profile_sink(
        native_profile_jsonl,
        ProfileSinkCliFields {
            commit_sha: profile_commit_sha,
            env_hash: profile_env_hash,
            model: profile_model,
            concurrency: profile_concurrency,
            runtime_flags_json: profile_runtime_flags_json,
        },
        &startup_auto_config,
        &model_id,
    )?;

    write_resolved_execution_config(
        effective_config_json.as_deref(),
        resolved_execution_metrics.as_ref(),
    )?;
    let server = server
        .with_auto_config(startup_auto_config)
        .with_default_enable_thinking(default_enable_thinking)
        .with_interleaved_system_coalescing(interleaved_system_coalescing);
    let model_loaded_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let model_loaded_memory = serve_process_memory_observation_between(
        backend_initialized_sample
            .clone()
            .or_else(|| process_start_sample.clone()),
        model_loaded_sample.clone(),
    );
    let model_loaded_duration_us = serve_start
        .elapsed()
        .as_micros()
        .try_into()
        .unwrap_or(u64::MAX);
    let profile_run_done_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let profile_run_done_memory = serve_process_memory_observation_between(
        model_loaded_sample.clone(),
        profile_run_done_sample.clone(),
    );
    let cache_allocated_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let cache_allocated_memory = serve_process_memory_observation_between(
        profile_run_done_sample
            .clone()
            .or_else(|| model_loaded_sample.clone()),
        cache_allocated_sample.clone(),
    );
    let server = server.with_served_model_registry(served_model_registry);
    crate::observability_product::write_actual_serve_startup_observability(
        &product_observability,
        model_loaded_duration_us,
        model_loaded_memory.clone(),
        actual_serve_startup_memory_stages(
            product_memory_enabled,
            process_start_memory.clone(),
            backend_initialized_memory.clone(),
            profile_run_done_memory.clone(),
            cache_allocated_memory.clone(),
            cache_allocated_status.clone(),
        ),
    )?;

    // Create server config
    let server_config = ServerConfig {
        host: host.clone(),
        port,
        request_dump_dir: request_dump_dir.clone(),
        profile_jsonl: product_observability
            .unified_product_profile_enabled()
            .then(|| profile_jsonl.clone())
            .flatten(),
        profile_detail: product_observability.profile_detail,
        memory_profile_jsonl: product_observability
            .unified_product_profile_enabled()
            .then(|| memory_profile_jsonl.clone())
            .flatten(),
        ..Default::default()
    };

    println!();
    println!(
        "{} {} {}",
        "🚀".green(),
        "Server running at".green().bold(),
        format!("http://{}:{}", host, port).cyan().bold()
    );
    println!();
    println!("Endpoints:");
    println!("  POST /v1/chat/completions      - OpenAI-compatible chat");
    println!("  POST /v1/responses             - OpenAI Responses API");
    println!("  GET  /v1/models                - List models");
    println!("  GET  /health                   - Health check");
    println!();
    println!("{}", "Press Ctrl+C to stop.".dimmed());
    println!();

    // Write PID file for stop command
    let pid_file = std::env::temp_dir().join("ferrum.pid");
    std::fs::write(&pid_file, std::process::id().to_string()).ok();

    // Keep the server future alive while stop requests graceful HTTP drain.
    let server = Arc::new(server);
    let mut server_task = {
        let server = Arc::clone(&server);
        let server_config = server_config.clone();
        tokio::spawn(async move { server.start(&server_config).await })
    };
    let shutdown_timeout = Duration::from_secs(30);
    let serve_result: Result<()> = tokio::select! {
        joined = &mut server_task => {
            let start_result = match joined {
                Ok(result) => result,
                Err(error) => Err(FerrumError::internal(format!(
                    "serve task failed: {error}"
                ))),
            };
            let stop_result = server.stop(shutdown_timeout).await;
            start_result.and(stop_result)
        }
        _ = serve_shutdown_signal() => {
            println!();
            println!("{}", "Shutting down...".yellow());
            let stop_result = server.stop(shutdown_timeout).await;
            let start_result = match tokio::time::timeout(shutdown_timeout, &mut server_task).await {
                Ok(Ok(result)) => result,
                Ok(Err(error)) => Err(FerrumError::internal(format!(
                    "serve task failed during shutdown: {error}"
                ))),
                Err(_) => {
                    server_task.abort();
                    let _ = server_task.await;
                    Err(FerrumError::internal(format!(
                        "serve task did not stop within {} ms",
                        shutdown_timeout.as_millis()
                    )))
                }
            };
            stop_result.and(start_result)
        }
    };

    // Clean up PID file
    std::fs::remove_file(&pid_file).ok();

    // PLAYBOOK § 1.5: Rust statics don't drop on exit, so the global
    // TraceWriter's buffered events would be lost. Force-flush on
    // ctrl_c-driven shutdown (matches bench / bench-serve exit paths).
    ferrum_bench_core::trace::flush_global_trace();
    ferrum_bench_core::flush_global_profile();
    let shutdown_after = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let shutdown_memory = serve_process_memory_observation_between(
        model_loaded_sample
            .clone()
            .or_else(|| backend_initialized_sample.clone())
            .or_else(|| process_start_sample.clone()),
        shutdown_after,
    );
    crate::observability_product::append_actual_serve_memory_stage_observability(
        &product_observability,
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_serve_shutdown",
            "shutdown",
            None,
            shutdown_memory,
        ),
    )?;

    serve_result?;
    Ok(())
}

async fn serve_shutdown_signal() {
    #[cfg(unix)]
    {
        match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
            Ok(mut terminate) => {
                tokio::select! {
                    _ = signal::ctrl_c() => {}
                    _ = terminate.recv() => {}
                }
            }
            Err(_) => {
                let _ = signal::ctrl_c().await;
            }
        }
    }
    #[cfg(not(unix))]
    {
        let _ = signal::ctrl_c().await;
    }
}

fn serve_process_memory_observation_between(
    before: Option<crate::memory_profile::ProcessMemorySample>,
    after: Option<crate::memory_profile::ProcessMemorySample>,
) -> Option<crate::memory_profile::ProcessMemoryObservation> {
    after.map(|after| crate::memory_profile::ProcessMemoryObservation::from_samples(before, after))
}

fn actual_serve_startup_memory_stages(
    enabled: bool,
    process_start_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    backend_initialized_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    profile_run_done_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    cache_allocated_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    cache_allocated_status: Option<ferrum_types::EngineStatus>,
) -> Vec<crate::observability_product::ActualMemoryStageObservation> {
    if !enabled {
        return Vec::new();
    }
    let profile_run_done = crate::observability_product::ActualMemoryStageObservation::new(
        "actual_serve_profile_run_done",
        "profile_run_done",
        None,
        profile_run_done_memory,
    )
    .with_profile_run_status(
        false,
        "not_configured",
        "product_basic_profile_does_not_execute_extra_warmup",
    );
    let mut cache_allocated = crate::observability_product::ActualMemoryStageObservation::new(
        "actual_serve_cache_allocated",
        "cache_allocated",
        None,
        cache_allocated_memory,
    );
    if let Some(status) = cache_allocated_status.as_ref() {
        cache_allocated = cache_allocated.with_engine_cache_status(status);
    }
    vec![
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_serve_process_start",
            "process_start",
            None,
            process_start_memory,
        ),
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_serve_backend_initialized",
            "backend_initialized",
            None,
            backend_initialized_memory,
        ),
        profile_run_done,
        cache_allocated,
    ]
}

fn print_banner() {
    println!();
    println!("{}", "  ______                            ".bright_red());
    println!("{}", " |  ____|                           ".bright_red());
    println!("{}", " | |__ ___ _ __ _ __ _   _ _ __ ___  ".bright_red());
    println!("{}", " |  __/ _ \\ '__| '__| | | | '_ ` _ \\ ".bright_red());
    println!("{}", " | | |  __/ |  | |  | |_| | | | | | ".bright_red());
    println!("{}", " |_|  \\___|_|  |_|   \\__,_|_| |_| |_|".bright_red());
    println!();
    println!("   {}", "🦀 Rust LLM Inference Server".bright_cyan().bold());
    println!(
        "   {}",
        format!("Version {}", env!("CARGO_PKG_VERSION")).dimmed()
    );
    println!();
}

fn parse_lora_specs(values: &[String]) -> Result<Vec<ferrum_models::StartupLoraSpec>> {
    let mut specs = Vec::with_capacity(values.len());
    for value in values {
        let (name, path) = value.split_once('=').ok_or_else(|| {
            ferrum_types::FerrumError::config(format!(
                "invalid --lora value {value:?}; expected NAME=PATH"
            ))
        })?;
        if name.is_empty() || path.is_empty() {
            return Err(ferrum_types::FerrumError::config(format!(
                "invalid --lora value {value:?}; expected non-empty NAME=PATH"
            )));
        }
        specs.push(ferrum_models::StartupLoraSpec {
            name: name.to_string(),
            path: PathBuf::from(shellexpand::tilde(path).to_string()),
        });
    }
    Ok(specs)
}

fn startup_auto_config(
    hardware: HardwareCapabilities,
    typed_model_capabilities: Option<ModelCapabilities>,
    execution_resource_authority: ferrum_types::ExecutionResourceAuthority,
    architecture: Option<ferrum_models::Architecture>,
    model_definition: Option<&ferrum_models::ModelDefinition>,
    model_weight_bytes: Option<u64>,
    runtime_preset: Option<&str>,
    runtime_config: RuntimeConfigSnapshot,
) -> Result<ResolvedFerrumConfig> {
    let model = typed_model_capabilities
        .or_else(|| {
            model_definition.map(|definition| {
                model_capabilities_from_definition_with_weight_bytes_for_hardware(
                    definition,
                    model_weight_bytes,
                    &hardware,
                )
            })
        })
        .unwrap_or_else(ModelCapabilities::unknown);
    let workload = match runtime_preset {
        Some(M3_QWEN3_30B_A3B_INT4_PRESET) => WorkloadProfile::m3_qwen3_30b_a3b_int4(),
        Some(QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET) => {
            WorkloadProfile::qwen25_72b_gptq_int4_2x4090_layer_split()
        }
        Some(other) => {
            return Err(ferrum_types::FerrumError::config(format!(
                "unknown runtime preset: {other}"
            )));
        }
        None => match infer_runtime_preset_for_startup(architecture, model_definition, None) {
            Some(M3_QWEN3_30B_A3B_INT4_PRESET) => WorkloadProfile::m3_qwen3_30b_a3b_int4(),
            _ => WorkloadProfile::serving_default_for_hardware(&hardware),
        },
    };

    crate::startup::resolve_config(
        runtime_config,
        model,
        hardware,
        workload,
        execution_resource_authority,
    )
}

pub(crate) fn merge_runtime_config_sources(
    config_file_entries: Vec<RuntimeConfigEntry>,
    env_snapshot: RuntimeConfigSnapshot,
    cli_entries: Vec<RuntimeConfigEntry>,
) -> RuntimeConfigSnapshot {
    let mut runtime_config = RuntimeConfigSnapshot::from_entries(config_file_entries);
    for entry in env_snapshot.entries {
        runtime_config.upsert_entry(entry);
    }
    for entry in cli_entries {
        runtime_config.upsert_entry(entry);
    }
    runtime_config
}

pub(crate) fn runtime_preset_entries(
    preset: &str,
    source: RuntimeConfigSource,
) -> Result<Vec<RuntimeConfigEntry>> {
    let pairs: &[(&str, &str)] = match preset {
        M3_QWEN3_30B_A3B_INT4_PRESET => &[
            ("FERRUM_BACKEND", "cuda"),
            ("FERRUM_MOE_DEVICE_ROUTE", "1"),
            ("FERRUM_MOE_STREAMS", "4"),
            ("FERRUM_GREEDY_ARGMAX", "1"),
            ("FERRUM_KV_MAX_BLOCKS", "2048"),
            ("FERRUM_PAGED_MAX_SEQS", "32"),
            ("FERRUM_KV_CAPACITY", "512"),
            ("FERRUM_MOE_GRAPH", "0"),
            ("FERRUM_VLLM_MOE", "1"),
            ("FERRUM_VLLM_MOE_PAIR_IDS", "1"),
            ("FERRUM_ATTENTION_POLICY", "native-adaptive"),
            ("FERRUM_PREFIX_CACHE", "0"),
        ],
        QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET => &[
            ("FERRUM_BACKEND", "cuda"),
            ("FERRUM_LAYER_SPLIT_PIPELINE_MODE", "batch"),
            ("FERRUM_MAX_MODEL_LEN", "4096"),
            ("FERRUM_KV_MAX_BLOCKS", "1024"),
            ("FERRUM_KV_CAPACITY", "1024"),
            ("FERRUM_PAGED_MAX_SEQS", "16"),
            ("FERRUM_MAX_BATCHED_TOKENS", "1536"),
            ("FERRUM_SCHED_PREFILL_FIRST_UNTIL_ACTIVE", "16"),
        ],
        other => {
            return Err(ferrum_types::FerrumError::config(format!(
                "unknown runtime preset: {other}"
            )));
        }
    };
    Ok(pairs
        .iter()
        .map(|(key, value)| RuntimeConfigEntry::new(*key, *value, source))
        .collect())
}

fn serve_cli_runtime_entries(
    kv_dtype: Option<&str>,
    kv_capacity: Option<usize>,
    kv_max_blocks: Option<usize>,
    max_model_len: Option<usize>,
    max_num_seqs: Option<usize>,
    max_num_batched_tokens: Option<usize>,
    runtime_memory_budget_bytes: Option<usize>,
    scheduler_prefill_first_until_active: Option<usize>,
    scheduler_prefill_step_chunk: Option<usize>,
    scheduler_active_decode_prefill_chunk: Option<usize>,
    greedy_argmax: Option<bool>,
    prefix_cache: Option<bool>,
    session_cache: Option<&str>,
    session_cache_max_entries: Option<usize>,
    session_cache_max_tokens: Option<usize>,
    profile_jsonl: Option<&PathBuf>,
    scheduler_trace_jsonl: Option<&PathBuf>,
    profile_commit_sha: Option<&str>,
    profile_env_hash: Option<&str>,
    profile_model: Option<&str>,
    profile_concurrency: Option<u32>,
    profile_runtime_flags_json: Option<&str>,
    layer_split_pipeline_mode: Option<crate::layer_split_pipeline::LayerSplitPipelineModeArg>,
) -> Vec<RuntimeConfigEntry> {
    let mut entries = Vec::new();
    push_cli_runtime_entry(&mut entries, "FERRUM_KV_DTYPE", kv_dtype);
    push_cli_runtime_usize(&mut entries, "FERRUM_KV_CAPACITY", kv_capacity);
    push_cli_runtime_usize(&mut entries, "FERRUM_KV_MAX_BLOCKS", kv_max_blocks);
    push_cli_runtime_usize(&mut entries, "FERRUM_MAX_MODEL_LEN", max_model_len);
    push_cli_runtime_usize(&mut entries, "FERRUM_PAGED_MAX_SEQS", max_num_seqs);
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_MAX_BATCHED_TOKENS",
        max_num_batched_tokens,
    );
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_RUNTIME_MEMORY_BUDGET_BYTES",
        runtime_memory_budget_bytes,
    );
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_SCHED_PREFILL_FIRST_UNTIL_ACTIVE",
        scheduler_prefill_first_until_active,
    );
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_SCHED_PREFILL_STEP_CHUNK",
        scheduler_prefill_step_chunk,
    );
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_ACTIVE_DECODE_PREFILL_CHUNK",
        scheduler_active_decode_prefill_chunk,
    );
    if let Some(enabled) = greedy_argmax {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_GREEDY_ARGMAX",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = prefix_cache {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PREFIX_CACHE_REQUESTED",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PREFIX_CACHE_PRODUCT",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PREFIX_CACHE",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    push_cli_runtime_entry(&mut entries, "FERRUM_SESSION_CACHE", session_cache);
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_SESSION_CACHE_MAX_ENTRIES",
        session_cache_max_entries,
    );
    push_cli_runtime_usize(
        &mut entries,
        "FERRUM_SESSION_CACHE_MAX_TOKENS",
        session_cache_max_tokens,
    );
    if let Some(path) = profile_jsonl {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PROFILE_JSONL",
            path.to_string_lossy().to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(path) = scheduler_trace_jsonl {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_SCHEDULER_TRACE_JSONL",
            path.to_string_lossy().to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    if profile_jsonl.is_some() || scheduler_trace_jsonl.is_some() {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PROFILE_ENTRYPOINT",
            "serve",
            RuntimeConfigSource::Cli,
        ));
    }
    push_cli_runtime_entry(
        &mut entries,
        "FERRUM_PROFILE_COMMIT_SHA",
        profile_commit_sha,
    );
    push_cli_runtime_entry(&mut entries, "FERRUM_PROFILE_ENV_HASH", profile_env_hash);
    push_cli_runtime_entry(&mut entries, "FERRUM_PROFILE_MODEL", profile_model);
    if let Some(concurrency) = profile_concurrency {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PROFILE_CONCURRENCY",
            concurrency.to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    push_cli_runtime_entry(
        &mut entries,
        "FERRUM_PROFILE_RUNTIME_FLAGS_JSON",
        profile_runtime_flags_json,
    );
    crate::layer_split_pipeline::push_cli_runtime_entry(&mut entries, layer_split_pipeline_mode);
    entries
}

fn prefix_cache_cli_override(
    enable_vllm: bool,
    disable_vllm: bool,
    enable_product: bool,
    disable_product: bool,
) -> Option<bool> {
    if enable_vllm || enable_product {
        Some(true)
    } else if disable_vllm || disable_product {
        Some(false)
    } else {
        None
    }
}

fn greedy_argmax_cli_override(enable: bool, disable: bool) -> Option<bool> {
    if enable {
        Some(true)
    } else if disable {
        Some(false)
    } else {
        None
    }
}

fn batched_graph_cli_override(enable: bool, disable: bool) -> Option<bool> {
    if enable {
        Some(true)
    } else if disable {
        Some(false)
    } else {
        None
    }
}

fn resolve_effective_kv_dtype<'a>(
    cli_arg: Option<&'a str>,
    env_value: Option<&'a str>,
    config_file_value: Option<&'a str>,
) -> Option<&'a str> {
    cli_arg.or(env_value).or(config_file_value)
}

fn validate_served_model_kv_dtype(
    kind: ServedModelKind,
    dtype: ferrum_types::KvCacheDtype,
) -> Result<()> {
    if kind != ServedModelKind::Llm && dtype != ferrum_types::KvCacheDtype::Fp16 {
        return Err(FerrumError::unsupported(format!(
            "KV storage overrides require a causal language executor; {kind:?} serving does not support KV dtype {}",
            dtype.as_str()
        )));
    }
    Ok(())
}

fn push_cli_runtime_entry(entries: &mut Vec<RuntimeConfigEntry>, key: &str, value: Option<&str>) {
    if let Some(value) = value.filter(|value| !value.trim().is_empty()) {
        entries.push(RuntimeConfigEntry::new(
            key,
            value.to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
}

fn push_sequence_fit_policy_cli_entry(
    entries: &mut Vec<RuntimeConfigEntry>,
    policy: Option<crate::commands::SequenceFitPolicyArg>,
) {
    push_cli_runtime_entry(
        entries,
        "FERRUM_SEQUENCE_FIT_POLICY",
        policy.map(crate::commands::SequenceFitPolicyArg::as_runtime_value),
    );
}

fn push_cli_runtime_usize(entries: &mut Vec<RuntimeConfigEntry>, key: &str, value: Option<usize>) {
    if let Some(value) = value {
        entries.push(RuntimeConfigEntry::new(
            key,
            value.to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
}

fn serve_kv_cache_type_for_device(device: &ferrum_types::Device) -> ferrum_types::KvCacheType {
    match device {
        ferrum_types::Device::CPU => ferrum_types::KvCacheType::Contiguous,
        _ => ferrum_types::KvCacheType::Paged,
    }
}

fn effective_served_model_names(
    default_model_id: &str,
    requested_model: Option<&str>,
    requested_names: Vec<String>,
) -> Result<Vec<String>> {
    let names = if requested_names.is_empty() {
        let mut names = Vec::with_capacity(2);
        if let Some(requested_model) = requested_model {
            names.push(requested_model.to_string());
        }
        if names.first().map(String::as_str) != Some(default_model_id) {
            names.push(default_model_id.to_string());
        }
        names
    } else {
        requested_names
    };
    let mut seen = HashSet::with_capacity(names.len());
    for name in &names {
        if name.is_empty() || name.trim() != name {
            return Err(FerrumError::config(
                "--served-model-name values must be non-empty and have no surrounding whitespace",
            ));
        }
        if !seen.insert(name.clone()) {
            return Err(FerrumError::config(format!(
                "duplicate --served-model-name value: {name}"
            )));
        }
    }
    Ok(names)
}

pub(crate) fn write_startup_config_artifacts(
    auto_config: &ResolvedFerrumConfig,
    resolution_evidence: Option<&ferrum_interfaces::vnext::ProductModelSourceIdentity>,
    numerical_execution: &ferrum_types::NumericalExecutionPolicy,
    effective_config_json: Option<&std::path::Path>,
    decision_trace_jsonl: Option<&std::path::Path>,
) -> Result<()> {
    write_startup_config_artifacts_with_failure(
        auto_config,
        resolution_evidence,
        numerical_execution,
        effective_config_json,
        decision_trace_jsonl,
        None,
    )
}

/// Preserve resolved sources and requested limits when initialization fails.
/// These diagnostics do not certify that the preliminary memory limits fit.
pub(crate) fn write_failed_startup_config_artifacts(
    auto_config: &ResolvedFerrumConfig,
    resolution_evidence: Option<&ferrum_interfaces::vnext::ProductModelSourceIdentity>,
    numerical_execution: &ferrum_types::NumericalExecutionPolicy,
    effective_config_json: Option<&std::path::Path>,
    decision_trace_jsonl: Option<&std::path::Path>,
    error: &FerrumError,
) {
    let failure = serde_json::json!({
        "status": "failed",
        "phase": "engine_initialization",
        "configuration": "preliminary",
        "error": error.to_string(),
    });
    if let Err(artifact_error) = write_startup_config_artifacts_with_failure(
        auto_config,
        resolution_evidence,
        numerical_execution,
        effective_config_json,
        decision_trace_jsonl,
        Some(&failure),
    ) {
        // A diagnostic write failure must not replace the initialization error.
        eprintln!("Could not write startup failure diagnostics: {artifact_error}");
    }
}

fn write_startup_config_artifacts_with_failure(
    auto_config: &ResolvedFerrumConfig,
    resolution_evidence: Option<&ferrum_interfaces::vnext::ProductModelSourceIdentity>,
    numerical_execution: &ferrum_types::NumericalExecutionPolicy,
    effective_config_json: Option<&std::path::Path>,
    decision_trace_jsonl: Option<&std::path::Path>,
    failure: Option<&serde_json::Value>,
) -> Result<()> {
    if let Some(path) = effective_config_json {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
        }
        let mut document = auto_config.effective_config_document();
        document["numerical_execution"] = serde_json::json!({ "requested": numerical_execution });
        if let Some(failure) = failure {
            document["startup"] = failure.clone();
        }
        if let Some(evidence) = resolution_evidence {
            let object = document.as_object_mut().ok_or_else(|| {
                ferrum_types::FerrumError::serialization(
                    "effective startup config document must be an object",
                )
            })?;
            object.insert(
                "resolution_evidence".to_owned(),
                serde_json::to_value(evidence)
                    .map_err(|err| ferrum_types::FerrumError::serialization(err.to_string()))?,
            );
        }
        let bytes = serde_json::to_vec_pretty(&document)
            .map_err(|err| ferrum_types::FerrumError::serialization(err.to_string()))?;
        std::fs::write(path, [bytes.as_slice(), b"\n"].concat())
            .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
    }
    if let Some(path) = decision_trace_jsonl {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
        }
        let trace = match failure {
            Some(failure) => {
                let mut trace = String::new();
                for decision in &auto_config.decisions {
                    let mut record = serde_json::to_value(decision)
                        .map_err(|err| FerrumError::serialization(err.to_string()))?;
                    record["startup"] = failure.clone();
                    trace.push_str(
                        &serde_json::to_string(&record)
                            .map_err(|err| FerrumError::serialization(err.to_string()))?,
                    );
                    trace.push('\n');
                }
                trace
            }
            None => auto_config
                .decision_trace_jsonl()
                .map_err(|err| ferrum_types::FerrumError::serialization(err.to_string()))?,
        };
        std::fs::write(path, trace)
            .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
    }
    Ok(())
}

/// Replace requested-only startup evidence with the selected plan after the
/// executor has compiled and initialized successfully. Both run and serve use
/// this path; legacy executors without numerical evidence leave it unchanged.
pub(crate) fn write_resolved_execution_config(
    path: Option<&std::path::Path>,
    executor_snapshot: Option<&serde_json::Value>,
) -> Result<()> {
    let (Some(path), Some(snapshot)) = (path, executor_snapshot) else {
        return Ok(());
    };
    let Some(numerical) = snapshot.get("numerical_execution") else {
        return Ok(());
    };
    let bytes =
        std::fs::read(path).map_err(|error| ferrum_types::FerrumError::io(error.to_string()))?;
    let mut document: serde_json::Value = serde_json::from_slice(&bytes)
        .map_err(|error| ferrum_types::FerrumError::serialization(error.to_string()))?;
    let object = document.as_object_mut().ok_or_else(|| {
        ferrum_types::FerrumError::serialization("effective startup config must be an object")
    })?;
    object.insert("numerical_execution".into(), numerical.clone());
    for field in ["kv_storage", "attention_execution_policy"] {
        if let Some(value) = snapshot.get(field) {
            object.insert(field.into(), value.clone());
        }
    }
    let bytes = serde_json::to_vec_pretty(&document)
        .map_err(|error| ferrum_types::FerrumError::serialization(error.to_string()))?;
    std::fs::write(path, [bytes.as_slice(), b"\n"].concat())
        .map_err(|error| ferrum_types::FerrumError::io(error.to_string()))
}

struct ProfileSinkCliFields {
    commit_sha: Option<String>,
    env_hash: Option<String>,
    model: Option<String>,
    concurrency: Option<u32>,
    runtime_flags_json: Option<String>,
}

fn configure_profile_sink(
    profile_jsonl: Option<PathBuf>,
    fields: ProfileSinkCliFields,
    auto_config: &ResolvedFerrumConfig,
    model_id: &str,
) -> Result<()> {
    let Some(path) = profile_jsonl else {
        return Ok(());
    };

    let runtime_flags = match fields.runtime_flags_json {
        Some(json) => {
            let value = serde_json::from_str::<serde_json::Value>(&json).map_err(|err| {
                ferrum_types::FerrumError::config(format!(
                    "invalid --profile-runtime-flags-json: {err}"
                ))
            })?;
            if !value.is_object() {
                return Err(ferrum_types::FerrumError::config(
                    "--profile-runtime-flags-json must be a JSON object",
                ));
            }
            value
        }
        None => auto_config.effective_config_document(),
    };

    let env_hash = match fields.env_hash {
        Some(value) if value.starts_with("sha256:") => value,
        Some(_) => {
            return Err(ferrum_types::FerrumError::config(
                "--profile-env-hash must start with sha256:",
            ))
        }
        None => auto_config.runtime_env_hash(),
    };

    let metadata = ProfileMetadata {
        commit_sha: fields.commit_sha.filter(|value| !value.trim().is_empty()),
        env_hash,
        model: fields
            .model
            .filter(|value| !value.trim().is_empty())
            .unwrap_or_else(|| model_id.to_string()),
        concurrency: fields
            .concurrency
            .filter(|value| *value > 0)
            .unwrap_or_else(|| auto_config.workload_profile.target_concurrency.max(1) as u32),
        runtime_flags,
    };
    let profile_config = ProfileSinkConfig::enabled(path, metadata);
    ferrum_bench_core::configure_global_profile(profile_config.clone())
        .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
    ferrum_kernels::configure_native_profile_sink(&profile_config)
        .map_err(|err| ferrum_types::FerrumError::io(err.to_string()))?;
    Ok(())
}

#[cfg(test)]
pub(crate) fn model_capabilities_from_definition_with_weight_bytes(
    definition: &ferrum_models::ModelDefinition,
    model_weight_bytes: Option<u64>,
) -> ModelCapabilities {
    ferrum_models::legacy_capabilities::from_definition_with_weight_bytes(
        definition,
        model_weight_bytes,
    )
}

pub(crate) fn model_capabilities_from_definition_with_weight_bytes_for_hardware(
    definition: &ferrum_models::ModelDefinition,
    model_weight_bytes: Option<u64>,
    hardware: &HardwareCapabilities,
) -> ModelCapabilities {
    ferrum_models::legacy_capabilities::from_definition_with_weight_bytes_for_hardware(
        definition,
        model_weight_bytes,
        hardware,
    )
}

pub(crate) fn model_weight_bytes_from_path(path: &Path) -> Option<u64> {
    if path.is_file() {
        return std::fs::metadata(path)
            .ok()
            .map(|metadata| metadata.len())
            .filter(|value| *value > 0);
    }
    if !path.is_dir() {
        return None;
    }
    let mut total = 0u64;
    for entry in std::fs::read_dir(path).ok()?.flatten() {
        let entry_path = entry.path();
        let is_weight = entry_path
            .extension()
            .and_then(|value| value.to_str())
            .map(|ext| ext == "safetensors" || ext == "bin")
            .unwrap_or(false);
        if !is_weight {
            continue;
        }
        if let Ok(metadata) = std::fs::metadata(&entry_path) {
            total = total.saturating_add(metadata.len());
        }
    }
    (total > 0).then_some(total)
}

pub(crate) fn hardware_capabilities_for_device(
    device: &ferrum_types::Device,
) -> HardwareCapabilities {
    let features = compiled_kernel_features();
    match device {
        ferrum_types::Device::CUDA(id) => {
            cuda_hardware_capabilities(features, probe_cuda_device(*id as usize))
        }
        ferrum_types::Device::ROCm(_) => HardwareCapabilities {
            backend: "rocm".to_string(),
            supported_dtypes: vec!["fp16".to_string(), "fp32".to_string()],
            supported_kv_dtypes: vec!["fp16".to_string()],
            compiled_features: features,
            ..HardwareCapabilities::unknown()
        },
        #[cfg(any(target_os = "macos", target_os = "ios"))]
        ferrum_types::Device::Metal => HardwareCapabilities {
            backend: "metal".to_string(),
            supported_dtypes: vec!["fp16".to_string(), "fp32".to_string()],
            // This is a startup hardware filter. The compiled vNext plan still
            // proves model/operation support for the requested storage ABI.
            supported_kv_dtypes: vec!["fp16".to_string(), "int8".to_string()],
            compiled_features: features,
            ..HardwareCapabilities::unknown()
        },
        ferrum_types::Device::CPU => HardwareCapabilities {
            backend: "cpu".to_string(),
            supported_dtypes: vec!["fp32".to_string()],
            supported_kv_dtypes: vec!["fp16".to_string()],
            compiled_features: features,
            ..HardwareCapabilities::unknown()
        },
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct CudaDeviceProbe {
    name: Option<String>,
    cuda_runtime: Option<String>,
    compute_capability: Option<String>,
    vram_bytes: Option<u64>,
    sm_count: Option<u32>,
}

fn cuda_hardware_capabilities(
    features: CompiledKernelFeatures,
    probe: CudaDeviceProbe,
) -> HardwareCapabilities {
    HardwareCapabilities {
        backend: "cuda".to_string(),
        cuda_runtime: probe.cuda_runtime,
        compute_capability: probe.compute_capability,
        vram_bytes: probe.vram_bytes,
        sm_count: probe.sm_count,
        supported_dtypes: vec!["fp16".to_string(), "fp32".to_string()],
        supported_kv_dtypes: vec![
            "fp16".to_string(),
            "bf16".to_string(),
            "int8".to_string(),
            "fp8".to_string(),
        ],
        graph_support: cfg!(feature = "cuda"),
        compiled_features: features,
    }
}

fn probe_cuda_device(device_id: usize) -> CudaDeviceProbe {
    let mut probe = run_nvidia_smi_query(device_id, "name,compute_cap,memory.total")
        .and_then(|output| parse_nvidia_smi_gpu_query(&output))
        .unwrap_or_default();
    probe.cuda_runtime = probe_cuda_runtime_version();
    probe.sm_count = run_nvidia_smi_query(device_id, "multiprocessor_count")
        .and_then(|output| parse_first_u32(&output))
        .or_else(|| probe.name.as_deref().and_then(infer_sm_count_from_gpu_name));
    probe
}

fn run_nvidia_smi_query(device_id: usize, query: &str) -> Option<String> {
    let output = Command::new("nvidia-smi")
        .args([
            format!("--query-gpu={query}"),
            "--format=csv,noheader,nounits".to_string(),
            "-i".to_string(),
            device_id.to_string(),
        ])
        .output()
        .ok()?;
    output
        .status
        .success()
        .then(|| String::from_utf8_lossy(&output.stdout).to_string())
}

fn probe_cuda_runtime_version() -> Option<String> {
    run_command_stdout("nvcc", &["--version"])
        .and_then(|output| parse_nvcc_cuda_release(&output))
        .or_else(|| {
            run_command_stdout("nvidia-smi", &[])
                .and_then(|output| parse_nvidia_smi_cuda_version(&output))
        })
}

fn run_command_stdout(command: &str, args: &[&str]) -> Option<String> {
    let output = Command::new(command).args(args).output().ok()?;
    output
        .status
        .success()
        .then(|| String::from_utf8_lossy(&output.stdout).to_string())
}

fn parse_nvidia_smi_gpu_query(output: &str) -> Option<CudaDeviceProbe> {
    let line = output.lines().find(|line| !line.trim().is_empty())?;
    let fields = line.split(',').map(str::trim).collect::<Vec<_>>();
    if fields.len() < 3 {
        return None;
    }
    let name = non_empty_probe_value(fields[0]).map(str::to_string);
    let compute_capability = non_empty_probe_value(fields[1]).map(str::to_string);
    let vram_bytes = parse_memory_total_bytes(fields[2]);
    Some(CudaDeviceProbe {
        name,
        compute_capability,
        vram_bytes,
        ..CudaDeviceProbe::default()
    })
}

fn parse_memory_total_bytes(raw: &str) -> Option<u64> {
    let lower = raw.trim().to_ascii_lowercase();
    let numeric = lower
        .trim_end_matches("mib")
        .trim_end_matches("mb")
        .trim_end_matches("gib")
        .trim_end_matches("gb")
        .trim();
    let value = numeric.parse::<f64>().ok()?;
    let multiplier = if lower.contains("gib") || lower.contains("gb") {
        1024.0 * 1024.0 * 1024.0
    } else {
        1024.0 * 1024.0
    };
    Some((value * multiplier).round() as u64)
}

fn parse_first_u32(output: &str) -> Option<u32> {
    output
        .lines()
        .find_map(|line| non_empty_probe_value(line)?.parse::<u32>().ok())
}

fn parse_nvcc_cuda_release(output: &str) -> Option<String> {
    let marker = "release ";
    let start = output.find(marker)? + marker.len();
    parse_version_prefix(&output[start..])
}

fn parse_nvidia_smi_cuda_version(output: &str) -> Option<String> {
    let marker = "CUDA Version:";
    let start = output.find(marker)? + marker.len();
    parse_version_prefix(output[start..].trim())
}

fn parse_version_prefix(raw: &str) -> Option<String> {
    let version = raw
        .chars()
        .take_while(|ch| ch.is_ascii_digit() || *ch == '.')
        .collect::<String>();
    (!version.is_empty()).then_some(version)
}

fn non_empty_probe_value(raw: &str) -> Option<&str> {
    let value = raw.trim();
    if value.is_empty() || value.eq_ignore_ascii_case("n/a") {
        None
    } else {
        Some(value)
    }
}

fn infer_sm_count_from_gpu_name(name: &str) -> Option<u32> {
    let normalized = name.to_ascii_lowercase();
    if normalized.contains("rtx 4090") {
        Some(128)
    } else {
        None
    }
}

fn compiled_kernel_features() -> CompiledKernelFeatures {
    let fa2_native = ferrum_kernels::native_ops::compiled_fa2_native_operator_artifact();
    let native_operator_artifacts =
        ferrum_kernels::native_ops::compiled_native_operator_artifacts().to_vec();
    let has_v2_fa2 = native_operator_artifacts
        .iter()
        .any(|artifact| artifact.operator == ferrum_kernels::native_ops::FA2_NATIVE_OPERATOR);
    CompiledKernelFeatures {
        cuda: cfg!(feature = "cuda"),
        vllm_paged_attn: cfg!(feature = "vllm-paged-attn-v2"),
        vllm_moe_marlin: cfg!(feature = "vllm-moe-marlin"),
        cuda_graph: cfg!(feature = "cuda"),
        greedy_argmax: cfg!(feature = "cuda") || cfg!(feature = "metal"),
        fa2_source: false,
        fa2_direct_ffi: cfg!(all(unix, feature = "cuda")),
        fa2_native_operator_artifact: fa2_native.is_some() || has_v2_fa2,
        fa2_native_operator_artifact_metadata: fa2_native.map(|artifact| {
            CompiledNativeOperatorArtifact {
                manifest_path: artifact.manifest_path.to_string(),
                artifact_path: artifact.artifact_path.to_string(),
                source_package_sha256: artifact.source_package_sha256.to_string(),
                inputs_sha256: artifact.inputs_sha256.to_string(),
                binary_sha256: artifact.binary_sha256.to_string(),
            }
        }),
        native_operator_artifacts,
    }
}

#[derive(Clone, Copy)]
struct RuntimePresetInferenceRule {
    preset: &'static str,
    architecture: ferrum_models::Architecture,
    quantization: Option<&'static str>,
    exact_hidden_size: Option<usize>,
    min_hidden_size: Option<usize>,
    exact_hidden_layers: Option<usize>,
    min_hidden_layers: Option<usize>,
    kv_heads: Option<usize>,
    num_experts: Option<u64>,
    experts_per_token: Option<u64>,
    distributed_strategy: Option<&'static str>,
    gpu_count: Option<usize>,
}

const RUNTIME_PRESET_INFERENCE_RULES: &[RuntimePresetInferenceRule] = &[
    RuntimePresetInferenceRule {
        preset: M3_QWEN3_30B_A3B_INT4_PRESET,
        architecture: ferrum_models::Architecture::Qwen3Moe,
        quantization: None,
        exact_hidden_size: Some(2048),
        min_hidden_size: None,
        exact_hidden_layers: None,
        min_hidden_layers: Some(40),
        kv_heads: Some(4),
        num_experts: Some(128),
        experts_per_token: Some(8),
        distributed_strategy: None,
        gpu_count: None,
    },
    RuntimePresetInferenceRule {
        preset: QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET,
        architecture: ferrum_models::Architecture::Qwen2,
        quantization: Some("gptq_int4"),
        exact_hidden_size: None,
        min_hidden_size: Some(8192),
        exact_hidden_layers: Some(80),
        min_hidden_layers: None,
        kv_heads: Some(8),
        num_experts: None,
        experts_per_token: None,
        distributed_strategy: Some("layer_split"),
        gpu_count: Some(2),
    },
];

fn infer_runtime_preset_for_startup(
    architecture: Option<ferrum_models::Architecture>,
    model_definition: Option<&ferrum_models::ModelDefinition>,
    gpu_selection: Option<&crate::gpu_devices::GpuDeviceSelection>,
) -> Option<&'static str> {
    let definition = model_definition?;
    RUNTIME_PRESET_INFERENCE_RULES
        .iter()
        .find(|rule| rule.matches(architecture, definition, gpu_selection))
        .map(|rule| rule.preset)
}

impl RuntimePresetInferenceRule {
    fn matches(
        &self,
        architecture: Option<ferrum_models::Architecture>,
        definition: &ferrum_models::ModelDefinition,
        gpu_selection: Option<&crate::gpu_devices::GpuDeviceSelection>,
    ) -> bool {
        if architecture != Some(self.architecture) {
            return false;
        }
        if self.quantization.is_some()
            && ferrum_models::legacy_capabilities::quantization_from_definition(definition)
                .as_deref()
                != self.quantization
        {
            return false;
        }
        if self
            .exact_hidden_size
            .is_some_and(|value| definition.hidden_size != value)
        {
            return false;
        }
        if self
            .min_hidden_size
            .is_some_and(|value| definition.hidden_size < value)
        {
            return false;
        }
        if self
            .exact_hidden_layers
            .is_some_and(|value| definition.num_hidden_layers != value)
        {
            return false;
        }
        if self
            .min_hidden_layers
            .is_some_and(|value| definition.num_hidden_layers < value)
        {
            return false;
        }
        if self
            .kv_heads
            .is_some_and(|value| definition.num_key_value_heads != Some(value))
        {
            return false;
        }
        if self.num_experts.is_some()
            && definition
                .extra_params
                .get("num_experts")
                .and_then(|value| value.as_u64())
                != self.num_experts
        {
            return false;
        }
        if self.experts_per_token.is_some()
            && definition
                .extra_params
                .get("num_experts_per_tok")
                .and_then(|value| value.as_u64())
                != self.experts_per_token
        {
            return false;
        }
        if self.distributed_strategy.is_some() || self.gpu_count.is_some() {
            let Some(selection) = gpu_selection else {
                return false;
            };
            if self
                .distributed_strategy
                .is_some_and(|value| selection.selected_distributed_strategy != value)
            {
                return false;
            }
            if self
                .gpu_count
                .is_some_and(|value| selection.selected_gpu_devices.len() != value)
            {
                return false;
            }
        }
        true
    }
}

// `find_cached_model` and `detect_format` previously lived here as forks
// of the `run.rs` versions. They moved to `crate::source_resolver` so the
// HF cache walk + format detection have a single source of truth across
// `run` / `serve` / `bench`. Use `crate::source_resolver::find_cached_model`
// / `crate::source_resolver::detect_format` directly.

fn to_candle_device(device: &ferrum_types::Device) -> ferrum_types::Result<candle_core::Device> {
    match device {
        #[cfg(all(target_os = "macos", feature = "metal"))]
        ferrum_types::Device::Metal => candle_core::Device::new_metal(0)
            .map_err(|error| ferrum_types::FerrumError::device(error.to_string())),
        #[cfg(feature = "candle-cuda-compat")]
        ferrum_types::Device::CUDA(id) => candle_core::Device::new_cuda(*id as usize)
            .map_err(|error| ferrum_types::FerrumError::device(error.to_string())),
        ferrum_types::Device::CUDA(_) => Err(ferrum_types::FerrumError::unsupported(
            "this Candle-backed server architecture requires the candle-cuda-compat feature",
        )),
        ferrum_types::Device::ROCm(_) => Err(ferrum_types::FerrumError::unsupported(
            "ROCm is not supported",
        )),
        _ => Ok(candle_core::Device::Cpu),
    }
}

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

    #[test]
    fn serve_prefix_rendezvous_is_explicit_and_rejects_zero_wait() {
        use clap::Parser;
        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            serve: ServeCliCommand,
        }
        assert!(TestCli::try_parse_from(["ferrum", "--model", "test-model"])
            .unwrap()
            .serve
            .prefix_rendezvous_max_wait_ms
            .is_none());
        let parsed = TestCli::try_parse_from([
            "ferrum",
            "--model",
            "test-model",
            "--prefix-rendezvous-max-wait-ms",
            "123",
        ])
        .unwrap();
        assert_eq!(
            parsed
                .serve
                .prefix_rendezvous_max_wait_ms
                .map(std::num::NonZeroU64::get),
            Some(123)
        );
        assert!(TestCli::try_parse_from([
            "ferrum",
            "--model",
            "test-model",
            "--prefix-rendezvous-max-wait-ms",
            "0"
        ])
        .is_err());
    }

    #[test]
    fn serve_exposes_typed_diagnostic_fault() {
        use clap::Parser;

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            serve: ServeCommand,
        }

        let parsed = TestCli::parse_from([
            "ferrum",
            "--model",
            "Qwen/Qwen3.5-4B",
            "--vnext-diagnostic-fault",
            "prefill-resource-after-submit-once",
        ]);

        assert_eq!(
            parsed.serve.vnext_diagnostic_fault,
            Some(crate::commands::VNextDiagnosticFaultArg::PrefillResourceAfterSubmitOnce)
        );
    }

    #[test]
    fn serve_parses_interleaved_system_coalescing_switches() {
        use clap::Parser;

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            serve: ServeCliCommand,
        }

        let default = TestCli::parse_from(["ferrum", "--model", "qwen3.5"]);
        assert!(!default.serve.enable_interleaved_system_coalescing);
        assert!(!default.serve.disable_interleaved_system_coalescing);
        assert!(resolve_interleaved_system_coalescing(false, false, true));
        assert!(!resolve_interleaved_system_coalescing(false, false, false));

        let disabled = TestCli::parse_from([
            "ferrum",
            "--model",
            "qwen3.5",
            "--disable-interleaved-system-coalescing",
        ]);
        assert!(disabled.serve.disable_interleaved_system_coalescing);
        assert!(!resolve_interleaved_system_coalescing(false, true, true));

        let enabled = TestCli::parse_from([
            "ferrum",
            "--model",
            "qwen3.5",
            "--enable-interleaved-system-coalescing",
        ]);
        assert!(enabled.serve.enable_interleaved_system_coalescing);
        assert!(resolve_interleaved_system_coalescing(true, false, false));

        let conflict = TestCli::try_parse_from([
            "ferrum",
            "--model",
            "qwen3.5",
            "--enable-interleaved-system-coalescing",
            "--disable-interleaved-system-coalescing",
        ]);
        assert!(conflict.is_err());
    }

    #[test]
    fn serve_rejects_removed_qwen35_flag() {
        use clap::Parser;

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            serve: ServeCommand,
        }

        let error =
            match TestCli::try_parse_from(["ferrum", "--model", "qwen3.5", "--qwen35-reference"]) {
                Ok(_) => panic!("product CLI exposed the legacy Qwen3.5 reference adapter"),
                Err(error) => error,
            };

        assert!(error.to_string().contains("--qwen35-reference"));
    }

    #[test]
    fn serve_parses_public_model_aliases() {
        use clap::Parser;

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            serve: ServeCommand,
        }

        let parsed = TestCli::parse_from([
            "ferrum",
            "--model",
            "Qwen/Qwen3.5-4B",
            "--served-model-name",
            "ferrum,qwen35",
            "--port",
            "8001",
        ]);

        assert_eq!(parsed.serve.served_model_name, ["ferrum", "qwen35"]);
    }

    #[test]
    fn served_model_names_default_and_reject_ambiguity() {
        assert_eq!(
            effective_served_model_names("Qwen/Qwen3.5-4B", Some("Qwen/Qwen3.5-4B"), vec![])
                .unwrap(),
            ["Qwen/Qwen3.5-4B"]
        );
        assert_eq!(
            effective_served_model_names("Qwen3.5-4B-Q4_K_M", Some("qwen3.5:4b-q4_k_m"), vec![])
                .unwrap(),
            ["qwen3.5:4b-q4_k_m", "Qwen3.5-4B-Q4_K_M"]
        );
        assert!(effective_served_model_names(
            "Qwen/Qwen3.5-4B",
            Some("qwen3.5:4b"),
            vec!["same".to_string(), "same".to_string()]
        )
        .is_err());
        assert!(effective_served_model_names(
            "Qwen/Qwen3.5-4B",
            Some("qwen3.5:4b"),
            vec![" ferrum".to_string()]
        )
        .is_err());
    }

    #[test]
    fn served_model_names_do_not_expose_non_hugging_face_source_names() {
        assert_eq!(
            effective_served_model_names("local-model", None, vec![]).unwrap(),
            ["local-model"]
        );
    }

    #[test]
    fn serve_cli_runtime_entries_are_cli_sourced_and_classified() {
        let mut entries = serve_cli_runtime_entries(
            Some("int8"),
            Some(1024),
            Some(4096),
            Some(4096),
            Some(64),
            Some(2048),
            Some(12_345),
            Some(8),
            Some(16),
            Some(24),
            Some(true),
            Some(false),
            Some("memory"),
            Some(16),
            Some(1024),
            Some(&PathBuf::from("/tmp/profile.jsonl")),
            Some(&PathBuf::from("/tmp/scheduler-trace.jsonl")),
            Some("abc123"),
            Some("sha256:test"),
            Some("Qwen/Qwen3-30B-A3B-GPTQ-Int4"),
            Some(32),
            Some("{\"schema_version\":1}"),
            Some(crate::layer_split_pipeline::LayerSplitPipelineModeArg::Batch),
        );
        push_sequence_fit_policy_cli_entry(
            &mut entries,
            Some(crate::commands::SequenceFitPolicyArg::FullInputMustFit),
        );
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_KV_DTYPE").effective_value, "int8");
        assert_eq!(entry("FERRUM_KV_DTYPE").source, RuntimeConfigSource::Cli);
        assert!(entry("FERRUM_KV_DTYPE")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Correctness));
        assert_eq!(entry("FERRUM_MAX_MODEL_LEN").effective_value, "4096");
        assert_eq!(entry("FERRUM_KV_CAPACITY").effective_value, "1024");
        assert_eq!(entry("FERRUM_KV_MAX_BLOCKS").effective_value, "4096");
        assert_eq!(
            entry("FERRUM_KV_MAX_BLOCKS").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(entry("FERRUM_PAGED_MAX_SEQS").effective_value, "64");
        assert_eq!(entry("FERRUM_MAX_BATCHED_TOKENS").effective_value, "2048");
        assert_eq!(
            entry("FERRUM_RUNTIME_MEMORY_BUDGET_BYTES").effective_value,
            "12345"
        );
        assert!(entry("FERRUM_RUNTIME_MEMORY_BUDGET_BYTES")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Memory));
        assert!(entry("FERRUM_RUNTIME_MEMORY_BUDGET_BYTES")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Correctness));
        assert_eq!(
            entry("FERRUM_SCHED_PREFILL_FIRST_UNTIL_ACTIVE").effective_value,
            "8"
        );
        assert_eq!(
            entry("FERRUM_SCHED_PREFILL_STEP_CHUNK").effective_value,
            "16"
        );
        assert_eq!(
            entry("FERRUM_ACTIVE_DECODE_PREFILL_CHUNK").effective_value,
            "24"
        );
        assert_eq!(entry("FERRUM_GREEDY_ARGMAX").effective_value, "1");
        assert_eq!(entry("FERRUM_PREFIX_CACHE").effective_value, "0");
        assert_eq!(entry("FERRUM_SESSION_CACHE").effective_value, "memory");
        assert_eq!(
            entry("FERRUM_SESSION_CACHE_MAX_ENTRIES").effective_value,
            "16"
        );
        assert_eq!(
            entry("FERRUM_SESSION_CACHE_MAX_TOKENS").effective_value,
            "1024"
        );
        assert_eq!(
            entry("FERRUM_MAX_MODEL_LEN").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(
            entry("FERRUM_PROFILE_JSONL").effective_value,
            "/tmp/profile.jsonl"
        );
        assert_eq!(
            entry("FERRUM_SCHEDULER_TRACE_JSONL").effective_value,
            "/tmp/scheduler-trace.jsonl"
        );
        assert_eq!(entry("FERRUM_PROFILE_ENTRYPOINT").effective_value, "serve");
        assert_eq!(
            entry("FERRUM_PROFILE_ENV_HASH").effective_value,
            "sha256:test"
        );
        assert_eq!(entry("FERRUM_PROFILE_CONCURRENCY").effective_value, "32");
        assert_eq!(
            entry("FERRUM_SEQUENCE_FIT_POLICY").effective_value,
            "full-input-must-fit"
        );
        assert_eq!(
            entry("FERRUM_SEQUENCE_FIT_POLICY").source,
            RuntimeConfigSource::Cli
        );
        assert!(entry("FERRUM_SEQUENCE_FIT_POLICY")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Memory));
        assert!(entry("FERRUM_SEQUENCE_FIT_POLICY")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Correctness));
        assert_eq!(
            entry(crate::layer_split_pipeline::LAYER_SPLIT_PIPELINE_MODE_KEY).effective_value,
            "batch"
        );
        assert!(entry("FERRUM_PROFILE_JSONL")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Diagnostics));
        assert!(entry("FERRUM_SCHEDULER_TRACE_JSONL")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Diagnostics));
    }

    #[test]
    fn cpu_serve_uses_contiguous_kv_cache() {
        assert!(matches!(
            serve_kv_cache_type_for_device(&ferrum_types::Device::CPU),
            ferrum_types::KvCacheType::Contiguous
        ));
    }

    #[cfg(any(all(target_os = "macos", feature = "metal"), feature = "cuda"))]
    #[test]
    fn accelerator_serve_uses_paged_kv_cache() {
        #[cfg(all(target_os = "macos", feature = "metal"))]
        let device = ferrum_types::Device::Metal;
        #[cfg(all(feature = "cuda", not(all(target_os = "macos", feature = "metal"))))]
        let device = ferrum_types::Device::CUDA(0);

        assert!(matches!(
            serve_kv_cache_type_for_device(&device),
            ferrum_types::KvCacheType::Paged
        ));
    }

    #[test]
    fn serve_runtime_snapshot_prefers_cli_over_config_file() {
        let config_entries = crate::config::RuntimeCliConfig {
            kv_dtype: Some("fp16".to_string()),
            sequence_fit_policy: Some(ferrum_types::SequenceFitPolicy::ImmediateOnly),
            ..Default::default()
        }
        .runtime_config_entries();
        let mut cli_entries = serve_cli_runtime_entries(
            Some("int8"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );
        push_sequence_fit_policy_cli_entry(
            &mut cli_entries,
            Some(crate::commands::SequenceFitPolicyArg::FullInputMustFit),
        );

        let snapshot = merge_runtime_config_sources(
            config_entries,
            RuntimeConfigSnapshot::default(),
            cli_entries,
        );
        let kv = snapshot
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_KV_DTYPE")
            .unwrap();
        assert_eq!(kv.effective_value, "int8");
        assert_eq!(kv.source, RuntimeConfigSource::Cli);
        let sequence_fit = snapshot
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_SEQUENCE_FIT_POLICY")
            .unwrap();
        assert_eq!(sequence_fit.effective_value, "full-input-must-fit");
        assert_eq!(sequence_fit.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn serve_runtime_snapshot_applies_recurrent_state_slots_to_engine_config() {
        let config_entries = crate::config::RuntimeCliConfig {
            recurrent_state_max_slots: Some(16),
            ..Default::default()
        }
        .runtime_config_entries();
        let snapshot = merge_runtime_config_sources(
            config_entries,
            RuntimeConfigSnapshot::default(),
            Vec::new(),
        );
        let entry = snapshot
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_RECURRENT_STATE_MAX_SLOTS")
            .expect("missing recurrent state slot entry");
        let mut engine_config = ferrum_types::EngineConfig::default();

        engine_config
            .apply_runtime_config_snapshot(&snapshot)
            .expect("serve runtime config should apply to engine config");

        assert_eq!(entry.effective_value, "16");
        assert_eq!(entry.source, RuntimeConfigSource::ConfigFile);
        assert_eq!(engine_config.runtime.recurrent_state_max_slots, Some(16));
    }

    #[test]
    fn vllm_compat_runtime_flags_follow_existing_precedence() {
        let config_entries = crate::config::RuntimeCliConfig {
            max_model_len: Some(1024),
            paged_max_seqs: Some(2),
            max_batched_tokens: Some(128),
            prefix_cache: Some(false),
            ..Default::default()
        }
        .runtime_config_entries();
        let env_snapshot = RuntimeConfigSnapshot::from_entries([
            RuntimeConfigEntry::new("FERRUM_MAX_MODEL_LEN", "2048", RuntimeConfigSource::Env),
            RuntimeConfigEntry::new("FERRUM_PAGED_MAX_SEQS", "4", RuntimeConfigSource::Env),
            RuntimeConfigEntry::new("FERRUM_MAX_BATCHED_TOKENS", "256", RuntimeConfigSource::Env),
            RuntimeConfigEntry::new("FERRUM_PREFIX_CACHE", "1", RuntimeConfigSource::Env),
        ]);

        let env_over_config =
            merge_runtime_config_sources(config_entries.clone(), env_snapshot.clone(), Vec::new());
        fn entry<'a>(snapshot: &'a RuntimeConfigSnapshot, key: &str) -> &'a RuntimeConfigEntry {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        }
        assert_eq!(
            entry(&env_over_config, "FERRUM_MAX_MODEL_LEN").effective_value,
            "2048"
        );
        assert_eq!(
            entry(&env_over_config, "FERRUM_PREFIX_CACHE").source,
            RuntimeConfigSource::Env
        );

        let cli_entries = serve_cli_runtime_entries(
            None,
            Some(1024),
            None,
            Some(4096),
            Some(8),
            Some(512),
            None,
            Some(8),
            Some(16),
            Some(32),
            Some(false),
            Some(false),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );
        let cli_over_env = merge_runtime_config_sources(config_entries, env_snapshot, cli_entries);
        assert_eq!(
            entry(&cli_over_env, "FERRUM_MAX_MODEL_LEN").effective_value,
            "4096"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_KV_CAPACITY").effective_value,
            "1024"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_PAGED_MAX_SEQS").effective_value,
            "8"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_MAX_BATCHED_TOKENS").effective_value,
            "512"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_ACTIVE_DECODE_PREFILL_CHUNK").effective_value,
            "32"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_PREFIX_CACHE").effective_value,
            "0"
        );
        assert_eq!(
            entry(&cli_over_env, "FERRUM_PREFIX_CACHE").source,
            RuntimeConfigSource::Cli
        );
    }

    #[test]
    fn nvidia_smi_gpu_query_parser_extracts_cuda_hardware_fields() {
        let probe = parse_nvidia_smi_gpu_query("NVIDIA GeForce RTX 4090, 8.9, 24564\n").unwrap();

        assert_eq!(probe.name.as_deref(), Some("NVIDIA GeForce RTX 4090"));
        assert_eq!(probe.compute_capability.as_deref(), Some("8.9"));
        assert_eq!(probe.vram_bytes, Some(24564 * 1024 * 1024));
    }

    #[test]
    fn nvidia_smi_gpu_query_parser_handles_units_and_empty_values() {
        let probe = parse_nvidia_smi_gpu_query("N/A, N/A, 24 GiB\n").unwrap();

        assert_eq!(probe.name, None);
        assert_eq!(probe.compute_capability, None);
        assert_eq!(probe.vram_bytes, Some(24 * 1024 * 1024 * 1024));
    }

    #[test]
    fn cuda_runtime_version_parsers_accept_nvcc_and_nvidia_smi_output() {
        let nvcc = "Cuda compilation tools, release 12.8, V12.8.93";
        let smi = "| NVIDIA-SMI 570.86.15    Driver Version: 570.86.15    CUDA Version: 12.8 |";

        assert_eq!(parse_nvcc_cuda_release(nvcc).as_deref(), Some("12.8"));
        assert_eq!(parse_nvidia_smi_cuda_version(smi).as_deref(), Some("12.8"));
        assert_eq!(parse_first_u32("128\n").unwrap(), 128);
        assert_eq!(
            infer_sm_count_from_gpu_name("NVIDIA GeForce RTX 4090"),
            Some(128)
        );
    }

    #[test]
    fn cuda_hardware_capabilities_uses_runtime_probe_values() {
        let hardware = cuda_hardware_capabilities(
            CompiledKernelFeatures {
                cuda: true,
                cuda_graph: true,
                ..CompiledKernelFeatures::default()
            },
            CudaDeviceProbe {
                name: Some("NVIDIA GeForce RTX 4090".to_string()),
                cuda_runtime: Some("12.8".to_string()),
                compute_capability: Some("8.9".to_string()),
                vram_bytes: Some(24 * 1024 * 1024 * 1024),
                sm_count: Some(128),
            },
        );

        assert_eq!(hardware.backend, "cuda");
        assert_eq!(hardware.cuda_runtime.as_deref(), Some("12.8"));
        assert_eq!(hardware.compute_capability.as_deref(), Some("8.9"));
        assert_eq!(hardware.vram_bytes, Some(24 * 1024 * 1024 * 1024));
        assert_eq!(hardware.sm_count, Some(128));
        assert!(hardware.supported_kv_dtypes.contains(&"int8".to_string()));
        assert!(hardware.compiled_features.cuda);
    }

    #[test]
    fn m3_runtime_preset_entries_are_cli_sourced_defaults() {
        let entries =
            runtime_preset_entries(M3_QWEN3_30B_A3B_INT4_PRESET, RuntimeConfigSource::Cli).unwrap();
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_BACKEND").effective_value, "cuda");
        assert_eq!(entry("FERRUM_MOE_GRAPH").effective_value, "0");
        assert_eq!(entry("FERRUM_VLLM_MOE").effective_value, "1");
        assert_eq!(entry("FERRUM_VLLM_MOE_PAIR_IDS").effective_value, "1");
        assert_eq!(
            entry("FERRUM_ATTENTION_POLICY").effective_value,
            "native-adaptive"
        );
        assert_eq!(entry("FERRUM_KV_CAPACITY").effective_value, "512");
        assert_eq!(entry("FERRUM_PREFIX_CACHE").effective_value, "0");
        assert_eq!(entry("FERRUM_BACKEND").source, RuntimeConfigSource::Cli);
        assert_eq!(snapshot.entries.len(), 12);
    }

    #[test]
    fn m3_runtime_preset_entries_can_be_default_sourced_for_model_inference() {
        let entries =
            runtime_preset_entries(M3_QWEN3_30B_A3B_INT4_PRESET, RuntimeConfigSource::Default)
                .unwrap();
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_MOE_GRAPH").effective_value, "0");
        assert_eq!(entry("FERRUM_VLLM_MOE").effective_value, "1");
        assert_eq!(entry("FERRUM_KV_CAPACITY").effective_value, "512");
        assert_eq!(
            entry("FERRUM_MOE_GRAPH").source,
            RuntimeConfigSource::Default
        );
        assert_eq!(
            entry("FERRUM_VLLM_MOE").source,
            RuntimeConfigSource::Default
        );
    }

    fn qwen25_72b_gptq_definition() -> ferrum_models::ModelDefinition {
        let mut definition = ferrum_models::ModelDefinition {
            architecture: ferrum_models::Architecture::Qwen2,
            hidden_size: 8192,
            num_hidden_layers: 80,
            num_key_value_heads: Some(8),
            ..Default::default()
        };
        definition.extra_params = serde_json::json!({
            "quantization_config": {
                "bits": 4,
                "quant_method": "gptq"
            }
        });
        definition
    }

    #[test]
    fn model_capabilities_prefer_measured_weight_bytes_from_model_source() {
        let mut definition = ferrum_models::ModelDefinition {
            architecture: ferrum_models::Architecture::Qwen3Moe,
            hidden_size: 2048,
            intermediate_size: 512,
            num_hidden_layers: 40,
            num_attention_heads: 16,
            num_key_value_heads: Some(2),
            max_position_embeddings: 262144,
            ..Default::default()
        };
        definition.extra_params = serde_json::json!({
            "head_dim": 256,
            "num_experts": 256,
            "num_experts_per_tok": 8,
            "moe_intermediate_size": 512,
            "shared_expert_intermediate_size": 512,
            "quantization_config": {
                "bits": 4,
                "quant_method": "gptq"
            }
        });

        let capabilities =
            model_capabilities_from_definition_with_weight_bytes(&definition, Some(19_123_456_789));

        assert_eq!(capabilities.estimated_weight_bytes, Some(19_123_456_789));
    }

    #[test]
    fn model_weight_bytes_from_path_sums_local_weight_files() {
        let workspace = tempfile::Builder::new()
            .prefix("ferrum-weight-bytes-test-")
            .tempdir()
            .expect("create temp model dir");
        let dir = workspace.path();
        std::fs::write(dir.join("model-00001-of-00002.safetensors"), vec![0u8; 7])
            .expect("write safetensors shard");
        std::fs::write(dir.join("model-00002-of-00002.safetensors"), vec![0u8; 11])
            .expect("write safetensors shard");
        std::fs::write(dir.join("tokenizer.json"), vec![0u8; 101]).expect("write non-weight file");

        let result = model_weight_bytes_from_path(dir);

        assert_eq!(result, Some(18));
    }

    fn two_gpu_layer_split_selection() -> crate::gpu_devices::GpuDeviceSelection {
        crate::gpu_devices::GpuDeviceSelection {
            raw_cli_value: "0,1".to_string(),
            requested_gpu_devices: vec![0, 1],
            selected_gpu_devices: vec![0, 1],
            cuda_device_count: 2,
            selected_distributed_strategy: "layer_split".to_string(),
            selected_layer_split_plan: Some(
                "stage0:cuda:0:layers=0-39;stage1:cuda:1:layers=40-79".to_string(),
            ),
            selected_layer_split_stages: None,
        }
    }

    #[test]
    fn qwen25_layer_split_runtime_preset_entries_are_default_sourced() {
        let entries = runtime_preset_entries(
            QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET,
            RuntimeConfigSource::Default,
        )
        .unwrap();
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(
            entry(crate::layer_split_pipeline::LAYER_SPLIT_PIPELINE_MODE_KEY).effective_value,
            "batch"
        );
        assert_eq!(entry("FERRUM_MAX_MODEL_LEN").effective_value, "4096");
        assert_eq!(entry("FERRUM_KV_MAX_BLOCKS").effective_value, "1024");
        assert_eq!(entry("FERRUM_KV_CAPACITY").effective_value, "1024");
        assert_eq!(entry("FERRUM_PAGED_MAX_SEQS").effective_value, "16");
        assert_eq!(entry("FERRUM_MAX_BATCHED_TOKENS").effective_value, "1536");
        assert_eq!(
            entry("FERRUM_SCHED_PREFILL_FIRST_UNTIL_ACTIVE").effective_value,
            "16"
        );
        assert_eq!(
            entry("FERRUM_PAGED_MAX_SEQS").source,
            RuntimeConfigSource::Default
        );
    }

    #[test]
    fn runtime_preset_inference_uses_capability_rules() {
        let definition = qwen25_72b_gptq_definition();
        let selection = two_gpu_layer_split_selection();
        assert_eq!(
            infer_runtime_preset_for_startup(
                Some(ferrum_models::Architecture::Qwen2),
                Some(&definition),
                Some(&selection),
            ),
            Some(QWEN25_72B_GPTQ_INT4_2X4090_LAYER_SPLIT_PRESET)
        );

        let mut one_gpu = selection.clone();
        one_gpu.selected_gpu_devices = vec![0];
        one_gpu.selected_distributed_strategy = "single_gpu".to_string();
        assert_eq!(
            infer_runtime_preset_for_startup(
                Some(ferrum_models::Architecture::Qwen2),
                Some(&definition),
                Some(&one_gpu),
            ),
            None
        );
    }

    #[test]
    fn qwen3_moe_serve_defaults_are_typed_default_entries() {
        let entries = crate::runtime_env::moe_graph_default_entries(
            &RuntimeConfigSnapshot::default(),
            RuntimeConfigSource::Default,
        );
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_MOE_GRAPH").effective_value, "0");
        assert_eq!(
            entry("FERRUM_MOE_GRAPH").source,
            RuntimeConfigSource::Default
        );
        assert_eq!(snapshot.entries.len(), 1);
    }

    #[test]
    fn qwen3_moe_serve_defaults_keep_config_file_overrides() {
        let config_entries = crate::config::RuntimeCliConfig {
            moe_graph: Some(false),
            vllm_moe: Some(false),
            ..Default::default()
        }
        .runtime_config_entries();
        let current = RuntimeConfigSnapshot::from_entries(config_entries.clone());
        let mut entries =
            crate::runtime_env::moe_graph_default_entries(&current, RuntimeConfigSource::Default);
        entries.extend(config_entries);
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_MOE_GRAPH").effective_value, "0");
        assert_eq!(
            entry("FERRUM_MOE_GRAPH").source,
            RuntimeConfigSource::ConfigFile
        );
        assert_eq!(entry("FERRUM_VLLM_MOE").effective_value, "0");
        assert_eq!(
            entry("FERRUM_VLLM_MOE").source,
            RuntimeConfigSource::ConfigFile
        );
    }

    #[test]
    fn model_inferred_m3_preset_keeps_config_file_overrides() {
        let mut inferred_entries =
            runtime_preset_entries(M3_QWEN3_30B_A3B_INT4_PRESET, RuntimeConfigSource::Default)
                .unwrap();
        inferred_entries.extend(
            crate::config::RuntimeCliConfig {
                prefix_cache: Some(true),
                kv_max_blocks: Some(4096),
                ..Default::default()
            }
            .runtime_config_entries(),
        );
        let snapshot = RuntimeConfigSnapshot::from_entries(inferred_entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_PREFIX_CACHE").effective_value, "1");
        assert_eq!(
            entry("FERRUM_PREFIX_CACHE").source,
            RuntimeConfigSource::ConfigFile
        );
        assert_eq!(entry("FERRUM_KV_MAX_BLOCKS").effective_value, "4096");
        assert_eq!(
            entry("FERRUM_KV_MAX_BLOCKS").source,
            RuntimeConfigSource::ConfigFile
        );
        assert_eq!(
            entry("FERRUM_MOE_GRAPH").source,
            RuntimeConfigSource::Default
        );
    }

    #[test]
    fn runtime_config_fields_override_preset_defaults_before_env() {
        let preset_entries =
            runtime_preset_entries(M3_QWEN3_30B_A3B_INT4_PRESET, RuntimeConfigSource::Cli).unwrap();
        let config_entries = crate::config::RuntimeCliConfig {
            prefix_cache: Some(true),
            kv_max_blocks: Some(4096),
            ..Default::default()
        }
        .runtime_config_entries();
        let mut non_env_entries = preset_entries;
        non_env_entries.extend(config_entries);
        let snapshot = RuntimeConfigSnapshot::from_entries(non_env_entries);
        let entry = |key: &str| {
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_PREFIX_CACHE").effective_value, "1");
        assert_eq!(
            entry("FERRUM_PREFIX_CACHE").source,
            RuntimeConfigSource::ConfigFile
        );
        assert_eq!(entry("FERRUM_KV_MAX_BLOCKS").effective_value, "4096");
        assert_eq!(
            entry("FERRUM_KV_MAX_BLOCKS").source,
            RuntimeConfigSource::ConfigFile
        );
        assert_eq!(entry("FERRUM_VLLM_MOE").effective_value, "1");
        assert_eq!(entry("FERRUM_VLLM_MOE").source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn serve_runtime_snapshot_prefers_env_over_config_file() {
        let config_entries = crate::config::RuntimeCliConfig {
            kv_dtype: Some("fp16".to_string()),
            ..Default::default()
        }
        .runtime_config_entries();
        let env_snapshot = RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
            "FERRUM_KV_DTYPE",
            "int8",
            RuntimeConfigSource::Env,
        )]);

        let snapshot = merge_runtime_config_sources(config_entries, env_snapshot, Vec::new());
        let kv = snapshot
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_KV_DTYPE")
            .unwrap();
        assert_eq!(kv.effective_value, "int8");
        assert_eq!(kv.source, RuntimeConfigSource::Env);
    }

    #[test]
    fn serve_runtime_snapshot_prefers_cli_over_env() {
        let env_snapshot = RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
            "FERRUM_KV_DTYPE",
            "int8",
            RuntimeConfigSource::Env,
        )]);
        let cli_entries = serve_cli_runtime_entries(
            Some("bf16"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );

        let snapshot = merge_runtime_config_sources(Vec::new(), env_snapshot, cli_entries);
        let kv = snapshot
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_KV_DTYPE")
            .unwrap();
        assert_eq!(kv.effective_value, "bf16");
        assert_eq!(kv.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn prefix_cache_vllm_and_product_aliases_resolve_identically() {
        assert_eq!(
            prefix_cache_cli_override(true, false, false, false),
            Some(true)
        );
        assert_eq!(
            prefix_cache_cli_override(false, false, true, false),
            Some(true)
        );
        assert_eq!(
            prefix_cache_cli_override(false, true, false, false),
            Some(false)
        );
        assert_eq!(
            prefix_cache_cli_override(false, false, false, true),
            Some(false)
        );

        let enabled_entries = serve_cli_runtime_entries(
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            prefix_cache_cli_override(true, false, false, false),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );
        let product_enabled_entries = serve_cli_runtime_entries(
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            prefix_cache_cli_override(false, false, true, false),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );
        let disabled_entries = serve_cli_runtime_entries(
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            prefix_cache_cli_override(false, true, false, false),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );
        let product_disabled_entries = serve_cli_runtime_entries(
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            prefix_cache_cli_override(false, false, false, true),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        );

        assert_eq!(enabled_entries, product_enabled_entries);
        assert_eq!(disabled_entries, product_disabled_entries);
        for (entries, enabled) in [
            (enabled_entries, true),
            (product_enabled_entries, true),
            (disabled_entries, false),
            (product_disabled_entries, false),
        ] {
            // Explicit serve flags must reach the actual typed vNext switch,
            // including disabling an already enabled config/environment value.
            let config_entries = crate::config::RuntimeCliConfig {
                prefix_cache: Some(!enabled),
                ..Default::default()
            }
            .runtime_config_entries();
            let environment = RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
                "FERRUM_PREFIX_CACHE",
                if enabled { "0" } else { "1" },
                RuntimeConfigSource::Env,
            )]);
            let effective = merge_runtime_config_sources(config_entries, environment, entries);
            let mut engine = ferrum_types::EngineConfig::default();
            engine.runtime.prefix_state_cache_enabled = !enabled;
            engine.apply_runtime_config_snapshot(&effective).unwrap();
            assert_eq!(engine.runtime.prefix_state_cache_enabled, enabled);
            assert!(!engine.runtime.prefix_cache_enabled);
        }
    }

    #[test]
    fn batched_graph_cli_override_records_flag_state() {
        assert_eq!(batched_graph_cli_override(true, false), Some(true));
        assert_eq!(batched_graph_cli_override(false, true), Some(false));
        assert_eq!(batched_graph_cli_override(false, false), None);
    }

    #[test]
    fn effective_kv_dtype_precedence_is_cli_env_config() {
        assert_eq!(
            resolve_effective_kv_dtype(Some("bf16"), Some("int8"), Some("fp16")),
            Some("bf16")
        );
        assert_eq!(
            resolve_effective_kv_dtype(None, Some("int8"), Some("fp16")),
            Some("int8")
        );
        assert_eq!(
            resolve_effective_kv_dtype(None, None, Some("fp16")),
            Some("fp16")
        );
        assert_eq!(resolve_effective_kv_dtype(None, None, None), None);
    }

    #[test]
    fn non_language_serving_rejects_kv_overrides_from_every_source() {
        use ferrum_types::KvCacheDtype;
        for kind in [
            ServedModelKind::Embedding,
            ServedModelKind::Transcription,
            ServedModelKind::Speech,
        ] {
            validate_served_model_kv_dtype(kind, KvCacheDtype::Fp16).unwrap();
            for sources in [
                (Some("int8"), Some("fp16"), Some("fp16")),
                (None, Some("int8"), Some("fp16")),
                (None, None, Some("int8")),
            ] {
                let mut engine = ferrum_types::EngineConfig::default();
                super::super::run::apply_kv_dtype_override(
                    &mut engine,
                    resolve_effective_kv_dtype(sources.0, sources.1, sources.2),
                )
                .unwrap();
                let error =
                    validate_served_model_kv_dtype(kind, engine.kv_cache.dtype).unwrap_err();
                assert!(error.to_string().contains("does not support KV dtype int8"));
            }
        }
        // The language executor validates its actual family/backend later.
        validate_served_model_kv_dtype(ServedModelKind::Llm, KvCacheDtype::Int8).unwrap();
    }

    #[test]
    fn resolved_execution_config_preserves_sources_and_records_actual_kv_plan() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("effective.json");
        let original = serde_json::json!({
            "resolution_evidence": {"source": "fixture"},
            "numerical_execution": {"requested": "auto"},
        });
        std::fs::write(&path, serde_json::to_vec(&original).unwrap()).unwrap();
        // Legacy/non-numerical snapshots must not claim the requested dtype
        // was selected merely because it appears in the input configuration.
        write_resolved_execution_config(Some(&path), Some(&serde_json::json!({}))).unwrap();
        let unchanged: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        assert_eq!(unchanged, original);

        let selected = serde_json::json!({
            "numerical_execution": {"selected_profile": "fixture.int8"},
            "kv_storage": {"source": "resolved_model_plan", "selected": "int8_per_token_head_f32_scale_v1"},
            "attention_execution_policy": "portable",
        });
        write_resolved_execution_config(Some(&path), Some(&selected)).unwrap();
        let actual: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        assert_eq!(
            actual["resolution_evidence"],
            original["resolution_evidence"]
        );
        assert_eq!(actual["kv_storage"], selected["kv_storage"]);
        assert_eq!(
            actual["numerical_execution"],
            selected["numerical_execution"]
        );
        assert_eq!(actual["attention_execution_policy"], "portable");
    }
}