ferrum-cli 0.12.1

//! Run command - Interactive chat with a model (ollama-style)

use crate::config::CliConfig;
use chrono::Utc;
use clap::{Args, ValueEnum};
use colored::*;
use console::{measure_text_width, Key, Term};
use ferrum_models::source::{ModelFormat, ResolvedModelSource};
use ferrum_server::chat_template::{ChatTemplateOptions, ModelChatTemplate, PromptMessage};
use ferrum_types::{
    has_unclosed_model_reasoning_block, model_reasoning_markers,
    parse_harmony_response_for_finish_reason, parse_model_reasoning_response,
    should_defer_model_reasoning_stream_delta, FerrumError, FinishReason, InferenceRequest,
    InferenceResponse, ModelCapabilities, ModelOutputProtocol, ParsedReasoningResponse, Priority,
    RequestId, ResolvedFerrumConfig, ResponseCompletionBoundary, Result, RuntimeConfigEntry,
    RuntimeConfigSnapshot, RuntimeConfigSource, SamplingParams, StreamChunk, TokenUsage,
    WorkloadProfile, DEFAULT_CHAT_REPETITION_PENALTY, THINK_START_TAG,
};
use futures::StreamExt;
use indicatif::{ProgressBar, ProgressStyle};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::io::{self, BufRead, IsTerminal, Write};
#[cfg(unix)]
use std::mem;
#[cfg(unix)]
use std::os::fd::{AsRawFd, RawFd};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use uuid::Uuid;

#[cfg(test)]
use crate::source_resolver::tokenizer_sibling_repo;
#[cfg(test)]
use ferrum_types::{has_unclosed_thinking_block, THINK_END_TAG};

const RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY: &str = "ferrum_initial_forbidden_token_texts";
const RUN_JSONL_SCHEMA_VERSION: u32 = 2;

/// Keep the raw history evidence separate from the model template's messages.
#[derive(Clone, Debug)]
struct RunHistoryMessage {
    prompt: PromptMessage,
    raw_content: String,
}

impl RunHistoryMessage {
    fn new(role: &str, content: &str) -> Self {
        Self {
            prompt: PromptMessage::new(role, content),
            raw_content: content.to_string(),
        }
    }

    fn assistant(
        raw_content: &str,
        protocol: ModelOutputProtocol,
        parsed: &ParsedReasoningResponse,
    ) -> Self {
        let mut message = Self::new("assistant", raw_content);
        match protocol {
            // Text templates own their policy for inline <think> history.
            ModelOutputProtocol::Text => {}
            // Reuse the validated result, including valid user/length truncation.
            // Native channel envelopes are not an assistant message's content.
            ModelOutputProtocol::HarmonyGptOss | ModelOutputProtocol::GemmaThought => {
                message.prompt.content = parsed.content.clone();
                message.prompt.reasoning_content = parsed.reasoning.clone();
            }
        }
        message
    }
}

/// Output format for `ferrum run`. JSONL mode emits one record per event
/// (assistant generation result, user input, exit) on stdout — used by
/// integration tests and scripting. Text mode is the default interactive UX.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum, Default)]
pub enum OutputFormat {
    /// Streaming text on stdout, stats on stderr (default — interactive UX).
    #[default]
    Text,
    /// One JSON record per event on stdout (machine-readable; tests).
    Jsonl,
}

fn finish_reason_str(r: FinishReason) -> &'static str {
    match r {
        FinishReason::Length => "length",
        FinishReason::Stop => "stop",
        FinishReason::EOS => "eos",
        FinishReason::Cancelled => "cancelled",
        FinishReason::Error => "error",
        FinishReason::ContentFilter => "content_filter",
    }
}

fn emit_jsonl_ready(
    session_id: &str,
    requested_model: &str,
    resolved_model: &str,
    backend: &str,
    template: Option<&ModelChatTemplate>,
) {
    let record = serde_json::json!({
        "schema_version": RUN_JSONL_SCHEMA_VERSION,
        "event": "ready",
        "session_id": session_id,
        "history_epoch": 0,
        "model": resolved_model,
        "requested_model": requested_model,
        "resolved_model": resolved_model,
        "backend": backend,
        "reasoning_protocol": template.map(ModelChatTemplate::reasoning_capability).unwrap_or_default(),
    });
    emit_jsonl_record(&record);
}

fn emit_jsonl_user(
    session_id: &str,
    history_epoch: usize,
    request_id: &str,
    turn: usize,
    content: &str,
    history: &[RunHistoryMessage],
) {
    let record = serde_json::json!({
        "schema_version": RUN_JSONL_SCHEMA_VERSION,
        "event": "user",
        "session_id": session_id,
        "history_epoch": history_epoch,
        "request_id": request_id,
        "turn": turn,
        "content": content,
        "history_before": history_evidence(history),
    });
    emit_jsonl_record(&record);
}

fn emit_jsonl_assistant_delta(
    session_id: &str,
    history_epoch: usize,
    request_id: &str,
    turn: usize,
    index: usize,
    raw_text_delta: &str,
    token_id: Option<u32>,
) {
    emit_jsonl_record(&jsonl_assistant_delta_record(
        session_id,
        history_epoch,
        request_id,
        turn,
        index,
        raw_text_delta,
        token_id,
    ));
}

fn jsonl_assistant_delta_record(
    session_id: &str,
    history_epoch: usize,
    request_id: &str,
    turn: usize,
    index: usize,
    raw_text_delta: &str,
    token_id: Option<u32>,
) -> serde_json::Value {
    serde_json::json!({
        "schema_version": RUN_JSONL_SCHEMA_VERSION,
        "event": "assistant_delta",
        "session_id": session_id,
        "history_epoch": history_epoch,
        "request_id": request_id,
        "turn": turn,
        "index": index,
        "raw_text_delta": raw_text_delta,
        "utf8_bytes": raw_text_delta.len(),
        "token_id": token_id,
    })
}

fn emit_jsonl_assistant(
    session_id: &str,
    history_epoch: usize,
    request_id: &str,
    turn: usize,
    content: &str,
    reasoning: Option<&str>,
    history: &[RunHistoryMessage],
    finish_reason: Option<FinishReason>,
    usage: Option<&TokenUsage>,
    n_tokens: usize,
    chunk_count: usize,
    raw_text: &str,
    ms: f64,
) {
    emit_jsonl_record(&jsonl_assistant_record(
        session_id,
        history_epoch,
        request_id,
        turn,
        content,
        reasoning,
        history,
        finish_reason,
        usage,
        n_tokens,
        chunk_count,
        raw_text,
        ms,
    ));
}

fn jsonl_assistant_record(
    session_id: &str,
    history_epoch: usize,
    request_id: &str,
    turn: usize,
    content: &str,
    reasoning: Option<&str>,
    history: &[RunHistoryMessage],
    finish_reason: Option<FinishReason>,
    usage: Option<&TokenUsage>,
    n_tokens: usize,
    chunk_count: usize,
    raw_text: &str,
    ms: f64,
) -> serde_json::Value {
    serde_json::json!({
        "schema_version": RUN_JSONL_SCHEMA_VERSION,
        "event": "assistant",
        "session_id": session_id,
        "history_epoch": history_epoch,
        "request_id": request_id,
        "turn": turn,
        "content": content,
        "reasoning": reasoning,
        "history_before": history_evidence(history),
        "finish_reason": finish_reason.map(finish_reason_str),
        "usage": usage,
        "n_tokens": n_tokens,
        "chunk_count": chunk_count,
        "raw_text_sha256": sha256_text(raw_text),
        "ms": ms,
    })
}

fn emit_jsonl_exit(session_id: &str, history_epoch: usize, reason: &str) {
    let record = serde_json::json!({
        "schema_version": RUN_JSONL_SCHEMA_VERSION,
        "event": "exit",
        "session_id": session_id,
        "history_epoch": history_epoch,
        "reason": reason,
    });
    emit_jsonl_record(&record);
}

fn emit_jsonl_record(record: &serde_json::Value) {
    println!("{record}");
    io::stdout().flush().ok();
}

fn sha256_text(value: &str) -> String {
    format!("{:x}", Sha256::digest(value.as_bytes()))
}

fn history_evidence(history: &[RunHistoryMessage]) -> serde_json::Value {
    let raw_history: Vec<_> = history
        .iter()
        .map(|message| (&message.prompt.role, &message.raw_content))
        .collect();
    let encoded = serde_json::to_vec(&raw_history).expect("run history serialization cannot fail");
    serde_json::json!({
        "message_count": history.len(),
        "turn_count": count_user_turns(history),
        "sha256": format!("{:x}", Sha256::digest(encoded)),
    })
}

fn run_request_metadata(
    prompt: &str,
    chat_template_options: &ChatTemplateOptions,
    protocol: ModelOutputProtocol,
    model_template: Option<&ModelChatTemplate>,
) -> HashMap<String, serde_json::Value> {
    let mut metadata = HashMap::new();
    if !has_unclosed_model_reasoning_block(protocol, prompt) {
        let mut forbidden = model_reasoning_markers(protocol)
            .map(|(_, closing)| vec![serde_json::Value::String(closing.to_string())])
            .unwrap_or_default();
        // A template may ignore the thinking option. Only an observed
        // reasoning protocol gives this opener a control-token meaning.
        if protocol == ModelOutputProtocol::Text
            && chat_template_options.enable_thinking == Some(false)
            && model_template
                .is_some_and(|template| template.reasoning_protocol.supports_reasoning())
        {
            forbidden.push(serde_json::Value::String(THINK_START_TAG.to_string()));
        }
        metadata.insert(
            RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY.to_string(),
            serde_json::Value::Array(forbidden),
        );
    }
    metadata
}

#[derive(Debug, Clone)]
struct RunPromptPlan {
    prompt: String,
    sampling_params: SamplingParams,
    prompt_token_ids: Option<Vec<u32>>,
    prompt_tokens: Option<usize>,
    kv_capacity: Option<usize>,
    dropped_history_messages: usize,
    dropped_history_turns: usize,
    max_tokens_clamped_from: Option<usize>,
}

#[derive(Debug, Clone)]
struct RunPromptTokenization {
    token_ids: Option<Vec<u32>>,
    token_count: Option<usize>,
}

struct RunBudget {
    tokenizer: Option<tokenizers::Tokenizer>,
    kv_capacity: Option<usize>,
    #[cfg(test)]
    prompt_token_id_mapper: Option<fn(&str) -> Vec<u32>>,
    #[cfg(test)]
    prompt_token_counter: Option<fn(&str) -> usize>,
}

impl RunBudget {
    fn from_product_sources(
        explicit_tokenizer: Option<&Path>,
        product_sources: Option<&ferrum_models::vnext::ProductionModelSourceBundle>,
        legacy_source_path: &Path,
        snapshot: &RuntimeConfigSnapshot,
        model_context_limit: Option<usize>,
    ) -> Result<Self> {
        let tokenizer = if let Some(path) = explicit_tokenizer {
            Some(tokenizers::Tokenizer::from_file(path).map_err(|error| {
                FerrumError::model(format!(
                    "failed to load explicit tokenizer {}: {error}",
                    path.display()
                ))
            })?)
        } else if let Some(sources) = product_sources {
            Some(
                tokenizers::Tokenizer::from_bytes(sources.tokenizer_json()).map_err(|error| {
                    FerrumError::model(format!(
                        "failed to parse tokenizer from product source {}: {error}",
                        sources.tokenizer_file().display()
                    ))
                })?,
            )
        } else {
            discover_run_tokenizer_path(legacy_source_path)
                .and_then(|path| tokenizers::Tokenizer::from_file(path).ok())
        };
        let kv_capacity = ["FERRUM_MAX_MODEL_LEN", "FERRUM_KV_CAPACITY"]
            .into_iter()
            .filter_map(|key| crate::runtime_env::runtime_snapshot_value(snapshot, key))
            .filter_map(|value| value.parse::<usize>().ok())
            .chain(model_context_limit)
            .filter(|&value| value > 0)
            .min();
        Ok(Self {
            tokenizer,
            kv_capacity,
            #[cfg(test)]
            prompt_token_id_mapper: None,
            #[cfg(test)]
            prompt_token_counter: None,
        })
    }

    fn prompt_tokenization(&self, prompt: &str) -> RunPromptTokenization {
        #[cfg(test)]
        if let Some(mapper) = self.prompt_token_id_mapper {
            let token_ids = mapper(prompt);
            return RunPromptTokenization {
                token_count: Some(token_ids.len()),
                token_ids: Some(token_ids),
            };
        }
        #[cfg(test)]
        if let Some(counter) = self.prompt_token_counter {
            return RunPromptTokenization {
                token_ids: None,
                token_count: Some(counter(prompt)),
            };
        }
        if let Some(encoding) = self
            .tokenizer
            .as_ref()
            .and_then(|tok| tok.encode(prompt, true).ok())
        {
            let token_ids = encoding.get_ids().to_vec();
            return RunPromptTokenization {
                token_count: Some(token_ids.len()),
                token_ids: Some(token_ids),
            };
        }
        RunPromptTokenization {
            token_ids: None,
            token_count: None,
        }
    }
}

fn display_response_text(text: &str) -> String {
    text.trim().to_string()
}

fn parse_run_model_output(
    protocol: ModelOutputProtocol,
    text: &str,
    prompt_opened_thinking: bool,
    finish_reason: Option<FinishReason>,
) -> Result<ParsedReasoningResponse> {
    match protocol {
        ModelOutputProtocol::Text | ModelOutputProtocol::GemmaThought => {
            parse_model_reasoning_response(protocol, text, prompt_opened_thinking)
        }
        ModelOutputProtocol::HarmonyGptOss => {
            let parsed = parse_harmony_response_for_finish_reason(text, finish_reason)?;
            if parsed.tool_call.is_some() {
                return Err(FerrumError::invalid_format(
                    "GPT-OSS emitted a Harmony tool call for `ferrum run`, which has no tool executor",
                ));
            }
            Ok(ParsedReasoningResponse {
                content: parsed.content,
                reasoning: parsed.reasoning_content,
            })
        }
    }
}

struct CollectedRunGeneration {
    request_id: String,
    raw_text: String,
    finish_reason: Option<FinishReason>,
    usage: Option<TokenUsage>,
    token_count: usize,
    token_ids: Vec<u32>,
    chunk_count: usize,
    execution_evidence: Option<ferrum_types::InferenceExecutionEvidence>,
}

impl CollectedRunGeneration {
    fn from_response(response: InferenceResponse) -> Result<Self> {
        require_run_terminal_reason(Some(response.finish_reason))?;
        Ok(Self {
            request_id: response.request_id.to_string(),
            raw_text: response.text,
            finish_reason: Some(response.finish_reason),
            usage: Some(response.usage),
            token_count: response.tokens.len(),
            token_ids: response
                .tokens
                .into_iter()
                .map(|token| token.get())
                .collect(),
            chunk_count: 1,
            execution_evidence: response.execution_evidence,
        })
    }
}

fn require_run_terminal_reason(reason: Option<FinishReason>) -> Result<()> {
    match reason {
        Some(FinishReason::Error) => Err(FerrumError::model(
            "Generation failed (finish_reason=error)",
        )),
        None => Err(FerrumError::internal(
            "Generation stream ended without a terminal finish reason",
        )),
        Some(_) => Ok(()),
    }
}

type RunResponseStream = Pin<Box<dyn futures::Stream<Item = Result<StreamChunk>> + Send + 'static>>;

struct RunStreamOutput {
    protocol: ModelOutputProtocol,
    prompt_opened_thinking: bool,
    emitted_content: String,
}

impl RunStreamOutput {
    fn new(protocol: ModelOutputProtocol, prompt_opened_thinking: bool) -> Self {
        Self {
            protocol,
            prompt_opened_thinking,
            emitted_content: String::new(),
        }
    }

    fn delta(&mut self, raw_text: &str, raw_delta: &str) -> Result<Option<String>> {
        match self.protocol {
            ModelOutputProtocol::Text => Ok(Some(raw_delta.to_string())),
            ModelOutputProtocol::HarmonyGptOss => Ok(None),
            ModelOutputProtocol::GemmaThought => {
                if should_defer_model_reasoning_stream_delta(self.protocol, raw_text) {
                    return Ok(None);
                }
                self.parsed_content_delta(raw_text)
            }
        }
    }

    fn finish(&mut self, raw_text: &str) -> Result<Option<String>> {
        match self.protocol {
            // Ordinary text has already streamed; Harmony is validated and
            // displayed as a complete message by the caller.
            ModelOutputProtocol::Text | ModelOutputProtocol::HarmonyGptOss => Ok(None),
            // A last chunk may contain visible text followed by an incomplete
            // channel marker. Flush its safe visible prefix without the marker.
            ModelOutputProtocol::GemmaThought => self.parsed_content_delta(raw_text),
        }
    }

    fn parsed_content_delta(&mut self, raw_text: &str) -> Result<Option<String>> {
        let parsed =
            parse_model_reasoning_response(self.protocol, raw_text, self.prompt_opened_thinking)?;
        let delta = parsed
            .content
            .strip_prefix(&self.emitted_content)
            .ok_or_else(|| {
                FerrumError::invalid_format(
                    "model reasoning parsing changed previously emitted content",
                )
            })?
            .to_string();
        self.emitted_content = parsed.content;
        Ok((!delta.is_empty()).then_some(delta))
    }
}

async fn collect_run_stream(
    mut stream: RunResponseStream,
    trace_tokens: bool,
    buffer_output: bool,
    turn: usize,
    session_id: &str,
    history_epoch: usize,
    expected_request_id: &str,
) -> Result<CollectedRunGeneration> {
    let mut request_id = None;
    let mut raw_text = String::new();
    let mut finish_reason = None;
    let mut latest_usage = None;
    let mut token_count = 0usize;
    let mut token_ids = Vec::new();
    let mut chunk_count = 0usize;
    let mut execution_evidence = None;
    while let Some(chunk) = stream.next().await {
        let mut chunk = chunk?;
        let chunk_request_id = chunk.request_id.to_string();
        if chunk_request_id != expected_request_id {
            return Err(FerrumError::internal(format!(
                "run stream request id drift: expected {expected_request_id}, got {chunk_request_id}"
            )));
        }
        request_id.get_or_insert_with(|| chunk_request_id.clone());
        let token_id = chunk.token.map(|token| token.get());
        if !chunk.text.is_empty() {
            raw_text.push_str(&chunk.text);
            // JSONL raw_text_delta is an evidence field: concatenation must
            // match raw_text_sha256, including native channel framing. The
            // final assistant record exposes parsed content and reasoning.
            if !buffer_output {
                emit_jsonl_assistant_delta(
                    session_id,
                    history_epoch,
                    expected_request_id,
                    turn,
                    chunk_count,
                    &chunk.text,
                    token_id,
                );
            }
            chunk_count += 1;
        }
        if let Some(token_id) = token_id {
            if trace_tokens {
                eprintln!(
                    "[run-token-trace] turn={turn} token={} text={:?}",
                    token_id, chunk.text
                );
            }
            token_ids.push(token_id);
            token_count += 1;
        }
        if let Some(usage) = chunk.usage.as_ref() {
            token_count = usage.completion_tokens;
            latest_usage = Some(usage.clone());
        }
        if chunk.finish_reason.is_some() {
            finish_reason = chunk.finish_reason;
        }
        if let Some(evidence) = chunk.execution_evidence.take() {
            if execution_evidence.replace(evidence).is_some() {
                return Err(FerrumError::internal(
                    "run stream emitted engine execution evidence more than once",
                ));
            }
        }
    }
    require_run_terminal_reason(finish_reason)?;
    Ok(CollectedRunGeneration {
        request_id: request_id.unwrap_or_else(|| expected_request_id.to_string()),
        raw_text,
        finish_reason,
        usage: latest_usage,
        token_count,
        token_ids,
        chunk_count,
        execution_evidence,
    })
}

async fn collect_run_text_stream(
    mut stream: RunResponseStream,
    trace_tokens: bool,
    mut output: RunStreamOutput,
    turn: usize,
    expected_request_id: &RequestId,
    stdin_is_tty: bool,
    capture_token_ids: bool,
) -> Result<CollectedRunGeneration> {
    let mut first_token_indicator = start_first_token_indicator(stdin_is_tty);
    let mut request_id = None;
    let mut raw_text = String::new();
    let mut finish_reason = None;
    let mut latest_usage = None;
    let mut token_count = 0usize;
    let mut token_ids = Vec::new();
    let mut chunk_count = 0usize;
    let mut execution_evidence = None;
    while let Some(chunk) = stream.next().await {
        let mut chunk = match chunk {
            Ok(chunk) => chunk,
            Err(error) => {
                clear_first_token_indicator(&mut first_token_indicator);
                return Err(error);
            }
        };
        if &chunk.request_id != expected_request_id {
            clear_first_token_indicator(&mut first_token_indicator);
            return Err(FerrumError::internal(format!(
                "run stream request id drift: expected {expected_request_id}, got {}",
                chunk.request_id
            )));
        }
        request_id.get_or_insert_with(|| chunk.request_id.clone());
        let token_id = chunk.token.map(|token| token.get());
        if first_token_indicator.is_some()
            && (!chunk.text.is_empty() || token_id.is_some() || chunk.finish_reason.is_some())
        {
            clear_first_token_indicator(&mut first_token_indicator);
        }
        if trace_tokens {
            if let Some(token_id) = token_id {
                eprintln!(
                    "[run-token-trace] turn={turn} token={} text={:?}",
                    token_id, chunk.text
                );
            }
        }
        if !chunk.text.is_empty() {
            raw_text.push_str(&chunk.text);
            let delta = output.delta(&raw_text, &chunk.text).inspect_err(|_| {
                clear_first_token_indicator(&mut first_token_indicator);
            })?;
            if let Some(delta) = delta {
                print!("{delta}");
                io::stdout().flush().ok();
            }
            chunk_count += 1;
        }
        if let Some(token_id) = token_id {
            if capture_token_ids {
                token_ids.push(token_id);
            }
            token_count += 1;
        }
        if let Some(usage) = chunk.usage.as_ref() {
            token_count = usage.completion_tokens;
            latest_usage = Some(usage.clone());
        }
        if chunk.finish_reason.is_some() {
            finish_reason = chunk.finish_reason;
        }
        if let Some(evidence) = chunk.execution_evidence.take() {
            if execution_evidence.replace(evidence).is_some() {
                clear_first_token_indicator(&mut first_token_indicator);
                return Err(FerrumError::internal(
                    "run stream emitted engine execution evidence more than once",
                ));
            }
        }
    }
    clear_first_token_indicator(&mut first_token_indicator);
    require_run_terminal_reason(finish_reason)?;
    if let Some(delta) = output.finish(&raw_text)? {
        print!("{delta}");
        io::stdout().flush().ok();
    }
    Ok(CollectedRunGeneration {
        request_id: request_id
            .unwrap_or_else(|| expected_request_id.clone())
            .to_string(),
        raw_text,
        finish_reason,
        usage: latest_usage,
        token_count,
        token_ids,
        chunk_count,
        execution_evidence,
    })
}

#[derive(Args)]
pub struct RunCommand {
    /// Model name (alias, owner/repository[:QUANT], local directory, or `.gguf` file).
    #[arg(value_name = "MODEL")]
    pub model: Option<String>,

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

    /// System prompt (interactive chat mode only).
    #[arg(long)]
    pub system: Option<String>,

    /// Maximum output-token ceiling. Context planning shrinks it to the
    /// remaining KV capacity before dropping any complete history turn.
    #[arg(long, default_value = "4096")]
    pub max_tokens: u32,

    /// Stop generation when this text appears. Can be provided multiple times.
    #[arg(long, value_name = "TEXT")]
    pub stop: Vec<String>,

    /// Enable model reasoning for chat templates that support it.
    #[arg(long, conflicts_with = "disable_thinking")]
    pub enable_thinking: bool,

    /// Disable model reasoning for chat templates that support it.
    #[arg(long, conflicts_with = "enable_thinking")]
    pub disable_thinking: bool,

    /// Disable CLI context shift. By default, `ferrum run` keeps the REPL
    /// alive by shrinking this turn's output budget before dropping history.
    /// Oldest complete turns are removed only when the rendered prompt itself
    /// no longer fits in KV.
    #[arg(long)]
    pub no_context_shift: bool,

    /// Sampling temperature (0.0–2.0). 0.0 = greedy / argmax (deterministic,
    /// what you want for benchmarks). >0 = softmax sample with `--top-k`
    /// and `--top-p` filtering applied.
    #[arg(long, default_value = "0.0")]
    pub temperature: f32,

    /// Backend: auto, cpu, metal, cuda (default: auto)
    #[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 runs.
    #[arg(long, value_enum)]
    pub layer_split_pipeline_mode: Option<crate::layer_split_pipeline::LayerSplitPipelineModeArg>,

    /// One-shot prompt (skip interactive REPL). When supplied, ferrum runs a
    /// single prefill+decode and exits — useful for benchmarking and shell
    /// scripting. For `.gguf` paths, omitting this drops into the GGUF REPL.
    #[arg(long)]
    pub prompt: Option<String>,

    /// Path to a HuggingFace `tokenizer.json` (only used for `.gguf` paths).
    /// If omitted, ferrum looks for `<gguf-stem>.tokenizer.json` and then
    /// `tokenizer.json` next to the `.gguf` file.
    #[arg(long)]
    pub tokenizer: Option<PathBuf>,

    /// Bench mode: skip generated text output, print only timing summary.
    /// Implies one-shot (`--prompt` is required).
    #[arg(long)]
    pub bench_mode: bool,

    /// Top-K sampling cutoff (0 disables — keep all). Only the K highest-
    /// probability tokens compete in the softmax sample. Default 50, a
    /// conservative value that filters obvious garbage without flattening
    /// the distribution.
    #[arg(long, default_value = "50")]
    pub top_k: usize,

    /// Top-P (nucleus) sampling cutoff (0.0 disables, 1.0 keeps all).
    /// Smallest set of tokens whose cumulative probability exceeds P
    /// is kept; the rest are zeroed before sampling. Default 0.95.
    #[arg(long, default_value = "0.95")]
    pub top_p: f32,

    /// Minimum probability cutoff relative to the most likely token.
    /// A value of 0 disables min-p filtering.
    #[arg(long, default_value = "0.0")]
    pub min_p: f32,

    /// Presence penalty applied to tokens that already occurred in the
    /// request. Qwen3.5 recommends 1.5 for general thinking workloads.
    #[arg(long, default_value = "0.0")]
    pub presence_penalty: f32,

    /// Repetition penalty applied to logits before sampling. >1 discourages
    /// repeats, <1 encourages, 1.0 disables. Defaults to 1.1 (OpenAI/llama.cpp
    /// standard) because the chat default is greedy (temperature 0): greedy
    /// with no penalty deterministically locks into token loops on some
    /// inputs (the "2D/3D 2D/3D..." degeneration). Pass `--repeat-penalty 1.0`
    /// for an unpenalized greedy baseline.
    #[arg(long, default_value_t = DEFAULT_CHAT_REPETITION_PENALTY)]
    pub repeat_penalty: f32,

    /// Number of recent tokens that the repetition penalty considers.
    /// Smaller = local repeat avoidance only.
    #[arg(long, default_value = "64")]
    pub repeat_last_n: usize,

    /// Random seed for sampling (when temperature > 0). Omit for non-deterministic chat.
    #[arg(long)]
    pub seed: Option<u64>,

    /// 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 a typed capacity ceiling shared by `run`
    /// and `serve`; omit it to use the normal pressure-threshold policy.
    #[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>,
    /// Wait at most this long for another request's exact in-flight prefix.
    #[arg(long, value_name = "MS")]
    pub prefix_rendezvous_max_wait_ms: Option<std::num::NonZeroU64>,

    /// 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,

    /// 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>,

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

    /// Write resolved startup runtime config JSON and exit artifacts.
    #[arg(long)]
    pub effective_config_json: Option<PathBuf>,

    /// Write one auto-config decision JSON record per line.
    #[arg(long)]
    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 product observability 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/admission 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,

    /// Output format. `text` (default) — streaming text + stats UX.
    /// `jsonl` — one JSON record per event on stdout; used by tests and scripts.
    #[arg(long, value_enum, default_value_t = OutputFormat::Text)]
    pub output_format: OutputFormat,
}

pub async fn execute(cmd: RunCommand, config: CliConfig) -> Result<()> {
    if let Some(out_dir) = cmd.observability_vertical_slice_out.as_ref() {
        crate::observability_vertical_slice::write_observability_vertical_slice(
            ferrum_types::ProfileEntrypoint::Run,
            out_dir,
        )?;
        println!(
            "OBSERVABILITY VERTICAL SLICE ARTIFACT: {}",
            out_dir.display()
        );
        return Ok(());
    }
    let model = cmd.model.as_deref().ok_or_else(|| {
        FerrumError::config(crate::source_resolver::first_success_model_help("run"))
    })?;
    let product_observability = crate::observability_product::ProductObservabilityConfig::new(
        ferrum_types::ProfileEntrypoint::Run,
        model,
        cmd.profile_jsonl.as_ref(),
        cmd.profile_detail,
        cmd.memory_profile_jsonl.as_ref(),
        cmd.scheduler_trace_jsonl.as_ref(),
        cmd.request_dump_dir.as_ref(),
        cmd.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(());
    }

    // Capture user input once. Inferred startup values never re-enter this
    // snapshot through the process environment.
    let user_environment = RuntimeConfigSnapshot::capture_current();

    // Select device before model resolution so CPU runs do not materialize
    // GPU/Metal chat-profile defaults such as paged KV.
    let mut device = select_device(&cmd.backend)?;
    let mut gpu_selection =
        crate::gpu_devices::resolve_cuda_gpu_devices(cmd.gpu_devices.as_deref(), &device)?;
    if let Some(selection) = &gpu_selection {
        device = selection.primary_device();
        eprintln!(
            "{} {} ({})",
            "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 = process_memory_observation_between(
        process_start_sample.clone(),
        backend_initialized_sample.clone(),
    );
    let mut startup_cli_runtime_entries =
        run_startup_cli_runtime_entries(&cmd, gpu_selection.as_ref());
    let early_runtime_config = run_base_runtime_config(&config, user_environment);
    let early_effective_runtime_config =
        run_effective_runtime_config(&early_runtime_config, &startup_cli_runtime_entries);
    // Source resolution selects artifacts only. Resource policy follows the
    // registered model definition, once its execution authority is known.
    let cache_dir = crate::source_resolver::hf_cache_dir(&config);
    let resolved = crate::source_resolver::resolve_model_source_with_product_sources(
        model,
        &cache_dir,
        crate::source_resolver::DownloadPolicy::AutoDownload,
        None,
        &cmd.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 engine_config = product_input.engine_config;
    engine_config.numerical_execution =
        config.resolve_numerical_execution(cmd.numerical_profile.as_ref());
    // Family candidates and startup sizing must see the same KV request as
    // the eventual executor; resolving it after model definition loses this
    // constraint when the numerical policy is Auto.
    apply_kv_dtype_override(
        &mut engine_config,
        crate::runtime_env::runtime_snapshot_value(
            &early_effective_runtime_config,
            "FERRUM_KV_DTYPE",
        ),
    )?;
    let model_sources = product_input.model_sources;
    let defined_model = crate::source_resolver::define_registered_product_model(
        model_sources.as_ref(),
        &engine_config.numerical_execution,
        engine_config.kv_cache.dtype,
    )?;
    let model_definition_for_config = if defined_model.is_none() {
        load_run_model_definition(&source, model_sources.as_deref()).await?
    } else {
        None
    };
    let model_layer_count = defined_model
        .as_ref()
        .map(|prepared| prepared.descriptor().layer_count())
        .or_else(|| {
            model_definition_for_config
                .as_ref()
                .map(|definition| definition.num_hidden_layers)
        });
    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() {
                eprintln!("{}", format!("CUDA layer split plan: {plan}").dimmed());
            }
            startup_cli_runtime_entries =
                run_startup_cli_runtime_entries(&cmd, gpu_selection.as_ref());
        }
    }
    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 chat_template_options = build_chat_template_options(&cmd, model_chat_template.as_ref());
    eprintln!("{}", format!("Loading {}...", model_id).dimmed());

    let engine_model_path = source.local_path.to_string_lossy().to_string();

    let device_label = format!("{device:?}");
    eprintln!("{}", format!("Using {device_label} backend").dimmed());

    // Create engine. Big-model loads (15-60 GB safetensors) are slow on
    // first run — print a hint so users don't think it's frozen. Per-
    // layer INFO logs fire from the model loaders once parsing starts;
    // utils::setup_logging whitelists them at INFO level.
    eprintln!(
        "{}",
        "Loading model weights... (30s+ for >10 GB models)".dimmed()
    );
    let load_start = std::time::Instant::now();
    engine_config.sampling.default_params = build_sampling_params(&cmd);
    if let Some(prepared) = defined_model.as_deref() {
        engine_config.sampling.default_params.model_output_protocol =
            prepared.descriptor().output_protocol();
    }
    engine_config.backend.device = device.clone();
    engine_config.scheduler.policy = ferrum_types::SchedulingPolicy::ContinuousBatch;
    engine_config.backend.backend_options.insert(
        "model_path".to_string(),
        serde_json::Value::String(engine_model_path),
    );
    let runtime_config = early_runtime_config;
    if let Some(selection) = &gpu_selection {
        selection.insert_backend_options(&mut engine_config.backend.backend_options);
    }
    let mut effective_runtime_config =
        run_effective_runtime_config(&runtime_config, &startup_cli_runtime_entries);
    crate::layer_split_pipeline::insert_backend_option_from_runtime(
        &effective_runtime_config,
        &mut engine_config.backend.backend_options,
    )?;
    let execution_resource_authority = if defined_model.is_some() {
        ferrum_types::ExecutionResourceAuthority::PlanRuntime
    } else {
        ferrum_types::ExecutionResourceAuthority::LegacyEngine
    };
    if execution_resource_authority == ferrum_types::ExecutionResourceAuthority::LegacyEngine {
        let defaults = crate::runtime_env::moe_graph_default_entries(
            &effective_runtime_config,
            RuntimeConfigSource::Default,
        );
        for entry in defaults {
            effective_runtime_config.upsert_entry(entry);
        }
        if let Some((profile, utilization)) =
            run_autosize_for_device(&device, cmd.gpu_memory_utilization)
        {
            let entries = crate::gpu_mem_autosize::auto_size_runtime_entries(
                &source.local_path,
                utilization,
                profile,
                &effective_runtime_config,
            );
            for entry in entries {
                effective_runtime_config.upsert_entry(entry);
            }
            let entries = crate::source_resolver::chat_profile_runtime_entries(
                &source.local_path,
                &effective_runtime_config,
                RuntimeConfigSource::Default,
            );
            for entry in entries {
                effective_runtime_config.upsert_entry(entry);
            }
        }
        let entries = crate::source_resolver::metal_gguf_moe_correctness_entries(
            &source.local_path,
            &device,
            &effective_runtime_config,
            RuntimeConfigSource::Default,
        );
        for entry in entries {
            effective_runtime_config.upsert_entry(entry);
        }
    }
    let typed_model_capabilities = defined_model
        .as_ref()
        .map(|defined| {
            defined.model_capabilities(
                &engine_config.numerical_execution,
                ferrum_types::KvStorageFormat::try_from(engine_config.kv_cache.dtype)
                    .map_err(ferrum_types::FerrumError::config)?,
            )
        })
        .transpose()?;
    let startup_memory_request = crate::startup::memory_request(
        &device,
        execution_resource_authority,
        cmd.gpu_memory_utilization,
        &effective_runtime_config,
    )?;
    let hardware = crate::startup::hardware_for_request(&device, startup_memory_request.as_ref());
    let mut startup_auto_config = run_startup_auto_config(
        hardware,
        typed_model_capabilities,
        execution_resource_authority,
        model_definition_for_config.as_ref(),
        crate::commands::serve::model_weight_bytes_from_path(&source.local_path),
        effective_runtime_config,
    )?;
    // One-way compatibility bridge for backend readers. Resource ownership
    // remains with the typed engine config and its compiled memory plan.
    crate::runtime_env::materialize_runtime_env_effective(&startup_auto_config.runtime_config);
    engine_config
        .apply_runtime_config_snapshot(&startup_auto_config.runtime_config)
        .map_err(ferrum_types::FerrumError::config)?;
    engine_config.runtime.startup_memory_request = startup_memory_request;
    let vnext_checkpoint_capture = cmd.vnext_checkpoint.to_config()?;
    validate_teacher_forced_checkpoint_run(&cmd, vnext_checkpoint_capture.as_ref())?;
    let teacher_forcing = vnext_checkpoint_capture
        .as_ref()
        .and_then(|capture| capture.teacher_forcing.clone());
    engine_config.runtime.vnext_checkpoint_capture = vnext_checkpoint_capture;
    if runtime_config_bool(&startup_auto_config.runtime_config, "FERRUM_PAGED_KV")
        .or_else(|| {
            runtime_config_bool(&startup_auto_config.runtime_config, "FERRUM_METAL_PAGED_KV")
        })
        .unwrap_or(false)
    {
        engine_config.kv_cache.cache_type = ferrum_types::KvCacheType::Paged;
    }
    let effective_kv_dtype = cmd
        .kv_dtype
        .as_deref()
        .or_else(|| crate::runtime_env::runtime_snapshot_value(&runtime_config, "FERRUM_KV_DTYPE"));
    apply_kv_dtype_override(&mut engine_config, effective_kv_dtype)?;
    let numerical_execution = engine_config.numerical_execution.clone();
    let engine_result = match (defined_model, model_sources.clone()) {
        (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,
    };
    let engine = match engine_result {
        Ok(engine) => engine,
        Err(error) => {
            crate::commands::serve::write_failed_startup_config_artifacts(
                &startup_auto_config,
                product_source_identity.as_ref(),
                &numerical_execution,
                cmd.effective_config_json.as_deref(),
                cmd.decision_trace_jsonl.as_deref(),
                &error,
            );
            return Err(error);
        }
    };
    crate::startup::apply_engine_plan(&mut startup_auto_config, engine.config());
    crate::commands::serve::write_startup_config_artifacts(
        &startup_auto_config,
        product_source_identity.as_ref(),
        &engine.config().numerical_execution,
        cmd.effective_config_json.as_deref(),
        cmd.decision_trace_jsonl.as_deref(),
    )?;
    let run_budget = RunBudget::from_product_sources(
        cmd.tokenizer.as_deref(),
        model_sources.as_deref(),
        &source.local_path,
        &startup_auto_config.runtime_config,
        engine.context_capacity(),
    )?;
    crate::commands::serve::write_resolved_execution_config(
        cmd.effective_config_json.as_deref(),
        engine.cache_metrics_snapshot().as_ref(),
    )?;
    let model_loaded_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let model_loaded_memory = process_memory_observation_between(
        backend_initialized_sample
            .clone()
            .or_else(|| process_start_sample.clone()),
        model_loaded_sample.clone(),
    );
    let model_loaded_duration_us = load_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 = process_memory_observation_between(
        model_loaded_sample.clone(),
        profile_run_done_sample.clone(),
    );
    let cache_allocated_status = if product_memory_enabled {
        Some(engine.status().await)
    } else {
        None
    };
    let cache_allocated_sample = product_memory_enabled
        .then(|| memory_sampler.sample())
        .flatten();
    let cache_allocated_memory = process_memory_observation_between(
        profile_run_done_sample
            .clone()
            .or_else(|| model_loaded_sample.clone()),
        cache_allocated_sample.clone(),
    );
    eprintln!(
        "{}",
        format!(
            "Model loaded in {:.1}s.",
            load_start.elapsed().as_secs_f64()
        )
        .dimmed()
    );
    let run_session_id = Uuid::new_v4().to_string();

    // One-shot mode: --prompt supplied → run a single request and exit.
    // Matches the GGUF run_gguf_one_shot UX. Previously cmd.prompt was
    // documented as "one-shot for non-interactive runs" but the alias
    // path ignored it and dropped into REPL, which exits silently when
    // stdin is not a TTY.
    if let Some(one_shot) = cmd.prompt.clone() {
        let format = cmd.output_format;
        let plan = build_run_prompt_plan(
            &[],
            &one_shot,
            cmd.system.as_deref(),
            &model_id,
            model_chat_template.as_ref(),
            &chat_template_options,
            &cmd,
            &run_budget,
        )?;
        if let Some(teacher) = &teacher_forcing {
            if plan.sampling_params.max_tokens != teacher.token_count() {
                return Err(ferrum_types::FerrumError::config(format!(
                    "teacher-forced checkpoint requested {} tokens, but context planning resolved max_tokens={}",
                    teacher.token_count(),
                    plan.sampling_params.max_tokens
                )));
            }
        }
        maybe_warn_context_shift(&plan, format);
        let model_output_protocol = plan.sampling_params.model_output_protocol;
        let prompt_opened_thinking =
            has_unclosed_model_reasoning_block(model_output_protocol, &plan.prompt);
        let metadata = run_request_metadata(
            &plan.prompt,
            &chat_template_options,
            model_output_protocol,
            model_chat_template.as_ref(),
        );
        let prompt_chars = plan.prompt.chars().count();
        let request_id = RequestId(Uuid::new_v4());
        let request_id_text = request_id.to_string();
        let one_shot_history = Vec::new();
        if format == OutputFormat::Jsonl {
            emit_jsonl_ready(
                &run_session_id,
                &requested_model,
                &model_id,
                &device_label,
                model_chat_template.as_ref(),
            );
            emit_jsonl_user(
                &run_session_id,
                0,
                &request_id_text,
                0,
                &one_shot,
                &one_shot_history,
            );
        }
        let request = InferenceRequest {
            id: request_id,
            model_id: ferrum_types::ModelId(model_id.clone()),
            prompt: plan.prompt,
            sampling_params: plan.sampling_params.clone(),
            stream: format == OutputFormat::Jsonl,
            priority: Priority::Normal,
            client_id: None,
            session_id: None,
            created_at: Utc::now(),
            api_request: None,
            evidence_request: ferrum_types::InferenceEvidenceRequest {
                capture_engine_token_timing: product_observability
                    .profile_detail
                    .captures_engine_token_timing(),
                ..Default::default()
            },
            metadata,
        };
        let profile_request_id = request.id.to_string();
        let memory_before = product_observability
            .enabled()
            .then(|| memory_sampler.sample())
            .flatten();
        let start = std::time::Instant::now();
        let trace_tokens =
            crate::runtime_env::runtime_snapshot_value(&runtime_config, "FERRUM_RUN_TRACE_TOKENS")
                .is_some();
        let generation_result = match format {
            OutputFormat::Text => engine
                .infer(request)
                .await
                .and_then(CollectedRunGeneration::from_response),
            OutputFormat::Jsonl => match engine.infer_stream(request).await {
                Ok(stream) => {
                    collect_run_stream(
                        stream,
                        trace_tokens,
                        model_output_protocol == ModelOutputProtocol::HarmonyGptOss,
                        0,
                        &run_session_id,
                        0,
                        &request_id_text,
                    )
                    .await
                }
                Err(error) => Err(error),
            },
        };
        let generation = match generation_result {
            Ok(generation) => generation,
            Err(err) => {
                let memory_after = product_memory_enabled
                    .then(|| memory_sampler.sample())
                    .flatten();
                let memory = process_memory_observation_between(memory_before, memory_after);
                let elapsed = start.elapsed().as_secs_f64();
                if let Err(observability_err) =
                    crate::observability_product::write_actual_run_failure_observability(
                        &product_observability,
                        &crate::observability_product::ActualRunFailureObservation {
                            request_id: profile_request_id,
                            duration_us: (elapsed * 1_000_000.0).max(0.0) as u64,
                            sampling_params: plan.sampling_params.clone(),
                            prompt_token_ids: plan.prompt_token_ids.clone(),
                            prompt_token_count: plan.prompt_tokens,
                            prompt_chars,
                            failure_kind: err.observability_failure_kind().to_string(),
                            error_kind: err.observability_error_kind().to_string(),
                            error_message: err.to_string(),
                            memory,
                            memory_stages: actual_run_memory_stages(
                                product_memory_enabled,
                                process_start_memory.clone(),
                                backend_initialized_memory.clone(),
                                model_loaded_memory.clone(),
                                model_loaded_duration_us,
                                profile_run_done_memory.clone(),
                                cache_allocated_memory.clone(),
                                cache_allocated_status.clone(),
                                None,
                            ),
                        },
                    )
                {
                    eprintln!("failed to write run failure observability: {observability_err}");
                }
                return Err(err);
            }
        };
        let memory_after = product_memory_enabled
            .then(|| memory_sampler.sample())
            .flatten();
        let memory = process_memory_observation_between(memory_before, memory_after.clone());
        let CollectedRunGeneration {
            request_id: response_request_id,
            raw_text,
            finish_reason,
            usage,
            token_count: tokens,
            token_ids: output_token_ids,
            chunk_count,
            execution_evidence,
        } = generation;
        if response_request_id != request_id_text {
            return Err(FerrumError::internal(format!(
                "run response request id drift: expected {request_id_text}, got {response_request_id}"
            )));
        }
        let raw_response = display_response_text(&raw_text);
        let parsed = parse_run_model_output(
            model_output_protocol,
            &raw_response,
            prompt_opened_thinking,
            finish_reason,
        )?;
        let content = display_response_text(&parsed.content);
        let reasoning = parsed
            .reasoning
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let bench = cmd.bench_mode;
        if format == OutputFormat::Text && !bench {
            print!(
                "{}",
                if model_output_protocol != ModelOutputProtocol::Text {
                    &content
                } else {
                    &raw_response
                }
            );
            io::stdout().flush().ok();
        }
        let elapsed = start.elapsed().as_secs_f64();
        let tps = if elapsed > 0.0 {
            tokens as f64 / elapsed
        } else {
            0.0
        };
        match format {
            OutputFormat::Text => {
                if !bench {
                    println!();
                }
                eprintln!(
                    "{}",
                    format!("[{tokens} tokens, {tps:.1} tok/s, {elapsed:.1}s]").dimmed()
                );
            }
            OutputFormat::Jsonl => {
                emit_jsonl_assistant(
                    &run_session_id,
                    0,
                    &request_id_text,
                    0,
                    &content,
                    reasoning,
                    &one_shot_history,
                    finish_reason,
                    usage.as_ref(),
                    tokens,
                    chunk_count,
                    &raw_response,
                    elapsed * 1000.0,
                );
            }
        }
        let shutdown_result = engine.shutdown().await;
        let shutdown_after = product_memory_enabled
            .then(|| memory_sampler.sample())
            .flatten();
        let shutdown_memory = process_memory_observation_between(
            memory_after
                .clone()
                .or_else(|| model_loaded_sample.clone())
                .or_else(|| backend_initialized_sample.clone())
                .or_else(|| process_start_sample.clone()),
            shutdown_after,
        );
        crate::observability_product::write_actual_run_observability(
            &product_observability,
            &crate::observability_product::ActualRunObservation {
                request_id: profile_request_id,
                duration_us: (elapsed * 1_000_000.0).max(0.0) as u64,
                sampling_params: plan.sampling_params.clone(),
                prompt_token_ids: plan.prompt_token_ids.clone(),
                prompt_token_count: plan.prompt_tokens,
                output_tokens: tokens,
                output_token_ids,
                chunk_count,
                finish_reason: finish_reason.map(finish_reason_str).map(str::to_string),
                prompt_chars,
                response_chars: raw_response.chars().count(),
                response_text: raw_response,
                execution_evidence,
                memory,
                memory_stages: actual_run_memory_stages(
                    product_memory_enabled,
                    process_start_memory.clone(),
                    backend_initialized_memory.clone(),
                    model_loaded_memory.clone(),
                    model_loaded_duration_us,
                    profile_run_done_memory.clone(),
                    cache_allocated_memory.clone(),
                    cache_allocated_status.clone(),
                    shutdown_memory,
                ),
            },
        )?;
        shutdown_result?;
        if format == OutputFormat::Jsonl {
            emit_jsonl_exit(&run_session_id, 0, "one_shot_complete");
        }
        return Ok(());
    }

    let mut history: Vec<RunHistoryMessage> = Vec::new();
    let mut history_epoch = 0usize;
    let mut turn = 0usize;
    let mut exit_reason: &str = "eof";

    // Print ready message
    let format = cmd.output_format;
    match format {
        OutputFormat::Text => {
            eprintln!();
            eprintln!("{}", "Ready. Type your message and press Enter.".green());
            eprintln!(
                "{}",
                "Use /clear to reset history; /bye or Ctrl+D to exit.".dimmed()
            );
            eprintln!();
        }
        OutputFormat::Jsonl => {
            emit_jsonl_ready(
                &run_session_id,
                &requested_model,
                &model_id,
                &device_label,
                model_chat_template.as_ref(),
            );
        }
    }

    // If stdin is not a TTY (piped input), don't print prompts and just consume lines.
    // This enables: `printf "hi\n/bye\n" | ferrum run ...` for automation/profiling.
    let stdin_handle = io::stdin();
    let stdin_is_tty = stdin_handle.is_terminal();
    let term = stdin_is_tty.then(Term::stdout);
    let mut stdin = stdin_handle.lock();

    loop {
        if stdin_is_tty {
            // Show prompt
            print!("{} ", ">>>".bright_green().bold());
            io::stdout().flush().unwrap();
        }

        // Read input
        let mut input = String::new();
        match read_repl_input_line(term.as_ref(), &mut stdin, &mut input) {
            Ok(0) => break, // EOF
            Ok(_) => {
                let input = input.trim();
                if input.is_empty() {
                    continue;
                }
                if input == "/bye" || input == "exit" || input == "quit" {
                    exit_reason = match input {
                        "/bye" => "bye",
                        "exit" => "exit",
                        "quit" => "quit",
                        _ => "command",
                    };
                    break;
                }
                if input == "/clear" {
                    let before = history_evidence(&history);
                    history.clear();
                    history_epoch += 1;
                    turn = 0;
                    match format {
                        OutputFormat::Text => {
                            eprintln!("{}", "History cleared.".dimmed());
                        }
                        OutputFormat::Jsonl => {
                            let record = serde_json::json!({
                                "schema_version": RUN_JSONL_SCHEMA_VERSION,
                                "event": "history_reset",
                                "session_id": run_session_id,
                                "history_epoch": history_epoch,
                                "turn": turn,
                                "history_before": before,
                                "history_after": history_evidence(&history),
                            });
                            emit_jsonl_record(&record);
                        }
                    }
                    continue;
                }

                let plan = build_run_prompt_plan(
                    &history,
                    input,
                    cmd.system.as_deref(),
                    &model_id,
                    model_chat_template.as_ref(),
                    &chat_template_options,
                    &cmd,
                    &run_budget,
                )?;
                maybe_warn_context_shift(&plan, format);
                let model_output_protocol = plan.sampling_params.model_output_protocol;
                let prompt_opened_thinking =
                    has_unclosed_model_reasoning_block(model_output_protocol, &plan.prompt);
                let observability_sampling_params =
                    product_memory_enabled.then(|| plan.sampling_params.clone());
                let prompt_token_ids = plan.prompt_token_ids;
                let prompt_token_count = plan.prompt_tokens;
                let prompt_chars = plan.prompt.chars().count();
                let metadata = run_request_metadata(
                    &plan.prompt,
                    &chat_template_options,
                    model_output_protocol,
                    model_chat_template.as_ref(),
                );
                let request_id = RequestId(Uuid::new_v4());
                let expected_request_id = request_id.clone();
                let request_id_text = request_id.to_string();
                if format == OutputFormat::Jsonl {
                    emit_jsonl_user(
                        &run_session_id,
                        history_epoch,
                        &request_id_text,
                        turn,
                        input,
                        &history,
                    );
                }
                // Create request
                let request = InferenceRequest {
                    id: request_id,
                    model_id: ferrum_types::ModelId(model_id.clone()),
                    prompt: plan.prompt,
                    sampling_params: plan.sampling_params,
                    stream: true,
                    priority: Priority::Normal,
                    client_id: None,
                    session_id: None,
                    created_at: Utc::now(),
                    api_request: None,
                    evidence_request: ferrum_types::InferenceEvidenceRequest {
                        capture_engine_token_timing: product_observability
                            .profile_detail
                            .captures_engine_token_timing(),
                        ..Default::default()
                    },
                    metadata,
                };

                let memory_before = product_memory_enabled
                    .then(|| memory_sampler.sample())
                    .flatten();
                let memory_stages = if product_memory_enabled && history_epoch == 0 && turn == 0 {
                    let mut stages = actual_run_memory_stages(
                        product_memory_enabled,
                        process_start_memory.clone(),
                        backend_initialized_memory.clone(),
                        model_loaded_memory.clone(),
                        model_loaded_duration_us,
                        profile_run_done_memory.clone(),
                        cache_allocated_memory.clone(),
                        cache_allocated_status.clone(),
                        None,
                    );
                    stages.retain(|stage| stage.stage != "shutdown");
                    stages
                } else {
                    Vec::new()
                };
                let start = std::time::Instant::now();
                let trace_tokens = crate::runtime_env::runtime_snapshot_value(
                    &runtime_config,
                    "FERRUM_RUN_TRACE_TOKENS",
                )
                .is_some();
                let generation_result = match format {
                    OutputFormat::Text => match engine.infer_stream(request).await {
                        Ok(stream) => {
                            collect_run_text_stream(
                                stream,
                                trace_tokens,
                                RunStreamOutput::new(model_output_protocol, prompt_opened_thinking),
                                turn,
                                &expected_request_id,
                                stdin_is_tty,
                                product_memory_enabled,
                            )
                            .await
                        }
                        Err(error) => Err(error),
                    },
                    OutputFormat::Jsonl => match engine.infer_stream(request).await {
                        Ok(stream) => {
                            collect_run_stream(
                                stream,
                                trace_tokens,
                                model_output_protocol == ModelOutputProtocol::HarmonyGptOss,
                                turn,
                                &run_session_id,
                                history_epoch,
                                &request_id_text,
                            )
                            .await
                        }
                        Err(error) => Err(error),
                    },
                };
                let generation = match generation_result {
                    Ok(generation) => generation,
                    Err(error) => {
                        let memory_after = product_memory_enabled
                            .then(|| memory_sampler.sample())
                            .flatten();
                        let memory =
                            process_memory_observation_between(memory_before, memory_after);
                        let elapsed = start.elapsed().as_secs_f64();
                        if product_memory_enabled {
                            if let Err(observability_error) =
                                crate::observability_product::write_actual_run_failure_observability(
                                    &product_observability,
                                    &crate::observability_product::ActualRunFailureObservation {
                                        request_id: request_id_text,
                                        duration_us: (elapsed * 1_000_000.0).max(0.0) as u64,
                                        sampling_params: observability_sampling_params.expect(
                                            "enabled run observability must capture sampling parameters",
                                        ),
                                        prompt_token_ids,
                                        prompt_token_count,
                                        prompt_chars,
                                        failure_kind: error
                                            .observability_failure_kind()
                                            .to_string(),
                                        error_kind: error.observability_error_kind().to_string(),
                                        error_message: error.to_string(),
                                        memory,
                                        memory_stages,
                                    },
                                )
                            {
                                eprintln!(
                                    "failed to write interactive run failure observability: {observability_error}"
                                );
                            }
                        }
                        return Err(error);
                    }
                };
                let memory_after = product_memory_enabled
                    .then(|| memory_sampler.sample())
                    .flatten();
                let memory = process_memory_observation_between(memory_before, memory_after);
                let CollectedRunGeneration {
                    request_id: response_request_id,
                    raw_text,
                    finish_reason,
                    usage,
                    token_count,
                    token_ids: output_token_ids,
                    chunk_count,
                    execution_evidence,
                } = generation;
                if response_request_id != request_id_text {
                    return Err(FerrumError::internal(format!(
                        "run response request id drift: expected {request_id_text}, got {response_request_id}"
                    )));
                }
                let raw_response = display_response_text(&raw_text);
                let parsed = parse_run_model_output(
                    model_output_protocol,
                    &raw_response,
                    prompt_opened_thinking,
                    finish_reason,
                )?;
                let clean_response = display_response_text(&parsed.content);
                let reasoning = (format == OutputFormat::Jsonl)
                    .then_some(parsed.reasoning.as_deref())
                    .flatten()
                    .map(|value| value.trim().to_string())
                    .filter(|value| !value.is_empty());

                if format == OutputFormat::Text
                    && model_output_protocol == ModelOutputProtocol::HarmonyGptOss
                {
                    print!("{clean_response}");
                    io::stdout().flush().ok();
                }

                let elapsed = start.elapsed();
                let elapsed_s = elapsed.as_secs_f64();
                let tps = if elapsed_s > 0.0 {
                    token_count as f64 / elapsed_s
                } else {
                    0.0
                };

                match format {
                    OutputFormat::Text => {
                        println!();
                        eprintln!(
                            "{}",
                            format!("[{token_count} tokens, {tps:.1} tok/s, {elapsed_s:.1}s]")
                                .dimmed()
                        );
                        eprintln!();
                    }
                    OutputFormat::Jsonl => {
                        emit_jsonl_assistant(
                            &run_session_id,
                            history_epoch,
                            &request_id_text,
                            turn,
                            &clean_response,
                            reasoning.as_deref(),
                            &history,
                            finish_reason,
                            usage.as_ref(),
                            token_count,
                            chunk_count,
                            &raw_response,
                            elapsed_s * 1000.0,
                        );
                    }
                }

                if product_memory_enabled {
                    crate::observability_product::write_actual_run_observability(
                        &product_observability,
                        &crate::observability_product::ActualRunObservation {
                            request_id: request_id_text.clone(),
                            duration_us: (elapsed_s * 1_000_000.0).max(0.0) as u64,
                            sampling_params: observability_sampling_params.expect(
                                "enabled run observability must capture sampling parameters",
                            ),
                            prompt_token_ids,
                            prompt_token_count,
                            output_tokens: token_count,
                            output_token_ids,
                            chunk_count,
                            finish_reason: finish_reason.map(finish_reason_str).map(str::to_string),
                            prompt_chars,
                            response_chars: raw_response.chars().count(),
                            response_text: raw_response.clone(),
                            execution_evidence,
                            memory,
                            memory_stages,
                        },
                    )?;
                }

                // In non-interactive mode, don't wait for terminal formatting/spacing.
                if !stdin_is_tty {
                    io::stdout().flush().ok();
                    io::stderr().flush().ok();
                }

                // Add to history
                history.push(RunHistoryMessage::new("user", input));
                if !raw_response.is_empty() {
                    history.push(RunHistoryMessage::assistant(
                        &raw_response,
                        model_output_protocol,
                        &parsed,
                    ));
                }

                // Limit history
                while history.len() > 10 {
                    history.remove(0);
                }
                turn += 1;
            }
            Err(e) if e.kind() == io::ErrorKind::Interrupted => {
                exit_reason = "interrupt";
                break;
            }
            Err(e) => {
                eprintln!("{} {}", "Error reading input:".red(), e);
                exit_reason = "read_error";
                break;
            }
        }
    }

    match format {
        OutputFormat::Text => {
            eprintln!("{}", "Goodbye!".bright_yellow());
        }
        OutputFormat::Jsonl => {
            emit_jsonl_exit(&run_session_id, history_epoch, exit_reason);
        }
    }
    engine.shutdown().await?;
    Ok(())
}

fn 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_run_memory_stages(
    enabled: bool,
    process_start_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    backend_initialized_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    model_loaded_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    model_loaded_duration_us: u64,
    profile_run_done_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    cache_allocated_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
    cache_allocated_status: Option<ferrum_types::EngineStatus>,
    shutdown_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
) -> Vec<crate::observability_product::ActualMemoryStageObservation> {
    if !enabled {
        return Vec::new();
    }
    let profile_run_done = crate::observability_product::ActualMemoryStageObservation::new(
        "actual_run_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_run_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_run_process_start",
            "process_start",
            None,
            process_start_memory,
        ),
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_run_backend_initialized",
            "backend_initialized",
            None,
            backend_initialized_memory,
        ),
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_run_model_loaded",
            "model_loaded",
            Some(model_loaded_duration_us),
            model_loaded_memory,
        ),
        profile_run_done,
        cache_allocated,
        crate::observability_product::ActualMemoryStageObservation::new(
            "actual_run_shutdown",
            "shutdown",
            None,
            shutdown_memory,
        ),
    ]
}

#[cfg(unix)]
fn read_repl_input_line<R: BufRead + AsRawFd>(
    term: Option<&Term>,
    stdin: &mut R,
    input: &mut String,
) -> io::Result<usize> {
    if let Some(term) = term {
        let Some(line) = read_tty_input_line(term, stdin)? else {
            return Ok(0);
        };
        input.push_str(&line);
        return Ok(input.len().max(1));
    }
    stdin.read_line(input)
}

#[cfg(not(unix))]
fn read_repl_input_line<R: BufRead>(
    term: Option<&Term>,
    stdin: &mut R,
    input: &mut String,
) -> io::Result<usize> {
    if let Some(term) = term {
        let Some(line) = read_tty_input_line(term)? else {
            return Ok(0);
        };
        input.push_str(&line);
        return Ok(input.len().max(1));
    }
    stdin.read_line(input)
}

#[cfg(unix)]
struct RawModeGuard {
    fd: RawFd,
    original: libc::termios,
}

#[cfg(unix)]
impl RawModeGuard {
    fn new(fd: RawFd) -> io::Result<Self> {
        let mut termios = mem::MaybeUninit::uninit();
        if unsafe { libc::tcgetattr(fd, termios.as_mut_ptr()) } != 0 {
            return Err(io::Error::last_os_error());
        }
        let original = unsafe { termios.assume_init() };
        let mut raw = original;
        unsafe { libc::cfmakeraw(&mut raw) };
        raw.c_oflag = original.c_oflag;
        if unsafe { libc::tcsetattr(fd, libc::TCSADRAIN, &raw) } != 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(Self { fd, original })
    }
}

#[cfg(unix)]
impl Drop for RawModeGuard {
    fn drop(&mut self) {
        unsafe {
            libc::tcsetattr(self.fd, libc::TCSADRAIN, &self.original);
        }
    }
}

#[cfg(unix)]
fn read_tty_input_line<R: AsRawFd>(term: &Term, stdin: &mut R) -> io::Result<Option<String>> {
    let _raw_mode = RawModeGuard::new(stdin.as_raw_fd())?;
    let mut chars: Vec<char> = Vec::new();
    loop {
        match term.read_key_raw()? {
            Key::Backspace => {
                if let Some(ch) = chars.pop() {
                    let width = measure_text_width(&ch.to_string());
                    if width > 0 {
                        term.clear_chars(width)?;
                    }
                    term.flush()?;
                }
            }
            Key::Char('\u{4}') => {
                term.write_str("\n")?;
                term.flush()?;
                if chars.is_empty() {
                    return Ok(None);
                }
                break;
            }
            Key::CtrlC | Key::Char('\u{3}') => {
                term.write_str("^C\n")?;
                term.flush()?;
                return Err(io::Error::new(io::ErrorKind::Interrupted, "interrupted"));
            }
            Key::Enter => {
                term.write_str("\n")?;
                term.flush()?;
                break;
            }
            Key::Char(ch) if !ch.is_ascii_control() => {
                chars.push(ch);
                term.write_str(&ch.to_string())?;
                term.flush()?;
            }
            _ => {}
        }
    }
    Ok(Some(chars.into_iter().collect()))
}

#[cfg(not(unix))]
fn read_tty_input_line(term: &Term) -> io::Result<Option<String>> {
    let mut chars: Vec<char> = Vec::new();
    loop {
        match term.read_key()? {
            Key::Backspace => {
                if let Some(ch) = chars.pop() {
                    let width = measure_text_width(&ch.to_string());
                    if width > 0 {
                        term.clear_chars(width)?;
                    }
                    term.flush()?;
                }
            }
            Key::Char('\u{4}') => {
                term.write_str("\n")?;
                term.flush()?;
                if chars.is_empty() {
                    return Ok(None);
                }
                break;
            }
            Key::CtrlC | Key::Char('\u{3}') => {
                term.write_str("^C\n")?;
                term.flush()?;
                return Err(io::Error::new(io::ErrorKind::Interrupted, "interrupted"));
            }
            Key::Enter => {
                term.write_str("\n")?;
                term.flush()?;
                break;
            }
            Key::Char(ch) if !ch.is_ascii_control() => {
                chars.push(ch);
                term.write_str(&ch.to_string())?;
                term.flush()?;
            }
            _ => {}
        }
    }
    Ok(Some(chars.into_iter().collect()))
}

fn start_first_token_indicator(enabled: bool) -> Option<ProgressBar> {
    if !enabled {
        return None;
    }
    let progress = ProgressBar::new_spinner();
    let style = ProgressStyle::with_template("{spinner} Working ({elapsed})")
        .unwrap_or_else(|_| ProgressStyle::default_spinner());
    progress.set_style(style);
    progress.enable_steady_tick(std::time::Duration::from_millis(120));
    progress.tick();
    Some(progress)
}

fn clear_first_token_indicator(progress: &mut Option<ProgressBar>) {
    if let Some(progress) = progress.take() {
        progress.finish_and_clear();
    }
}

fn runtime_config_bool(snapshot: &RuntimeConfigSnapshot, key: &str) -> Option<bool> {
    crate::runtime_env::runtime_snapshot_value(snapshot, key).map(|value| {
        matches!(
            value.trim().to_ascii_lowercase().as_str(),
            "" | "1" | "true" | "yes" | "on"
        )
    })
}

fn run_autosize_for_device(
    device: &ferrum_types::Device,
    gpu_memory_utilization: f32,
) -> Option<(crate::gpu_mem_autosize::AutoSizeProfile, f32)> {
    match device {
        ferrum_types::Device::CPU => None,
        _ => Some((
            crate::gpu_mem_autosize::AutoSizeProfile::Chat,
            gpu_memory_utilization,
        )),
    }
}

fn build_sampling_params(cmd: &RunCommand) -> SamplingParams {
    let greedy = cmd.temperature <= 0.0;
    let mut stop_sequences = vec![
        "<|im_end|>".to_string(),
        "</s>".to_string(),
        "<|endoftext|>".to_string(),
    ];
    stop_sequences.extend(cmd.stop.iter().filter(|stop| !stop.is_empty()).cloned());
    SamplingParams {
        max_tokens: cmd.max_tokens as usize,
        temperature: cmd.temperature,
        top_p: if greedy { 1.0 } else { cmd.top_p },
        top_k: if greedy || cmd.top_k == 0 {
            None
        } else {
            Some(cmd.top_k)
        },
        min_p: (cmd.min_p != 0.0).then_some(cmd.min_p),
        presence_penalty: cmd.presence_penalty,
        repetition_penalty: cmd.repeat_penalty,
        stop_sequences,
        seed: cmd.seed,
        ..Default::default()
    }
}

fn validate_teacher_forced_checkpoint_run(
    cmd: &RunCommand,
    capture: Option<&ferrum_types::VNextCheckpointCaptureConfig>,
) -> Result<()> {
    let Some(teacher) = capture.and_then(|capture| capture.teacher_forcing.as_ref()) else {
        return Ok(());
    };
    if cmd.prompt.is_none() {
        return Err(ferrum_types::FerrumError::config(
            "teacher-forced checkpoint capture requires one-shot --prompt",
        ));
    }
    if cmd.max_tokens as usize != teacher.token_count() {
        return Err(ferrum_types::FerrumError::config(format!(
            "teacher-forced checkpoint requires --max-tokens {}, got {}",
            teacher.token_count(),
            cmd.max_tokens
        )));
    }
    if cmd.max_num_seqs != Some(1) {
        return Err(ferrum_types::FerrumError::config(
            "teacher-forced checkpoint capture requires --max-num-seqs 1",
        ));
    }
    if cmd.temperature != 0.0
        || cmd.top_k != 0
        || cmd.top_p != 1.0
        || cmd.min_p != 0.0
        || cmd.presence_penalty != 0.0
        || cmd.repeat_penalty != 1.0
        || !cmd.stop.is_empty()
    {
        return Err(ferrum_types::FerrumError::config(
            "teacher-forced checkpoint capture requires unpenalized greedy settings: \
             --temperature 0 --top-k 0 --top-p 1 --min-p 0 \
             --presence-penalty 0 --repeat-penalty 1 and no --stop",
        ));
    }
    Ok(())
}

fn sampling_params_for_prompt(mut sampling_params: SamplingParams, prompt: &str) -> SamplingParams {
    if has_unclosed_model_reasoning_block(sampling_params.model_output_protocol, prompt) {
        let (_, closing) = model_reasoning_markers(sampling_params.model_output_protocol)
            .expect("an open reasoning block has declared markers");
        sampling_params.response_completion_boundary =
            ResponseCompletionBoundary::AfterDelimiterAndPayload {
                delimiter: closing.to_string(),
                alternate_envelope: None,
            };
    }
    sampling_params
}

fn build_chat_template_options(
    cmd: &RunCommand,
    model_template: Option<&ModelChatTemplate>,
) -> ChatTemplateOptions {
    let mut options = ChatTemplateOptions::default_for_template(model_template);
    if cmd.enable_thinking {
        options.enable_thinking = Some(true);
    } else if cmd.disable_thinking {
        options.enable_thinking = Some(false);
    }
    options
}

fn build_run_prompt_plan(
    history: &[RunHistoryMessage],
    user_input: &str,
    system: Option<&str>,
    model_id: &str,
    model_template: Option<&ModelChatTemplate>,
    chat_template_options: &ChatTemplateOptions,
    cmd: &RunCommand,
    budget: &RunBudget,
) -> Result<RunPromptPlan> {
    let mut base_sampling = build_sampling_params(cmd);
    base_sampling.model_output_protocol = model_template
        .map(|template| template.output_protocol)
        .unwrap_or(ModelOutputProtocol::Text);

    if cmd.no_context_shift {
        let prompt = build_chat_prompt(
            history,
            user_input,
            system,
            model_id,
            model_template,
            chat_template_options,
        )?;
        let prompt_tokenization = budget.prompt_tokenization(&prompt);
        let prompt_tokens = prompt_tokenization.token_count;
        if !fits_kv_budget(&base_sampling, prompt_tokens, budget.kv_capacity) {
            return Err(FerrumError::invalid_request(format!(
                "This model context is limited to {} tokens, but this turn needs {} input tokens + {} output tokens. Reduce --max-tokens, use /clear, or shorten the prompt.",
                budget.kv_capacity.unwrap_or(0),
                prompt_tokens.unwrap_or(0),
                base_sampling.max_tokens,
            )));
        }

        let sampling_params = sampling_params_for_prompt(base_sampling, &prompt);
        return Ok(RunPromptPlan {
            prompt,
            sampling_params,
            prompt_token_ids: prompt_tokenization.token_ids,
            prompt_tokens,
            kv_capacity: budget.kv_capacity,
            dropped_history_messages: 0,
            dropped_history_turns: 0,
            max_tokens_clamped_from: None,
        });
    }

    let mut history_start = 0usize;
    loop {
        let prompt = build_chat_prompt(
            &history[history_start..],
            user_input,
            system,
            model_id,
            model_template,
            chat_template_options,
        )?;
        let prompt_tokenization = budget.prompt_tokenization(&prompt);
        let prompt_tokens = prompt_tokenization.token_count;

        let Some(kv_capacity) = budget.kv_capacity else {
            let sampling_params = sampling_params_for_prompt(base_sampling, &prompt);
            return Ok(RunPromptPlan {
                prompt,
                sampling_params,
                prompt_token_ids: prompt_tokenization.token_ids,
                prompt_tokens,
                kv_capacity: None,
                dropped_history_messages: history_start,
                dropped_history_turns: count_user_turns(&history[..history_start]),
                max_tokens_clamped_from: None,
            });
        };
        let Some(prompt_tokens) = prompt_tokens else {
            let sampling_params = sampling_params_for_prompt(base_sampling, &prompt);
            return Ok(RunPromptPlan {
                prompt,
                sampling_params,
                prompt_token_ids: prompt_tokenization.token_ids,
                prompt_tokens: None,
                kv_capacity: Some(kv_capacity),
                dropped_history_messages: history_start,
                dropped_history_turns: count_user_turns(&history[..history_start]),
                max_tokens_clamped_from: None,
            });
        };

        if prompt_tokens < kv_capacity {
            let remaining = kv_capacity - prompt_tokens;
            let mut sampling_params = base_sampling.clone();
            let max_tokens_clamped_from = if sampling_params.max_tokens > remaining {
                let old = sampling_params.max_tokens;
                sampling_params.max_tokens = remaining;
                Some(old)
            } else {
                None
            };
            let sampling_params = sampling_params_for_prompt(sampling_params, &prompt);
            return Ok(RunPromptPlan {
                prompt,
                sampling_params,
                prompt_token_ids: prompt_tokenization.token_ids,
                prompt_tokens: Some(prompt_tokens),
                kv_capacity: Some(kv_capacity),
                dropped_history_messages: history_start,
                dropped_history_turns: count_user_turns(&history[..history_start]),
                max_tokens_clamped_from,
            });
        }

        if history_start >= history.len() {
            return Err(FerrumError::invalid_request(format!(
                "This model context is limited to {kv_capacity} tokens, but the current turn needs {prompt_tokens} input tokens before generation. Use a shorter prompt or increase KV capacity.",
            )));
        }
        history_start = next_context_shift_history_start(history, history_start);
    }
}

fn next_context_shift_history_start(history: &[RunHistoryMessage], start: usize) -> usize {
    if start + 1 < history.len()
        && history[start].prompt.role == "user"
        && history[start + 1].prompt.role == "assistant"
    {
        start + 2
    } else {
        start + 1
    }
}

fn count_user_turns(history: &[RunHistoryMessage]) -> usize {
    history
        .iter()
        .filter(|message| message.prompt.role == "user")
        .count()
}

fn maybe_warn_context_shift(plan: &RunPromptPlan, format: OutputFormat) {
    if format != OutputFormat::Text {
        return;
    }
    if plan.dropped_history_messages == 0 && plan.max_tokens_clamped_from.is_none() {
        return;
    }

    let prompt_tokens = plan
        .prompt_tokens
        .map(|value| value.to_string())
        .unwrap_or_else(|| "?".to_string());
    let kv_capacity = plan
        .kv_capacity
        .map(|value| value.to_string())
        .unwrap_or_else(|| "?".to_string());
    let mut parts = Vec::new();
    if plan.dropped_history_messages > 0 {
        parts.push(format!(
            "dropped {} old message(s) / {} turn(s)",
            plan.dropped_history_messages, plan.dropped_history_turns
        ));
    }
    if let Some(old) = plan.max_tokens_clamped_from {
        parts.push(format!(
            "max_tokens {} -> {}",
            old, plan.sampling_params.max_tokens
        ));
    }

    eprintln!(
        "{}",
        format!(
            "[context-shift] {} (prompt_tokens={}, kv_capacity={})",
            parts.join("; "),
            prompt_tokens,
            kv_capacity
        )
        .dimmed()
    );
}

fn fits_kv_budget(
    base: &SamplingParams,
    prompt_tokens: Option<usize>,
    kv_capacity: Option<usize>,
) -> bool {
    let (Some(prompt_tokens), Some(kv_capacity)) = (prompt_tokens, kv_capacity) else {
        return true;
    };
    prompt_tokens < kv_capacity && prompt_tokens + base.max_tokens <= kv_capacity
}

fn discover_run_tokenizer_path(source_path: &Path) -> Option<PathBuf> {
    if source_path.is_file()
        && source_path
            .extension()
            .map(|e| e.eq_ignore_ascii_case("gguf"))
            .unwrap_or(false)
    {
        return ferrum_models::gguf_engine_loader::auto_discover_tokenizer_path(source_path);
    }
    let tokenizer = source_path.join("tokenizer.json");
    tokenizer.is_file().then_some(tokenizer)
}

pub fn select_device(backend: &str) -> Result<ferrum_types::Device> {
    crate::backend_selection::select_device(backend)
}

fn build_chat_prompt(
    history: &[RunHistoryMessage],
    user_input: &str,
    system: Option<&str>,
    model_id: &str,
    model_template: Option<&ModelChatTemplate>,
    chat_template_options: &ChatTemplateOptions,
) -> Result<String> {
    let mut messages = Vec::new();
    if let Some(sys) = system {
        messages.push(PromptMessage::new("system", sys));
    }
    messages.extend(history.iter().map(|message| message.prompt.clone()));
    messages.push(PromptMessage::new("user", user_input));
    ferrum_server::chat_template::render_prompt_messages_with_options(
        &messages,
        model_id,
        model_template,
        chat_template_options,
    )
}

async fn load_run_model_definition(
    source: &ResolvedModelSource,
    product_sources: Option<&ferrum_models::vnext::ProductionModelSourceBundle>,
) -> Result<Option<ferrum_models::ModelDefinition>> {
    if let Some(sources) = product_sources {
        let mut config_manager = ferrum_models::ConfigManager::new();
        return config_manager
            .load_from_bytes(sources.config_json())
            .map(Some);
    }
    if source.format != ModelFormat::SafeTensors {
        return Ok(None);
    }
    let mut config_manager = ferrum_models::ConfigManager::new();
    Ok(Some(
        config_manager.load_from_path(&source.local_path).await?,
    ))
}

fn run_effective_runtime_config(
    runtime_config: &RuntimeConfigSnapshot,
    cli_runtime_entries: &[RuntimeConfigEntry],
) -> RuntimeConfigSnapshot {
    let mut snapshot = runtime_config.clone();
    for entry in cli_runtime_entries {
        snapshot.upsert_entry(entry.clone());
    }
    snapshot
}

fn run_base_runtime_config(
    config: &CliConfig,
    env_snapshot: RuntimeConfigSnapshot,
) -> RuntimeConfigSnapshot {
    let mut config_entries = run_product_default_runtime_entries();
    config_entries.extend(config.runtime.runtime_config_entries());
    crate::commands::serve::merge_runtime_config_sources(config_entries, env_snapshot, Vec::new())
}

fn run_product_default_runtime_entries() -> Vec<RuntimeConfigEntry> {
    vec![
        RuntimeConfigEntry::new("FERRUM_PAGED_MAX_SEQS", "1", RuntimeConfigSource::Default),
        RuntimeConfigEntry::new(
            "FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS",
            "1",
            RuntimeConfigSource::Default,
        ),
    ]
}

fn run_startup_cli_runtime_entries(
    cmd: &RunCommand,
    gpu_selection: Option<&crate::gpu_devices::GpuDeviceSelection>,
) -> Vec<RuntimeConfigEntry> {
    let mut entries = Vec::new();
    entries.push(RuntimeConfigEntry::new(
        "FERRUM_PROFILE_DETAIL",
        cmd.profile_detail.as_str(),
        RuntimeConfigSource::Cli,
    ));
    crate::runtime_env::push_cli_runtime_entry(
        &mut entries,
        "FERRUM_KV_DTYPE",
        cmd.kv_dtype.as_deref(),
    );
    crate::runtime_env::push_cli_runtime_usize(&mut entries, "FERRUM_KV_CAPACITY", cmd.kv_capacity);
    crate::runtime_env::push_cli_runtime_usize(
        &mut entries,
        "FERRUM_KV_MAX_BLOCKS",
        cmd.kv_max_blocks,
    );
    crate::runtime_env::push_cli_runtime_usize(
        &mut entries,
        "FERRUM_MAX_MODEL_LEN",
        cmd.max_model_len,
    );
    crate::runtime_env::push_cli_runtime_usize(
        &mut entries,
        "FERRUM_PAGED_MAX_SEQS",
        cmd.max_num_seqs,
    );
    crate::runtime_env::push_cli_runtime_usize(
        &mut entries,
        "FERRUM_MAX_BATCHED_TOKENS",
        cmd.max_num_batched_tokens,
    );
    crate::runtime_env::push_cli_runtime_entry(
        &mut entries,
        "FERRUM_SEQUENCE_FIT_POLICY",
        cmd.sequence_fit_policy
            .map(crate::commands::SequenceFitPolicyArg::as_runtime_value),
    );
    crate::runtime_env::push_cli_runtime_entry(
        &mut entries,
        "FERRUM_VNEXT_DIAGNOSTIC_FAULT",
        cmd.vnext_diagnostic_fault
            .map(crate::commands::VNextDiagnosticFaultArg::as_runtime_value),
    );
    crate::runtime_env::push_cli_runtime_usize(
        &mut entries,
        "FERRUM_RUNTIME_MEMORY_BUDGET_BYTES",
        cmd.runtime_memory_budget_bytes
            .map(std::num::NonZeroUsize::get),
    );
    if let Some(wait) = cmd.prefix_rendezvous_max_wait_ms {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PREFIX_RENDEZVOUS_MAX_WAIT_MS",
            wait.to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = bool_cli_override(cmd.batched_graph, cmd.disable_batched_graph) {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_BATCHED_GRAPH",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = bool_cli_override(cmd.reusable_execution, cmd.disable_reusable_execution)
    {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_REUSABLE_EXECUTION",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = bool_cli_override(cmd.unified_graph, cmd.disable_unified_graph) {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = bool_cli_override(
        cmd.unified_graph_layers_only,
        cmd.disable_unified_graph_layers_only,
    ) {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH_LAYERS_ONLY",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(enabled) = bool_cli_override(
        cmd.unified_graph_lm_head_eager,
        cmd.disable_unified_graph_lm_head_eager,
    ) {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_UNIFIED_GRAPH_LM_HEAD_EAGER",
            if enabled { "1" } else { "0" },
            RuntimeConfigSource::Cli,
        ));
    }
    crate::layer_split_pipeline::push_cli_runtime_entry(
        &mut entries,
        cmd.layer_split_pipeline_mode,
    );
    if let Some(path) = &cmd.profile_jsonl {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PROFILE_JSONL",
            path.to_string_lossy().to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(path) = &cmd.scheduler_trace_jsonl {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_SCHEDULER_TRACE_JSONL",
            path.to_string_lossy().to_string(),
            RuntimeConfigSource::Cli,
        ));
    }
    if cmd.profile_jsonl.is_some() || cmd.scheduler_trace_jsonl.is_some() {
        entries.push(RuntimeConfigEntry::new(
            "FERRUM_PROFILE_ENTRYPOINT",
            "run",
            RuntimeConfigSource::Cli,
        ));
    }
    if let Some(selection) = gpu_selection {
        entries.extend(selection.runtime_config_entries());
    }
    entries
}

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

fn run_startup_auto_config(
    hardware: ferrum_types::HardwareCapabilities,
    typed_model_capabilities: Option<ModelCapabilities>,
    execution_resource_authority: ferrum_types::ExecutionResourceAuthority,
    model_definition: Option<&ferrum_models::ModelDefinition>,
    model_weight_bytes: Option<u64>,
    runtime_config: RuntimeConfigSnapshot,
) -> Result<ResolvedFerrumConfig> {
    let model = typed_model_capabilities
        .or_else(|| model_definition.map(|definition| {
            crate::commands::serve::model_capabilities_from_definition_with_weight_bytes_for_hardware(
                definition,
                model_weight_bytes,
                &hardware,
            )
        }))
        .unwrap_or_else(ModelCapabilities::unknown);
    let mut workload = WorkloadProfile::serving_default();
    workload.serving_mode = "interactive".into();
    workload.priority = ferrum_types::WorkloadPriority::Latency;
    crate::startup::resolve_config(
        runtime_config,
        model,
        hardware,
        workload,
        execution_resource_authority,
    )
}

/// Apply the resolved `--kv-dtype` / runtime-config override to an engine
/// config, validating syntax early. The registered numerical profiles and
/// actual backend providers validate the requested FP16/INT8 combination.
pub fn apply_kv_dtype_override(
    engine_config: &mut ferrum_types::EngineConfig,
    raw: Option<&str>,
) -> ferrum_types::Result<()> {
    use ferrum_types::KvCacheDtype;
    let Some(raw) = raw else {
        // No override → keep config default (FP16).
        return Ok(());
    };
    let parsed = KvCacheDtype::parse(raw).ok_or_else(|| {
        ferrum_types::FerrumError::config(format!(
            "Unknown --kv-dtype value '{}'. Accepts: fp16, bf16, int8, fp8.",
            raw
        ))
    })?;
    match parsed {
        KvCacheDtype::Fp16 => {
            engine_config.kv_cache.dtype = KvCacheDtype::Fp16;
            Ok(())
        }
        KvCacheDtype::Int8 => {
            // This is a request, not proof that a backend supports the format.
            // vNext must bind it into its numerical profile before compilation;
            // the explicit legacy registry retains its own capability checks.
            engine_config.kv_cache.dtype = KvCacheDtype::Int8;
            Ok(())
        }
        KvCacheDtype::Fp8 => Err(ferrum_types::FerrumError::unsupported(
            "FP8 KV cache: kernels not yet implemented. Tracked as PR D.",
        )),
        KvCacheDtype::Bf16 => Err(ferrum_types::FerrumError::unsupported(
            "BF16 KV cache: marker only, no backend impl ships yet.",
        )),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ferrum_types::{RuntimeConfigSource, SequenceFitPolicy, TokenId};

    fn default_params(max_tokens: usize) -> SamplingParams {
        SamplingParams {
            max_tokens,
            ..SamplingParams::default()
        }
    }

    fn test_run_cmd() -> RunCommand {
        RunCommand {
            model: Some("tinyllama".to_string()),
            product_sources: crate::source_resolver::ProductSourceArgs::default(),
            system: None,
            max_tokens: 4096,
            stop: Vec::new(),
            no_context_shift: false,
            enable_thinking: false,
            disable_thinking: false,
            temperature: 0.0,
            backend: "auto".to_string(),
            numerical_profile: None,
            gpu_devices: None,
            layer_split_pipeline_mode: None,
            prompt: None,
            tokenizer: None,
            bench_mode: false,
            top_k: 50,
            top_p: 0.95,
            min_p: 0.0,
            presence_penalty: 0.0,
            repeat_penalty: 1.0,
            repeat_last_n: 64,
            seed: None,
            gpu_memory_utilization: 0.9,
            runtime_memory_budget_bytes: None,
            max_model_len: None,
            max_num_seqs: None,
            max_num_batched_tokens: None,
            sequence_fit_policy: None,
            prefix_rendezvous_max_wait_ms: None,
            batched_graph: false,
            disable_batched_graph: false,
            reusable_execution: false,
            disable_reusable_execution: false,
            unified_graph: false,
            disable_unified_graph: false,
            unified_graph_layers_only: false,
            disable_unified_graph_layers_only: false,
            unified_graph_lm_head_eager: false,
            disable_unified_graph_lm_head_eager: false,
            kv_dtype: None,
            kv_capacity: None,
            kv_max_blocks: None,
            effective_config_json: None,
            decision_trace_jsonl: None,
            observability_vertical_slice_out: None,
            vnext_checkpoint: Default::default(),
            profile_jsonl: None,
            profile_detail: crate::observability_product::ProfileDetailArg::Off,
            vnext_diagnostic_fault: None,
            memory_profile_jsonl: None,
            scheduler_trace_jsonl: None,
            request_dump_dir: None,
            profile_sample_rate: crate::observability_product::default_profile_sample_rate(),
            output_format: OutputFormat::Text,
        }
    }

    #[test]
    fn teacher_forced_checkpoint_is_one_shot_single_sequence_and_deterministic() {
        let teacher =
            ferrum_types::VNextTeacherForcingConfig::new(vec![TokenId::new(7), TokenId::new(11)])
                .unwrap();
        let capture = ferrum_types::VNextCheckpointCaptureConfig {
            output_dir: PathBuf::from("capture"),
            value_ids: Vec::new(),
            maximum_prefill_waves: 1,
            maximum_decode_waves: 1,
            capture_product_output: true,
            teacher_forcing: Some(teacher),
        };
        let mut cmd = test_run_cmd();
        cmd.prompt = Some("hello".to_owned());
        cmd.max_tokens = 2;
        cmd.max_num_seqs = Some(1);
        cmd.top_k = 0;
        cmd.top_p = 1.0;
        assert!(validate_teacher_forced_checkpoint_run(&cmd, Some(&capture)).is_ok());

        cmd.prompt = None;
        assert!(validate_teacher_forced_checkpoint_run(&cmd, Some(&capture)).is_err());
        cmd.prompt = Some("hello".to_owned());
        cmd.max_num_seqs = Some(2);
        assert!(validate_teacher_forced_checkpoint_run(&cmd, Some(&capture)).is_err());
        cmd.max_num_seqs = Some(1);
        cmd.temperature = 0.5;
        assert!(validate_teacher_forced_checkpoint_run(&cmd, Some(&capture)).is_err());
    }

    fn whitespace_budget(kv_capacity: usize) -> RunBudget {
        RunBudget {
            tokenizer: None,
            kv_capacity: Some(kv_capacity),
            prompt_token_id_mapper: None,
            prompt_token_counter: Some(|prompt| prompt.split_whitespace().count()),
        }
    }

    fn mapped_token_budget(kv_capacity: usize) -> RunBudget {
        RunBudget {
            tokenizer: None,
            kv_capacity: Some(kv_capacity),
            prompt_token_id_mapper: Some(|prompt| {
                prompt
                    .split_whitespace()
                    .enumerate()
                    .map(|(index, _)| (index + 1) as u32)
                    .collect()
            }),
            prompt_token_counter: None,
        }
    }

    fn default_template_options() -> ChatTemplateOptions {
        ChatTemplateOptions::default()
    }

    #[test]
    fn run_effective_runtime_config_records_cli_kv_dtype() {
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let mut cmd = test_run_cmd();
        cmd.kv_dtype = Some("int8".to_string());
        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = effective
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_KV_DTYPE")
            .expect("missing kv dtype entry");
        assert_eq!(entry.effective_value, "int8");
        assert_eq!(entry.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn run_prefix_rendezvous_has_explicit_cli_authority_and_no_default_wait() {
        use clap::Parser;
        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            run: RunCommand,
        }
        let parsed = TestCli::try_parse_from([
            "ferrum",
            "test-model",
            "--prefix-rendezvous-max-wait-ms",
            "123",
        ])
        .unwrap();
        let snapshot =
            RuntimeConfigSnapshot::from_entries(run_startup_cli_runtime_entries(&parsed.run, None));
        let mut config = ferrum_types::EngineConfig::default();
        assert!(config.scheduler.prefix_rendezvous_max_wait_ms.is_none());
        config.apply_runtime_config_snapshot(&snapshot).unwrap();
        assert_eq!(
            config
                .scheduler
                .prefix_rendezvous_max_wait_ms
                .map(std::num::NonZeroU64::get),
            Some(123)
        );
        assert_eq!(
            snapshot
                .entries
                .iter()
                .find(|entry| entry.key == "FERRUM_PREFIX_RENDEZVOUS_MAX_WAIT_MS")
                .unwrap()
                .source,
            RuntimeConfigSource::Cli
        );
        assert!(TestCli::try_parse_from([
            "ferrum",
            "test-model",
            "--prefix-rendezvous-max-wait-ms",
            "0"
        ])
        .is_err());
    }

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

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            run: RunCommand,
        }

        let parsed = TestCli::parse_from([
            "ferrum",
            "Qwen/Qwen3.5-4B",
            "--vnext-diagnostic-fault",
            "prefill-resource-after-submit-once",
        ]);
        let entries = run_startup_cli_runtime_entries(&parsed.run, None);
        let entry = entries
            .iter()
            .find(|entry| entry.key == "FERRUM_VNEXT_DIAGNOSTIC_FAULT")
            .expect("diagnostic fault CLI entry");

        assert_eq!(
            parsed.run.vnext_diagnostic_fault,
            Some(crate::commands::VNextDiagnosticFaultArg::PrefillResourceAfterSubmitOnce)
        );
        assert_eq!(entry.effective_value, "prefill-resource-after-submit-once");
        assert_eq!(entry.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn run_effective_runtime_config_records_memory_budget() {
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let mut cmd = test_run_cmd();
        cmd.runtime_memory_budget_bytes = std::num::NonZeroUsize::new(12_345);

        let effective =
            run_effective_runtime_config(&snapshot, &run_startup_cli_runtime_entries(&cmd, None));
        let entry = effective
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_RUNTIME_MEMORY_BUDGET_BYTES")
            .expect("missing runtime memory budget entry");

        assert_eq!(entry.effective_value, "12345");
        assert_eq!(entry.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn run_effective_runtime_config_records_observability_paths() {
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let mut cmd = test_run_cmd();
        cmd.profile_jsonl = Some(PathBuf::from("/tmp/run-profile.jsonl"));
        cmd.scheduler_trace_jsonl = Some(PathBuf::from("/tmp/run-scheduler.jsonl"));

        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = |key: &str| {
            effective
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key} entry"))
        };

        assert_eq!(
            entry("FERRUM_PROFILE_JSONL").effective_value,
            "/tmp/run-profile.jsonl"
        );
        assert_eq!(
            entry("FERRUM_SCHEDULER_TRACE_JSONL").effective_value,
            "/tmp/run-scheduler.jsonl"
        );
        assert_eq!(entry("FERRUM_PROFILE_ENTRYPOINT").effective_value, "run");
        assert!(entry("FERRUM_PROFILE_ENTRYPOINT")
            .affects
            .contains(&ferrum_types::RuntimeConfigEffect::Diagnostics));
    }

    #[test]
    fn run_prefix_state_cache_config_reaches_typed_engine_config() {
        for enabled in [true, false] {
            let config: CliConfig =
                toml::from_str(&format!("[runtime]\nprefix_cache = {enabled}\n")).unwrap();
            let base = run_base_runtime_config(&config, RuntimeConfigSnapshot::default());
            let effective = run_effective_runtime_config(
                &base,
                &run_startup_cli_runtime_entries(&test_run_cmd(), None),
            );
            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 run_full_profile_detail_reaches_typed_engine_config() {
        let mut cmd = test_run_cmd();
        cmd.profile_detail = crate::observability_product::ProfileDetailArg::Full;
        let effective = run_effective_runtime_config(
            &RuntimeConfigSnapshot::from_entries(Vec::new()),
            &run_startup_cli_runtime_entries(&cmd, None),
        );
        let mut engine = ferrum_types::EngineConfig::default();

        engine
            .apply_runtime_config_snapshot(&effective)
            .expect("full profile detail should apply");

        assert_eq!(
            engine.runtime.profile_detail,
            ferrum_types::ObservabilityProfileDetail::Full
        );
    }

    #[test]
    fn run_latency_profile_detail_reaches_typed_engine_config() {
        let mut cmd = test_run_cmd();
        cmd.profile_detail = crate::observability_product::ProfileDetailArg::Latency;
        let effective = run_effective_runtime_config(
            &RuntimeConfigSnapshot::from_entries(Vec::new()),
            &run_startup_cli_runtime_entries(&cmd, None),
        );
        let mut engine = ferrum_types::EngineConfig::default();

        engine
            .apply_runtime_config_snapshot(&effective)
            .expect("latency profile detail should apply");

        assert_eq!(
            engine.runtime.profile_detail,
            ferrum_types::ObservabilityProfileDetail::Latency
        );
    }

    #[test]
    fn run_replay_profile_detail_reaches_typed_engine_config() {
        let mut cmd = test_run_cmd();
        cmd.profile_detail = crate::observability_product::ProfileDetailArg::Replay;
        let effective = run_effective_runtime_config(
            &RuntimeConfigSnapshot::from_entries(Vec::new()),
            &run_startup_cli_runtime_entries(&cmd, None),
        );
        let mut engine = ferrum_types::EngineConfig::default();

        engine
            .apply_runtime_config_snapshot(&effective)
            .expect("replay profile detail should apply");

        assert_eq!(
            engine.runtime.profile_detail,
            ferrum_types::ObservabilityProfileDetail::Replay
        );
    }

    #[test]
    fn run_verify_profile_detail_reaches_typed_engine_config() {
        let mut cmd = test_run_cmd();
        cmd.profile_detail = crate::observability_product::ProfileDetailArg::Verify;
        let effective = run_effective_runtime_config(
            &RuntimeConfigSnapshot::from_entries(Vec::new()),
            &run_startup_cli_runtime_entries(&cmd, None),
        );
        let mut engine = ferrum_types::EngineConfig::default();

        engine
            .apply_runtime_config_snapshot(&effective)
            .expect("verify profile detail should apply");

        assert_eq!(
            engine.runtime.profile_detail,
            ferrum_types::ObservabilityProfileDetail::Verify
        );
    }

    #[test]
    fn run_effective_runtime_config_records_layer_split_pipeline_mode() {
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let mut cmd = test_run_cmd();
        cmd.layer_split_pipeline_mode =
            Some(crate::layer_split_pipeline::LayerSplitPipelineModeArg::Batch);
        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = effective
            .entries
            .iter()
            .find(|entry| entry.key == crate::layer_split_pipeline::LAYER_SPLIT_PIPELINE_MODE_KEY)
            .expect("missing layer split pipeline mode entry");
        assert_eq!(entry.effective_value, "batch");
        assert_eq!(entry.source, RuntimeConfigSource::Cli);
        let mut engine = ferrum_types::EngineConfig::default();
        crate::layer_split_pipeline::insert_backend_option_from_runtime(
            &effective,
            &mut engine.backend.backend_options,
        )
        .unwrap();
        assert_eq!(
            engine.backend.backend_options
                [crate::layer_split_pipeline::LAYER_SPLIT_PIPELINE_MODE_BACKEND_OPTION],
            serde_json::json!("batch")
        );
    }

    #[test]
    fn run_effective_runtime_config_records_batched_graph_flag() {
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let mut cmd = test_run_cmd();
        cmd.batched_graph = true;
        cmd.disable_reusable_execution = true;
        cmd.unified_graph = true;
        cmd.unified_graph_layers_only = true;
        cmd.unified_graph_lm_head_eager = true;
        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = |key: &str| {
            effective
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key} entry"))
        };
        assert_eq!(entry("FERRUM_BATCHED_GRAPH").effective_value, "1");
        assert_eq!(
            entry("FERRUM_BATCHED_GRAPH").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(entry("FERRUM_REUSABLE_EXECUTION").effective_value, "0");
        assert_eq!(
            entry("FERRUM_REUSABLE_EXECUTION").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(entry("FERRUM_UNIFIED_GRAPH").effective_value, "1");
        assert_eq!(
            entry("FERRUM_UNIFIED_GRAPH").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(
            entry("FERRUM_UNIFIED_GRAPH_LAYERS_ONLY").effective_value,
            "1"
        );
        assert_eq!(
            entry("FERRUM_UNIFIED_GRAPH_LAYERS_ONLY").source,
            RuntimeConfigSource::Cli
        );
        assert_eq!(
            entry("FERRUM_UNIFIED_GRAPH_LM_HEAD_EAGER").effective_value,
            "1"
        );
        assert_eq!(
            entry("FERRUM_UNIFIED_GRAPH_LM_HEAD_EAGER").source,
            RuntimeConfigSource::Cli
        );
    }

    #[test]
    fn run_effective_runtime_config_records_gpu_device_selection() {
        let selection = crate::gpu_devices::GpuDeviceSelection {
            raw_cli_value: "1".to_string(),
            requested_gpu_devices: vec![1],
            selected_gpu_devices: vec![1],
            cuda_device_count: 2,
            selected_distributed_strategy: "single_gpu".to_string(),
            selected_layer_split_plan: None,
            selected_layer_split_stages: None,
        };
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let cmd = test_run_cmd();
        let cli_entries = run_startup_cli_runtime_entries(&cmd, Some(&selection));
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = |key: &str| {
            effective
                .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_REQUESTED_GPU_DEVICES").effective_value, "1");
        assert_eq!(entry("FERRUM_SELECTED_GPU_DEVICES").effective_value, "1");
        assert_eq!(
            entry("FERRUM_SELECTED_DISTRIBUTED_STRATEGY").effective_value,
            "single_gpu"
        );
    }

    #[test]
    fn run_effective_runtime_config_records_cli_runtime_limits() {
        let mut cmd = test_run_cmd();
        cmd.kv_capacity = Some(2048);
        cmd.kv_max_blocks = Some(4096);
        cmd.max_model_len = Some(8192);
        cmd.max_num_seqs = Some(8);
        cmd.max_num_batched_tokens = Some(1024);
        cmd.sequence_fit_policy = Some(crate::commands::SequenceFitPolicyArg::FullInputMustFit);
        let snapshot = RuntimeConfigSnapshot::from_entries(Vec::new());
        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let entry = |key: &str| {
            effective
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_KV_CAPACITY").effective_value, "2048");
        assert_eq!(entry("FERRUM_KV_MAX_BLOCKS").effective_value, "4096");
        assert_eq!(entry("FERRUM_MAX_MODEL_LEN").effective_value, "8192");
        assert_eq!(entry("FERRUM_PAGED_MAX_SEQS").effective_value, "8");
        assert_eq!(entry("FERRUM_MAX_BATCHED_TOKENS").effective_value, "1024");
        assert_eq!(
            entry("FERRUM_SEQUENCE_FIT_POLICY").effective_value,
            "full-input-must-fit"
        );
        assert_eq!(
            entry("FERRUM_MAX_MODEL_LEN").source,
            RuntimeConfigSource::Cli
        );
    }

    #[test]
    fn run_runtime_config_precedence_is_config_then_env_then_cli() {
        let mut config = CliConfig::default();
        config.runtime.sequence_fit_policy = Some(SequenceFitPolicy::FullInputMustFit);

        let config_only =
            run_base_runtime_config(&config, RuntimeConfigSnapshot::from_entries(Vec::new()));
        let config_entry = config_only
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_SEQUENCE_FIT_POLICY")
            .expect("config sequence fit policy is missing");
        assert_eq!(config_entry.effective_value, "full-input-must-fit");
        assert_eq!(config_entry.source, RuntimeConfigSource::ConfigFile);

        let env_wins = run_base_runtime_config(
            &config,
            RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
                "FERRUM_SEQUENCE_FIT_POLICY",
                "immediate-only",
                RuntimeConfigSource::Env,
            )]),
        );
        let env_entry = env_wins
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_SEQUENCE_FIT_POLICY")
            .expect("env sequence fit policy is missing");
        assert_eq!(env_entry.effective_value, "immediate-only");
        assert_eq!(env_entry.source, RuntimeConfigSource::Env);

        let mut cmd = test_run_cmd();
        cmd.sequence_fit_policy = Some(crate::commands::SequenceFitPolicyArg::FullInputMustFit);
        let effective =
            run_effective_runtime_config(&env_wins, &run_startup_cli_runtime_entries(&cmd, None));
        let cli_entry = effective
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_SEQUENCE_FIT_POLICY")
            .expect("CLI sequence fit policy is missing");
        assert_eq!(cli_entry.effective_value, "full-input-must-fit");
        assert_eq!(cli_entry.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn run_product_defaults_are_typed_and_apply_to_engine_config() {
        let config = CliConfig::default();
        let effective =
            run_base_runtime_config(&config, RuntimeConfigSnapshot::from_entries(Vec::new()));
        let entry = |key: &str| {
            effective
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };

        assert_eq!(entry("FERRUM_PAGED_MAX_SEQS").effective_value, "1");
        assert_eq!(
            entry("FERRUM_PAGED_MAX_SEQS").source,
            RuntimeConfigSource::Default
        );
        assert_eq!(
            entry("FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS").effective_value,
            "1"
        );
        assert_eq!(
            entry("FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS").source,
            RuntimeConfigSource::Default
        );

        let mut engine_config = ferrum_types::EngineConfig::default();
        engine_config
            .apply_runtime_config_snapshot(&effective)
            .expect("run defaults should apply to engine config");
        assert_eq!(engine_config.scheduler.max_running_requests, 1);
        assert_eq!(
            engine_config
                .backend
                .reusable_execution_capture
                .exact_decode_widths,
            Some(vec![1])
        );
    }

    #[test]
    fn run_product_config_and_env_override_entrypoint_defaults() {
        let mut config = CliConfig::default();
        config.runtime.paged_max_seqs = Some(4);
        config.runtime.reusable_execution_exact_decode_widths = Some(vec![1, 2, 4]);

        let config_only =
            run_base_runtime_config(&config, RuntimeConfigSnapshot::from_entries(Vec::new()));
        let config_entry = |key: &str| {
            config_only
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };
        assert_eq!(config_entry("FERRUM_PAGED_MAX_SEQS").effective_value, "4");
        assert_eq!(
            config_entry("FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS").effective_value,
            "1,2,4"
        );
        assert_eq!(
            config_entry("FERRUM_PAGED_MAX_SEQS").source,
            RuntimeConfigSource::ConfigFile
        );

        let env_wins = run_base_runtime_config(
            &config,
            RuntimeConfigSnapshot::from_entries([
                RuntimeConfigEntry::new("FERRUM_PAGED_MAX_SEQS", "8", RuntimeConfigSource::Env),
                RuntimeConfigEntry::new(
                    "FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS",
                    "1,2,4,8",
                    RuntimeConfigSource::Env,
                ),
            ]),
        );
        let env_entry = |key: &str| {
            env_wins
                .entries
                .iter()
                .find(|entry| entry.key == key)
                .unwrap_or_else(|| panic!("missing {key}"))
        };
        assert_eq!(env_entry("FERRUM_PAGED_MAX_SEQS").effective_value, "8");
        assert_eq!(
            env_entry("FERRUM_REUSABLE_EXECUTION_EXACT_DECODE_WIDTHS").effective_value,
            "1,2,4,8"
        );
        assert_eq!(
            env_entry("FERRUM_PAGED_MAX_SEQS").source,
            RuntimeConfigSource::Env
        );

        let mut cmd = test_run_cmd();
        cmd.max_num_seqs = Some(16);
        let cli_wins =
            run_effective_runtime_config(&env_wins, &run_startup_cli_runtime_entries(&cmd, None));
        let admission = cli_wins
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_PAGED_MAX_SEQS")
            .expect("CLI admission is missing");
        assert_eq!(admission.effective_value, "16");
        assert_eq!(admission.source, RuntimeConfigSource::Cli);
    }

    #[test]
    fn run_effective_runtime_config_applies_recurrent_state_slots_to_engine_config() {
        let cmd = test_run_cmd();
        let snapshot = RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
            "FERRUM_RECURRENT_STATE_MAX_SLOTS",
            "16",
            RuntimeConfigSource::ConfigFile,
        )]);
        let cli_entries = run_startup_cli_runtime_entries(&cmd, None);
        let effective = run_effective_runtime_config(&snapshot, &cli_entries);
        let mut engine_config = ferrum_types::EngineConfig::default();

        engine_config
            .apply_runtime_config_snapshot(&effective)
            .expect("run effective runtime config should apply");

        assert_eq!(engine_config.runtime.recurrent_state_max_slots, Some(16));
    }

    #[test]
    fn run_startup_auto_config_renders_effective_config_schema() {
        let resolved = run_startup_auto_config(
            crate::commands::serve::hardware_capabilities_for_device(&ferrum_types::Device::CPU),
            None,
            ferrum_types::ExecutionResourceAuthority::LegacyEngine,
            None,
            None,
            RuntimeConfigSnapshot::from_entries(Vec::new()),
        )
        .expect("auto config");
        let doc = resolved.effective_config_document();
        assert_eq!(doc["schema_version"], 1);
        assert!(doc["entries"].is_array());
        assert!(doc["model_capabilities"].is_object());
        assert!(doc["hardware_capabilities"].is_object());
        assert!(doc["workload_profile"].is_object());
        assert_eq!(doc["workload_profile"]["target_concurrency"], 1);
        assert_eq!(doc["admission"]["effective_max_concurrent"], 1);
        assert!(doc["decisions"].is_array());
    }

    #[test]
    fn unknown_backend_is_rejected() {
        let err = select_device("not-a-backend").expect_err("unknown backend must fail");
        assert!(
            err.to_string().contains("unknown backend"),
            "unexpected error: {err}"
        );
    }

    #[cfg(not(all(target_os = "macos", feature = "metal")))]
    #[test]
    fn explicit_metal_backend_without_compiled_support_is_rejected() {
        let err = select_device("metal").expect_err("unsupported explicit Metal must fail");
        assert!(
            err.to_string()
                .contains("requested backend 'metal' but this ferrum binary was not built"),
            "unexpected error: {err}"
        );
    }

    #[cfg(not(feature = "cuda"))]
    #[test]
    fn explicit_cuda_backend_without_compiled_support_is_rejected() {
        let err = select_device("cuda").expect_err("unsupported explicit CUDA must fail");
        assert!(
            err.to_string()
                .contains("requested backend 'cuda' but this ferrum binary was not built"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn run_metadata_forbids_initial_thinking_close_without_open_block() {
        let metadata = run_request_metadata(
            "<|im_start|>assistant\n",
            &ChatTemplateOptions::default(),
            ModelOutputProtocol::Text,
            None,
        );
        let forbidden = metadata
            .get(RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY)
            .and_then(|value| value.as_array())
            .expect("initial forbidden token texts");
        assert_eq!(
            forbidden,
            &[serde_json::Value::String(THINK_END_TAG.to_string())]
        );
    }

    #[test]
    fn run_thinking_options_preserve_model_default_and_explicit_overrides() {
        let template = ModelChatTemplate::new(
            "{% if enable_thinking is defined %}{{ enable_thinking }}{% endif %}",
            "thinking-template",
        );
        let mut cmd = test_run_cmd();

        assert_eq!(
            build_chat_template_options(&cmd, Some(&template)).enable_thinking,
            None
        );

        cmd.enable_thinking = true;
        assert_eq!(
            build_chat_template_options(&cmd, Some(&template)).enable_thinking,
            Some(true)
        );

        cmd.enable_thinking = false;
        cmd.disable_thinking = true;
        assert_eq!(
            build_chat_template_options(&cmd, Some(&template)).enable_thinking,
            Some(false)
        );
    }

    #[test]
    fn run_metadata_forbids_initial_thinking_start_when_template_disables_thinking() {
        for (source, expected_protocol) in [
            (
                "<assistant>{% if enable_thinking is defined and enable_thinking is false %}<think>\n\n</think>\n\n{% else %}<think>\n{% endif %}",
                ferrum_types::ModelReasoningProtocol::PromptOpened,
            ),
            (
                "<assistant>{% if enable_thinking is defined and enable_thinking is false %}<think>\n\n</think>\n\n{% endif %}",
                ferrum_types::ModelReasoningProtocol::ModelGenerated,
            ),
        ] {
            let template = ModelChatTemplate::new(source, "thinking-contract");
            assert_eq!(template.reasoning_protocol, expected_protocol);
            let mut cmd = test_run_cmd();
            cmd.disable_thinking = true;
            let options = build_chat_template_options(&cmd, Some(&template));
            let prompt = build_chat_prompt(
                &[],
                "hello",
                None,
                "served-alias",
                Some(&template),
                &options,
            )
            .unwrap();
            let metadata = run_request_metadata(
                &prompt,
                &options,
                template.output_protocol,
                Some(&template),
            );
            assert_eq!(
                metadata[RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY],
                serde_json::json!([THINK_END_TAG, THINK_START_TAG]),
                "{expected_protocol:?}",
            );

            cmd.disable_thinking = false;
            let options = build_chat_template_options(&cmd, Some(&template));
            let prompt = build_chat_prompt(
                &[],
                "hello",
                None,
                "served-alias",
                Some(&template),
                &options,
            )
            .unwrap();
            let metadata = run_request_metadata(
                &prompt,
                &options,
                template.output_protocol,
                Some(&template),
            );
            let forbidden = metadata
                .get(RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY)
                .and_then(serde_json::Value::as_array);
            assert!(forbidden.is_none_or(|tokens| !tokens.contains(&serde_json::json!(THINK_START_TAG))));
        }
    }

    #[test]
    fn run_disabling_thinking_preserves_sampling_for_plain_or_unknown_templates() {
        let requires_system_message = concat!(
            "{% if messages[0].role != 'system' %}",
            "{{ raise_exception('a system message is required') }}{% endif %}",
            "{{ messages[0].content }}|{{ messages[1].content }}",
        );
        for (source, expected_protocol) in [
            (
                "{{ messages[-1].content }}",
                ferrum_types::ModelReasoningProtocol::None,
            ),
            (
                requires_system_message,
                ferrum_types::ModelReasoningProtocol::Unknown,
            ),
        ] {
            let template = ModelChatTemplate::new(source, "conversation-contract");
            assert_eq!(template.reasoning_protocol, expected_protocol);
            let mut cmd = test_run_cmd();
            let omitted = build_chat_template_options(&cmd, Some(&template));
            cmd.disable_thinking = true;
            let disabled = build_chat_template_options(&cmd, Some(&template));
            let render = |options: &ChatTemplateOptions| {
                build_chat_prompt(
                    &[],
                    "hello",
                    Some("Be concise."),
                    "served-alias",
                    Some(&template),
                    options,
                )
                .unwrap()
            };
            let original_prompt = render(&omitted);
            let disabled_prompt = render(&disabled);
            assert_eq!(original_prompt, disabled_prompt);
            let original_metadata = run_request_metadata(
                &original_prompt,
                &omitted,
                template.output_protocol,
                Some(&template),
            );
            let disabled_metadata = run_request_metadata(
                &disabled_prompt,
                &disabled,
                template.output_protocol,
                Some(&template),
            );
            assert_eq!(
                original_metadata, disabled_metadata,
                "{expected_protocol:?}"
            );
            assert_eq!(
                disabled_metadata[RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY],
                serde_json::json!([THINK_END_TAG]),
            );
        }
    }

    #[test]
    fn run_metadata_allows_thinking_close_when_prompt_opened_block() {
        let metadata = run_request_metadata(
            "<|im_start|>assistant\n<think>\n",
            &ChatTemplateOptions::default(),
            ModelOutputProtocol::Text,
            None,
        );
        assert!(!metadata.contains_key(RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY));
    }

    #[test]
    fn gemma_run_metadata_and_completion_follow_the_rendered_thought_state() {
        let protocol = ModelOutputProtocol::GemmaThought;
        let options = ChatTemplateOptions {
            enable_thinking: Some(false),
            ..Default::default()
        };
        for (prompt, opened) in [
            ("<|turn>model\n", false),
            ("<|turn>model\n<|channel>thought\n<channel|>", false),
            ("<|turn>model\n<|tool_response>579<tool_response|>", false),
            (
                "<|turn>model\n<|tool_response>579<tool_response|><|channel>thought\n",
                true,
            ),
        ] {
            let metadata = run_request_metadata(prompt, &options, protocol, None);
            if opened {
                assert!(!metadata.contains_key(RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY));
            } else {
                assert_eq!(
                    metadata[RUN_INITIAL_FORBIDDEN_TOKEN_TEXTS_METADATA_KEY],
                    serde_json::json!(["<channel|>"]),
                    "disabling reasoning must still allow an empty native thought header"
                );
            }
            let mut params = SamplingParams::greedy();
            params.model_output_protocol = protocol;
            let params = sampling_params_for_prompt(params, prompt);
            assert_eq!(
                params.response_completion_boundary,
                if opened {
                    ResponseCompletionBoundary::AfterDelimiterAndPayload {
                        delimiter: "<channel|>".to_string(),
                        alternate_envelope: None,
                    }
                } else {
                    ResponseCompletionBoundary::Immediate
                }
            );
        }
    }

    #[test]
    fn gemma_run_stream_splits_never_expose_thought_frames_or_reasoning() {
        for (raw, opened, expected) in [
            ("<|channel>thought\n<channel|>579", false, "579"),
            ("Compute.<channel|>579", true, "579"),
            (
                "Before.<|channel>thought\nOne.<channel|>Between.\
                 <|channel>thought\nTwo 🧠.<channel|>After.",
                false,
                "Before.Between.After.",
            ),
            ("thought\n579", false, "thought\n579"),
            ("Answer.<|channel>th", false, "Answer."),
        ] {
            for split in raw
                .char_indices()
                .map(|(index, _)| index)
                .chain(std::iter::once(raw.len()))
            {
                let mut output = RunStreamOutput::new(ModelOutputProtocol::GemmaThought, opened);
                let mut cumulative = String::new();
                let mut visible = String::new();
                for chunk in [&raw[..split], &raw[split..]] {
                    cumulative.push_str(chunk);
                    if let Some(delta) = output.delta(&cumulative, chunk).unwrap() {
                        visible.push_str(&delta);
                    }
                    assert!(expected.starts_with(&visible));
                }
                if let Some(delta) = output.finish(&cumulative).unwrap() {
                    visible.push_str(&delta);
                }
                assert_eq!(visible, expected, "split at {split}");
                assert!(output.finish(&cumulative).unwrap().is_none());
                let parsed = parse_run_model_output(
                    ModelOutputProtocol::GemmaThought,
                    &cumulative,
                    opened,
                    Some(FinishReason::Length),
                )
                .unwrap();
                assert_eq!(parsed.content, expected);
            }
        }
    }

    #[test]
    fn gemma_run_stream_delivers_plain_content_early_and_rejects_unknown_channels() {
        let mut output = RunStreamOutput::new(ModelOutputProtocol::GemmaThought, false);
        assert_eq!(
            output.delta("Hello.", "Hello.").unwrap().as_deref(),
            Some("Hello.")
        );
        assert!(output
            .delta("Hello.<|channel>", "<|channel>")
            .unwrap()
            .is_none());
        let error = output
            .delta("Hello.<|channel>unknown\nsecret", "unknown\nsecret")
            .unwrap_err();
        assert!(!error.to_string().contains("secret"));
        assert!(!error.to_string().contains("<|"));
    }

    #[test]
    fn jsonl_v2_assistant_binds_reasoning_usage_and_history() {
        let history = vec![
            RunHistoryMessage::new("user", "first"),
            RunHistoryMessage::new("assistant", "<think>why</think>answer"),
        ];
        let usage = TokenUsage::new(7, 3);
        let record = jsonl_assistant_record(
            "session-1",
            2,
            "request-1",
            1,
            "answer",
            Some("why"),
            &history,
            Some(FinishReason::EOS),
            Some(&usage),
            3,
            2,
            "<think>why</think>answer",
            12.5,
        );
        assert_eq!(record["schema_version"], RUN_JSONL_SCHEMA_VERSION);
        assert_eq!(record["session_id"], "session-1");
        assert_eq!(record["history_epoch"], 2);
        assert_eq!(record["request_id"], "request-1");
        assert_eq!(record["content"], "answer");
        assert_eq!(record["reasoning"], "why");
        assert_eq!(record["usage"]["prompt_tokens"], 7);
        assert_eq!(record["usage"]["completion_tokens"], 3);
        assert_eq!(record["usage"]["total_tokens"], 10);
        assert_eq!(record["history_before"]["message_count"], 2);
        assert_eq!(record["history_before"]["turn_count"], 1);
        assert_eq!(
            record["raw_text_sha256"],
            sha256_text("<think>why</think>answer")
        );
    }

    #[test]
    fn text_run_jsonl_preserves_literal_think_tags_after_open_reasoning() {
        let raw = "reason</think>\n{\"text\":\"<think>literal</think>\"}";
        let parsed = parse_run_model_output(
            ModelOutputProtocol::Text,
            raw,
            true,
            Some(FinishReason::EOS),
        )
        .unwrap();
        let record = jsonl_assistant_record(
            "session-1",
            0,
            "request-1",
            0,
            &parsed.content,
            parsed.reasoning.as_deref(),
            &[],
            Some(FinishReason::EOS),
            None,
            1,
            1,
            raw,
            0.0,
        );

        assert_eq!(record["content"], r#"{"text":"<think>literal</think>"}"#);
        assert_eq!(record["reasoning"], "reason");
        assert_eq!(record["raw_text_sha256"], sha256_text(raw));

        // Text history remains raw; these parsed fields belong to JSONL output.
        let history = RunHistoryMessage::assistant(raw, ModelOutputProtocol::Text, &parsed);
        assert_eq!(history.prompt.content, raw);
        assert!(history.prompt.reasoning_content.is_none());
    }

    #[test]
    fn history_evidence_changes_when_reasoning_history_changes() {
        let first = vec![RunHistoryMessage::new("assistant", "answer")];
        let second = vec![RunHistoryMessage::new(
            "assistant",
            "<think>why</think>answer",
        )];
        assert_ne!(
            history_evidence(&first)["sha256"],
            history_evidence(&second)["sha256"]
        );
    }

    #[test]
    fn jsonl_v2_delta_preserves_utf8_bytes_and_request_binding() {
        let record =
            jsonl_assistant_delta_record("session-1", 3, "request-1", 4, 2, "🙂", Some(9271));
        assert_eq!(record["event"], "assistant_delta");
        assert_eq!(record["session_id"], "session-1");
        assert_eq!(record["history_epoch"], 3);
        assert_eq!(record["request_id"], "request-1");
        assert_eq!(record["turn"], 4);
        assert_eq!(record["index"], 2);
        assert_eq!(record["raw_text_delta"], "🙂");
        assert_eq!(record["utf8_bytes"], 4);
        assert_eq!(record["token_id"], 9271);
    }

    #[tokio::test]
    async fn run_collectors_preserve_tokenless_tail_before_terminal() {
        let request_id = RequestId::new();
        let expected_request_id = request_id.to_string();
        for text_output in [false, true] {
            let chunks = [
                ("hello ", Some(TokenId::new(11)), None, None),
                ("尾", None, None, None),
                (
                    "",
                    None,
                    Some(FinishReason::Length),
                    Some(TokenUsage::new(7, 2)),
                ),
            ]
            .into_iter()
            .map(|(text, token, finish_reason, usage)| {
                Ok(StreamChunk {
                    request_id: request_id.clone(),
                    text: text.to_string(),
                    token,
                    finish_reason,
                    usage,
                    created_at: chrono::Utc::now(),
                    metadata: HashMap::new(),
                    api_response: None,
                    execution_evidence: None,
                })
            })
            .collect::<Vec<_>>();
            let stream: RunResponseStream = Box::pin(futures::stream::iter(chunks));
            let result = if text_output {
                collect_run_text_stream(
                    stream,
                    false,
                    RunStreamOutput::new(ModelOutputProtocol::HarmonyGptOss, false),
                    0,
                    &request_id,
                    false,
                    true,
                )
                .await
            } else {
                collect_run_stream(stream, false, true, 0, "session", 0, &expected_request_id).await
            }
            .expect("collect stream with a separately flushed tail");

            assert_eq!(result.raw_text, "hello 尾");
            assert_eq!(result.finish_reason, Some(FinishReason::Length));
            assert_eq!(result.token_ids, vec![11]);
            assert_eq!(result.token_count, 2, "final usage is authoritative");
            assert_eq!(result.chunk_count, 2);
            let usage = result.usage.expect("terminal usage");
            assert_eq!(usage.prompt_tokens, 7);
            assert_eq!(usage.completion_tokens, 2);
        }
    }

    #[tokio::test]
    async fn run_collectors_reject_failed_or_truncated_generation() {
        let request_id = RequestId(Uuid::new_v4());
        for terminal in [
            Some(FinishReason::Error),
            None,
            Some(FinishReason::EOS),
            Some(FinishReason::Stop),
            Some(FinishReason::Length),
        ] {
            for text_output in [false, true] {
                let chunks = [("partial", None), ("", terminal)]
                    .into_iter()
                    .map(|(text, finish_reason)| {
                        Ok(StreamChunk {
                            request_id: request_id.clone(),
                            text: text.into(),
                            token: None,
                            finish_reason,
                            usage: None,
                            created_at: Utc::now(),
                            metadata: HashMap::new(),
                            api_response: None,
                            execution_evidence: None,
                        })
                    })
                    .collect::<Vec<_>>();
                let stream: RunResponseStream = Box::pin(futures::stream::iter(chunks));
                let result = if text_output {
                    collect_run_text_stream(
                        stream,
                        false,
                        RunStreamOutput::new(ModelOutputProtocol::Text, false),
                        0,
                        &request_id,
                        false,
                        false,
                    )
                    .await
                } else {
                    collect_run_stream(
                        stream,
                        false,
                        true,
                        0,
                        "session",
                        0,
                        &request_id.to_string(),
                    )
                    .await
                };
                assert_eq!(
                    result.is_ok(),
                    matches!(
                        terminal,
                        Some(FinishReason::EOS | FinishReason::Stop | FinishReason::Length)
                    ),
                    "terminal={terminal:?}, text_output={text_output}",
                );
            }
        }
    }

    #[test]
    fn run_one_shot_response_rejects_engine_error_after_partial_output() {
        let mut response = InferenceResponse {
            request_id: RequestId(Uuid::new_v4()),
            text: "partial output before failure".into(),
            tokens: vec![TokenId::new(1)],
            finish_reason: FinishReason::Error,
            usage: TokenUsage::new(7, 1),
            latency_ms: 1,
            created_at: Utc::now(),
            metadata: HashMap::new(),
            api_response: None,
            execution_evidence: None,
        };
        assert!(CollectedRunGeneration::from_response(response.clone()).is_err());
        response.finish_reason = FinishReason::Length;
        let collected = CollectedRunGeneration::from_response(response).unwrap();
        assert_eq!(collected.raw_text, "partial output before failure");
        assert_eq!(collected.token_count, 1);
    }

    #[test]
    fn cli_display_preserves_thinking_markers() {
        assert_eq!(
            display_response_text("<think>\nreasoning\n</think>\n\n最终答案"),
            "<think>\nreasoning\n</think>\n\n最终答案"
        );
    }

    #[test]
    fn cli_display_preserves_orphan_think_close() {
        assert_eq!(
            display_response_text("</think>\n\n你好！很高兴见到你。"),
            "</think>\n\n你好！很高兴见到你。"
        );
    }

    #[test]
    fn default_run_temperature_is_greedy() {
        let cmd = test_run_cmd();
        assert_eq!(build_sampling_params(&cmd).temperature, 0.0);
        assert_eq!(build_sampling_params(&cmd).max_tokens, 4096);
    }

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

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            run: RunCommand,
        }

        let parsed = TestCli::parse_from([
            "ferrum",
            "qwen3.5",
            "--temperature",
            "1.0",
            "--min-p",
            "0.05",
            "--presence-penalty",
            "1.5",
        ]);
        let params = build_sampling_params(&parsed.run);
        assert_eq!(params.min_p, Some(0.05));
        assert_eq!(params.presence_penalty, 1.5);
        params
            .validate()
            .expect("official sampling controls must validate");

        let disabled = TestCli::parse_from(["ferrum", "qwen3.5", "--min-p", "0.0"]);
        assert_eq!(build_sampling_params(&disabled.run).min_p, None);
    }

    #[test]
    fn chat_default_applies_repetition_penalty() {
        // The chat default is greedy (temperature 0). Greedy with NO repetition
        // penalty deterministically locks into token loops on some inputs (the
        // "2D/3D 2D/3D..." degeneration a user hit). The clap default must carry
        // a penalty (OpenAI/llama.cpp standard 1.1) so the out-of-box chat does
        // not loop. Parsing the real CLI (not the struct-literal test fixture)
        // is what pins the actual default users get.
        use clap::Parser;
        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            run: RunCommand,
        }
        let parsed = TestCli::parse_from(["ferrum", "qwen3:0.6b"]);
        assert!(
            parsed.run.repeat_penalty > 1.0,
            "chat default repeat_penalty must discourage repeats, got {}",
            parsed.run.repeat_penalty
        );
        assert!(
            build_sampling_params(&parsed.run).repetition_penalty > 1.0,
            "build_sampling_params must propagate the default penalty"
        );
    }

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

        #[derive(Parser)]
        struct TestCli {
            #[command(flatten)]
            run: RunCommand,
        }

        let error = match TestCli::try_parse_from(["ferrum", "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 run_sampling_params_include_cli_stop_sequences() {
        let mut cmd = test_run_cmd();
        cmd.stop = vec!["\n".to_string(), String::new(), "END".to_string()];
        let params = build_sampling_params(&cmd);
        assert!(params.stop_sequences.contains(&"\n".to_string()));
        assert!(params.stop_sequences.contains(&"END".to_string()));
        assert!(!params.stop_sequences.contains(&String::new()));
    }

    #[test]
    fn response_completion_contract_is_set_on_run_prompt_plan() {
        let cmd = test_run_cmd();
        let budget = whitespace_budget(8192);
        let template = ModelChatTemplate::new(
            "{% if add_generation_prompt %}<assistant><think>\n{% endif %}",
            "thinking-test-template",
        );
        let plan = build_run_prompt_plan(
            &[],
            "demo",
            None,
            "tinyllama",
            Some(&template),
            &ChatTemplateOptions::default(),
            &cmd,
            &budget,
        )
        .unwrap();

        assert!(has_unclosed_thinking_block(&plan.prompt));
        assert_eq!(
            plan.sampling_params.response_completion_boundary,
            ResponseCompletionBoundary::AfterDelimiterAndPayload {
                delimiter: THINK_END_TAG.to_string(),
                alternate_envelope: None,
            }
        );
    }

    #[test]
    fn cpu_run_skips_gpu_chat_autosize_defaults() {
        assert!(run_autosize_for_device(&ferrum_types::Device::CPU, 0.9).is_none());
    }

    #[cfg(any(all(target_os = "macos", feature = "metal"), feature = "cuda"))]
    #[test]
    fn accelerator_run_keeps_chat_autosize_defaults() {
        #[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);

        let autosize = run_autosize_for_device(&device, 0.75);
        assert_eq!(
            autosize,
            Some((crate::gpu_mem_autosize::AutoSizeProfile::Chat, 0.75))
        );
    }

    #[test]
    fn context_shift_clamps_output_to_remaining_kv_budget() {
        let cmd = test_run_cmd();
        let budget = whitespace_budget(64);
        let options = default_template_options();
        let plan = build_run_prompt_plan(
            &[],
            "demo",
            None,
            "tinyllama",
            None,
            &options,
            &cmd,
            &budget,
        )
        .unwrap();

        let prompt_tokens = plan.prompt_tokens.unwrap();
        assert!(prompt_tokens < 64);
        assert_eq!(plan.max_tokens_clamped_from, Some(4096));
        assert_eq!(plan.sampling_params.max_tokens, 64 - prompt_tokens);
    }

    #[test]
    fn run_plan_context_uses_model_limit_for_engine_and_history() {
        let model_dir = tempfile::tempdir().unwrap();
        let requested =
            run_base_runtime_config(&CliConfig::default(), RuntimeConfigSnapshot::default());
        let mut effective = requested.clone();
        effective.upsert(
            "FERRUM_MAX_BATCHED_TOKENS",
            "2048",
            RuntimeConfigSource::MemoryProfile,
        );
        let mut engine = ferrum_types::EngineConfig::default();
        let authority = ferrum_types::ExecutionResourceAuthority::PlanRuntime;
        let mut capabilities = ModelCapabilities::unknown();
        capabilities.max_context_len = Some(262_144);
        let resolved = run_startup_auto_config(
            crate::commands::serve::hardware_capabilities_for_device(&ferrum_types::Device::CPU),
            Some(capabilities),
            authority,
            None,
            None,
            effective,
        )
        .unwrap();
        engine
            .apply_runtime_config_snapshot(&resolved.runtime_config)
            .unwrap();
        assert_eq!(engine.runtime.kv_capacity, None);
        assert_eq!(engine.runtime.max_model_len, Some(262_144));
        assert_eq!(engine.batching.max_num_batched_tokens, 2048);
        assert_eq!(engine.scheduler.max_running_requests, 1);
        assert_eq!(
            crate::runtime_env::runtime_snapshot_value(
                &resolved.runtime_config,
                "FERRUM_KV_CAPACITY"
            ),
            None
        );

        let mut budget = RunBudget::from_product_sources(
            None,
            None,
            model_dir.path(),
            &resolved.runtime_config,
            resolved.model_capabilities.max_context_len,
        )
        .unwrap();
        budget.prompt_token_counter = Some(|prompt| prompt.split_whitespace().count());
        let long = std::iter::repeat_n("old", 4300)
            .collect::<Vec<_>>()
            .join(" ");
        let history = [
            RunHistoryMessage::new("user", &long),
            RunHistoryMessage::new("assistant", &long),
        ];
        let cmd = test_run_cmd();
        let plan = build_run_prompt_plan(
            &history,
            "continue",
            None,
            "context-fixture",
            None,
            &default_template_options(),
            &cmd,
            &budget,
        )
        .unwrap();
        assert_eq!(plan.kv_capacity, Some(262_144));
        assert!(plan.prompt_tokens.unwrap() > 8192);
        assert_eq!(plan.dropped_history_messages, 0);
        assert_eq!(plan.max_tokens_clamped_from, None);
        assert_eq!(plan.sampling_params.max_tokens, cmd.max_tokens as usize);
    }

    #[test]
    fn run_plan_context_preserves_explicit_kv_limits_and_precedence() {
        let model_dir = tempfile::tempdir().unwrap();
        let mut config = CliConfig::default();
        config.runtime.kv_capacity = Some(16_384);
        let config_only = run_base_runtime_config(&config, RuntimeConfigSnapshot::default());
        let with_env = run_base_runtime_config(
            &config,
            RuntimeConfigSnapshot::from_env_vars([("FERRUM_KV_CAPACITY", "12288")]),
        );
        let mut cmd = test_run_cmd();
        cmd.kv_capacity = Some(4096);
        let with_cli =
            run_effective_runtime_config(&with_env, &run_startup_cli_runtime_entries(&cmd, None));

        for (requested, expected, source) in [
            (config_only, 16_384, RuntimeConfigSource::ConfigFile),
            (with_env, 12_288, RuntimeConfigSource::Env),
            (with_cli, 4096, RuntimeConfigSource::Cli),
        ] {
            let effective = requested.clone();
            let mut engine = ferrum_types::EngineConfig::default();
            engine.apply_runtime_config_snapshot(&effective).unwrap();
            assert_eq!(engine.runtime.kv_capacity, Some(expected));
            let entry = effective
                .entries
                .iter()
                .find(|entry| entry.key == "FERRUM_KV_CAPACITY")
                .unwrap();
            assert_eq!(entry.effective_value, expected.to_string());
            assert_eq!(entry.source, source);
            let budget = RunBudget::from_product_sources(
                None,
                None,
                model_dir.path(),
                &effective,
                Some(262_144),
            )
            .unwrap();
            assert_eq!(budget.kv_capacity, Some(expected));
        }
    }

    #[test]
    fn legacy_run_keeps_automatic_kv_capacity() {
        let source = tempfile::tempdir().unwrap();
        let entries = crate::source_resolver::chat_profile_runtime_entries(
            source.path(),
            &RuntimeConfigSnapshot::default(),
            RuntimeConfigSource::Default,
        );
        let snapshot = RuntimeConfigSnapshot::from_entries(entries);
        let mut engine = ferrum_types::EngineConfig::default();
        engine.apply_runtime_config_snapshot(&snapshot).unwrap();
        assert_eq!(engine.runtime.kv_capacity, Some(8192));
    }

    #[tokio::test]
    async fn run_explicit_kv_limit_cannot_expand_the_model_context() {
        let model_dir = tempfile::tempdir().unwrap();
        let requested = RuntimeConfigSnapshot::from_entries([RuntimeConfigEntry::new(
            "FERRUM_KV_CAPACITY",
            "65536",
            RuntimeConfigSource::Cli,
        )]);
        let effective = requested.clone();
        let mut engine_config = ferrum_types::EngineConfig::default();
        let authority = ferrum_types::ExecutionResourceAuthority::PlanRuntime;
        let mut capabilities = ModelCapabilities::unknown();
        capabilities.max_context_len = Some(32_768);
        let resolved = run_startup_auto_config(
            crate::commands::serve::hardware_capabilities_for_device(&ferrum_types::Device::CPU),
            Some(capabilities),
            authority,
            None,
            None,
            effective,
        )
        .unwrap();
        engine_config
            .apply_runtime_config_snapshot(&resolved.runtime_config)
            .unwrap();
        assert_eq!(engine_config.runtime.kv_capacity, Some(65_536));
        assert_eq!(engine_config.runtime.max_model_len, Some(32_768));
        let maximum = resolved
            .runtime_config
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_MAX_MODEL_LEN")
            .unwrap();
        assert_eq!(maximum.source, RuntimeConfigSource::Default);
        let budget = RunBudget::from_product_sources(
            None,
            None,
            model_dir.path(),
            &resolved.runtime_config,
            resolved.model_capabilities.max_context_len,
        )
        .unwrap();
        // The CPU stub exercises the production engine's common request guard
        // without loading weights or requiring the selected native backend.
        let engine = ferrum_engine::EngineBuilder::new(engine_config)
            .with_tokenizer("stub")
            .with_executor("stub")
            .build()
            .await
            .unwrap();
        assert_eq!(budget.kv_capacity, Some(32_768));
        assert_eq!(engine.context_capacity(), budget.kv_capacity);
        assert_eq!(engine.config().runtime.max_model_len, Some(32_768));
        engine.shutdown().await.unwrap();
    }

    #[test]
    fn run_memory_plan_updates_artifacts_and_history_budget() {
        let model_dir = tempfile::tempdir().unwrap();
        let mut model = ModelCapabilities::unknown();
        model.max_context_len = Some(32_768);
        let mut resolved = run_startup_auto_config(
            crate::commands::serve::hardware_capabilities_for_device(&ferrum_types::Device::CPU),
            Some(model),
            ferrum_types::ExecutionResourceAuthority::PlanRuntime,
            None,
            None,
            run_base_runtime_config(&CliConfig::default(), RuntimeConfigSnapshot::default()),
        )
        .unwrap();
        let request = ferrum_types::StartupMemoryRequest::from_snapshot(
            ferrum_types::DeviceMemorySnapshot {
                capacity_bytes: 1024 * 1024 * 1024,
                available_bytes: 512 * 1024 * 1024,
                source: "fixture".into(),
            },
            1.0,
            &resolved.runtime_config,
        )
        .unwrap();
        let plan = ferrum_types::StartupMemoryPlan {
            request,
            requested: ferrum_types::StartupResourceLimits {
                context_tokens: 32_768,
                max_sequences: 1,
                max_batch_tokens: 2048,
            },
            selected: ferrum_types::StartupResourceLimits {
                context_tokens: 4096,
                max_sequences: 1,
                max_batch_tokens: 256,
            },
            context_peak_bytes: 500 * 1024 * 1024,
            decode_peak_bytes: 480 * 1024 * 1024,
            reasons: vec!["context and batch reduced to fit compiled workspace".into()],
        };
        let mut engine = ferrum_types::EngineConfig::default();
        engine
            .apply_runtime_config_snapshot(&resolved.runtime_config)
            .unwrap();
        let artifact = model_dir.path().join("effective.json");
        let trace = model_dir.path().join("decisions.jsonl");
        crate::commands::serve::write_failed_startup_config_artifacts(
            &resolved,
            None,
            &engine.numerical_execution,
            Some(&artifact),
            Some(&trace),
            &FerrumError::config("fixture initialization failed"),
        );
        let preliminary: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&artifact).unwrap()).unwrap();
        assert_eq!(preliminary["startup"]["status"], "failed");
        assert_eq!(preliminary["startup"]["configuration"], "preliminary");
        assert_eq!(preliminary["selected_max_model_len"], 32_768);
        plan.apply_to_engine_config(&mut engine).unwrap();
        crate::startup::apply_engine_plan(&mut resolved, &engine);
        let budget = RunBudget::from_product_sources(
            None,
            None,
            model_dir.path(),
            &resolved.runtime_config,
            engine.runtime.max_model_len,
        )
        .unwrap();
        assert_eq!(budget.kv_capacity, Some(4096));
        assert_eq!(resolved.startup_memory_plan.as_ref(), Some(&plan));
        crate::commands::serve::write_startup_config_artifacts(
            &resolved,
            None,
            &engine.numerical_execution,
            Some(&artifact),
            Some(&trace),
        )
        .unwrap();
        let document: serde_json::Value =
            serde_json::from_slice(&std::fs::read(artifact).unwrap()).unwrap();
        assert!(document.get("startup").is_none());
        assert_eq!(document["selected_max_model_len"], 4096);
        assert_eq!(
            document["startup_memory_plan"]["selected"]["context_tokens"],
            4096
        );
        assert_eq!(
            document["startup_memory_plan"]["selected"]["max_batch_tokens"],
            256
        );
        for line in std::fs::read_to_string(trace).unwrap().lines() {
            let decision: serde_json::Value = serde_json::from_str(line).unwrap();
            assert!(decision.get("startup").is_none());
        }
    }

    #[test]
    fn run_budget_clamps_generation_to_the_tighter_configured_context_bound() {
        let model_dir = tempfile::tempdir().unwrap();
        let cmd = test_run_cmd();
        for (declared_limit, model_limit, kv_limit, expected_limit) in [
            (None, Some(64), None, 64),
            (None, None, Some(48), 48),
            (None, Some(64), Some(128), 64),
            (None, Some(128), Some(48), 48),
            (None, Some(0), Some(48), 48),
            (None, Some(64), Some(0), 64),
            (Some(64), None, None, 64),
            (Some(32), Some(128), Some(48), 32),
            (Some(128), Some(64), Some(48), 48),
        ] {
            let snapshot = RuntimeConfigSnapshot::from_entries(
                [
                    ("FERRUM_MAX_MODEL_LEN", model_limit),
                    ("FERRUM_KV_CAPACITY", kv_limit),
                ]
                .into_iter()
                .filter_map(|(key, value)| {
                    value.map(|value| {
                        RuntimeConfigEntry::new(key, value.to_string(), RuntimeConfigSource::Cli)
                    })
                }),
            );
            let mut budget = RunBudget::from_product_sources(
                None,
                None,
                model_dir.path(),
                &snapshot,
                declared_limit,
            )
            .unwrap();
            budget.prompt_token_counter = Some(|prompt| prompt.split_whitespace().count());
            let plan = build_run_prompt_plan(
                &[],
                "demo",
                None,
                "tinyllama",
                None,
                &default_template_options(),
                &cmd,
                &budget,
            )
            .unwrap();
            assert_eq!(plan.kv_capacity, Some(expected_limit));
            assert_eq!(
                plan.prompt_tokens.unwrap() + plan.sampling_params.max_tokens,
                expected_limit,
            );
            assert_eq!(plan.max_tokens_clamped_from, Some(cmd.max_tokens as usize));
        }
    }

    #[test]
    fn run_budget_context_shift_uses_the_final_cli_model_length() {
        let model_dir = tempfile::tempdir().unwrap();
        let mut config = CliConfig::default();
        config.runtime.max_model_len = Some(8192);
        config.runtime.kv_capacity = Some(8192);
        let mut cmd = test_run_cmd();
        cmd.max_model_len = Some(64);
        let requested = run_effective_runtime_config(
            &run_base_runtime_config(&config, RuntimeConfigSnapshot::default()),
            &run_startup_cli_runtime_entries(&cmd, None),
        );
        let effective = requested.clone();
        let maximum = effective
            .entries
            .iter()
            .find(|entry| entry.key == "FERRUM_MAX_MODEL_LEN")
            .unwrap();
        assert_eq!(maximum.source, RuntimeConfigSource::Cli);
        assert_eq!(maximum.effective_value, "64");
        let resolved = run_startup_auto_config(
            crate::commands::serve::hardware_capabilities_for_device(&ferrum_types::Device::CPU),
            None,
            ferrum_types::ExecutionResourceAuthority::PlanRuntime,
            None,
            None,
            effective,
        )
        .unwrap();
        let mut budget = RunBudget::from_product_sources(
            None,
            None,
            model_dir.path(),
            &resolved.runtime_config,
            resolved.model_capabilities.max_context_len,
        )
        .unwrap();
        budget.prompt_token_counter = Some(|prompt| prompt.split_whitespace().count());
        let long = std::iter::repeat_n("old", 80).collect::<Vec<_>>().join(" ");
        let history = [
            RunHistoryMessage::new("user", &long),
            RunHistoryMessage::new("assistant", &long),
        ];
        let plan = build_run_prompt_plan(
            &history,
            "demo",
            None,
            "tinyllama",
            None,
            &default_template_options(),
            &cmd,
            &budget,
        )
        .unwrap();
        assert_eq!(plan.kv_capacity, Some(64));
        assert_eq!(plan.dropped_history_turns, 1);
        assert_eq!(plan.dropped_history_messages, 2);
        assert_eq!(
            plan.prompt_tokens.unwrap() + plan.sampling_params.max_tokens,
            64
        );

        cmd.no_context_shift = true;
        assert!(build_run_prompt_plan(
            &history,
            "demo",
            None,
            "tinyllama",
            None,
            &default_template_options(),
            &cmd,
            &budget,
        )
        .is_err());
    }

    #[test]
    fn run_prompt_plan_retains_prompt_token_ids_for_observability() {
        let cmd = test_run_cmd();
        let budget = mapped_token_budget(64);
        let options = default_template_options();
        let plan = build_run_prompt_plan(
            &[],
            "demo prompt",
            None,
            "tinyllama",
            None,
            &options,
            &cmd,
            &budget,
        )
        .unwrap();

        let token_ids = plan
            .prompt_token_ids
            .as_ref()
            .expect("prompt token ids should be retained");
        assert_eq!(plan.prompt_tokens, Some(token_ids.len()));
        assert!(!token_ids.is_empty());
    }

    #[test]
    fn context_shift_drops_oldest_history_until_prompt_fits() {
        let cmd = test_run_cmd();
        let budget = whitespace_budget(64);
        let long = std::iter::repeat_n("old", 80).collect::<Vec<_>>().join(" ");
        let history = vec![
            RunHistoryMessage::new("user", &long),
            RunHistoryMessage::new("assistant", &long),
        ];
        let options = default_template_options();
        let plan = build_run_prompt_plan(
            &history,
            "demo",
            None,
            "tinyllama",
            None,
            &options,
            &cmd,
            &budget,
        )
        .unwrap();

        assert_eq!(plan.dropped_history_messages, 2);
        assert_eq!(plan.dropped_history_turns, 1);
        assert!(plan.prompt_tokens.unwrap() < 64);
    }

    #[test]
    fn context_shift_clamps_output_before_dropping_history() {
        let mut cmd = test_run_cmd();
        cmd.max_tokens = 1024;
        let budget = whitespace_budget(64);
        let history = vec![
            RunHistoryMessage::new("user", "Remember the identifier G00-c03-001-OK."),
            RunHistoryMessage::new("assistant", "ACKNOWLEDGED"),
        ];
        let options = default_template_options();
        let plan = build_run_prompt_plan(
            &history,
            "What identifier did I ask you to remember?",
            None,
            "tinyllama",
            None,
            &options,
            &cmd,
            &budget,
        )
        .unwrap();

        let prompt_tokens = plan.prompt_tokens.unwrap();
        assert!(prompt_tokens < 64);
        assert!(plan.prompt.contains("G00-c03-001-OK"));
        assert_eq!(plan.dropped_history_messages, 0);
        assert_eq!(plan.dropped_history_turns, 0);
        assert_eq!(plan.max_tokens_clamped_from, Some(1024));
        assert_eq!(plan.sampling_params.max_tokens, 64 - prompt_tokens);
    }

    #[test]
    fn no_context_shift_preserves_history_at_capacity_and_rejects_overflow() {
        let mut cmd = test_run_cmd();
        cmd.no_context_shift = true;
        cmd.max_tokens = 8;
        let history = vec![
            RunHistoryMessage::new("user", "Remember the identifier cobalt-731."),
            RunHistoryMessage::new("assistant", "I will remember cobalt-731."),
        ];
        let input = "What identifier did I ask you to remember?";
        let options = default_template_options();
        let template = ModelChatTemplate::new(
            "{% for message in messages %}{{ message.role }} {{ message.content }} {% endfor %}{% if add_generation_prompt %}assistant{% endif %}",
            "context-budget-fixture",
        );
        let full_prompt =
            build_chat_prompt(&history, input, None, "fixture", Some(&template), &options).unwrap();
        let prompt_tokens = full_prompt.split_whitespace().count();
        let output_tokens = usize::try_from(cmd.max_tokens).unwrap();
        let capacity = prompt_tokens + output_tokens;
        let budget = whitespace_budget(capacity);
        let plan = build_run_prompt_plan(
            &history,
            input,
            None,
            "fixture",
            Some(&template),
            &options,
            &cmd,
            &budget,
        )
        .expect("no-context-shift accepts the complete history at the exact boundary");

        assert_eq!(plan.prompt, full_prompt);
        assert!(plan.prompt.contains(&history[0].prompt.content));
        assert!(plan.prompt.contains(&history[1].prompt.content));
        assert_eq!(plan.prompt_tokens, Some(prompt_tokens));
        assert_eq!(plan.sampling_params.max_tokens, output_tokens);
        assert_eq!(plan.dropped_history_messages, 0);
        assert_eq!(plan.dropped_history_turns, 0);
        assert_eq!(plan.max_tokens_clamped_from, None);

        cmd.max_tokens += 1;
        assert!(
            build_run_prompt_plan(
                &history,
                input,
                None,
                "fixture",
                Some(&template),
                &options,
                &cmd,
                &budget,
            )
            .is_err(),
            "no-context-shift must not silently reduce the output budget"
        );

        assert!(
            build_run_prompt_plan(
                &history,
                input,
                None,
                "fixture",
                Some(&template),
                &options,
                &cmd,
                &whitespace_budget(prompt_tokens),
            )
            .is_err(),
            "no-context-shift must not discard history to create output space"
        );
    }

    #[test]
    fn kv_budget_accepts_request_inside_capacity() {
        assert!(fits_kv_budget(&default_params(512), Some(64), Some(2048)));
    }

    #[test]
    fn kv_budget_rejects_output_past_capacity() {
        assert!(!fits_kv_budget(&default_params(2048), Some(64), Some(2048)));
    }

    #[test]
    fn kv_budget_rejects_prompt_at_capacity() {
        assert!(!fits_kv_budget(&default_params(1), Some(2048), Some(2048)));
    }

    #[test]
    fn gpt_oss_run_parses_harmony_without_exposing_control_markers() {
        let parsed = parse_run_model_output(
            ModelOutputProtocol::HarmonyGptOss,
            "<|channel|>analysis<|message|>Reason.<|end|>\
             <|start|>assistant<|channel|>final<|message|>Answer.<|return|>",
            false,
            Some(FinishReason::Stop),
        )
        .unwrap();
        assert_eq!(parsed.content, "Answer.");
        assert_eq!(parsed.reasoning.as_deref(), Some("Reason."));
        assert!(!parsed.content.contains("<|"));
    }

    #[test]
    fn gpt_oss_run_rejects_harmony_tool_calls_without_a_tool_executor() {
        let error = parse_run_model_output(
            ModelOutputProtocol::HarmonyGptOss,
            "<|channel|>analysis<|message|>Need weather.<|end|>\
             <|start|>assistant<|channel|>commentary to=functions.weather\
             <|constrain|>json<|message|>{\"city\":\"Paris\"}<|call|>",
            false,
            Some(FinishReason::Stop),
        )
        .unwrap_err();
        assert!(error.to_string().contains("has no tool executor"));
    }

    #[test]
    fn gpt_oss_run_accepts_user_or_length_truncated_harmony_text() {
        for finish_reason in [FinishReason::Stop, FinishReason::Length] {
            let parsed = parse_run_model_output(
                ModelOutputProtocol::HarmonyGptOss,
                "<|channel|>final<|message|>Partial answer",
                false,
                Some(finish_reason),
            )
            .unwrap();
            assert_eq!(parsed.content, "Partial answer");
        }
    }

    #[test]
    fn gpt_oss_run_rejects_incomplete_harmony_after_natural_eos_or_unknown_finish() {
        for raw in [
            "<|channel|>final<|message|>Partial answer",
            "<|channel|>analysis<|message|>Still reasoning",
        ] {
            for finish_reason in [Some(FinishReason::EOS), None] {
                assert!(
                    parse_run_model_output(
                        ModelOutputProtocol::HarmonyGptOss,
                        raw,
                        false,
                        finish_reason,
                    )
                    .is_err(),
                    "accepted incomplete output {raw:?} for {finish_reason:?}"
                );
            }
        }
    }

    #[test]
    fn harmony_run_history_renders_the_next_turn_after_complete_or_truncated_output() {
        // The model's inference template rejects channel envelopes in content
        // and intentionally omits analysis from previous assistant turns.
        let mut template = ModelChatTemplate::new(
            r#"{%- for message in messages -%}
{%- if message.role == 'assistant' -%}
{%- if '<|channel|>analysis<|message|>' in message.content or '<|channel|>final<|message|>' in message.content -%}
{{- raise_exception('channel envelopes must not be passed as content') -}}
{%- endif -%}
{{- '<|start|>assistant<|channel|>final<|message|>' + message.content + '<|end|>' -}}
{%- else -%}
{{- '<|start|>' + message.role + '<|message|>' + message.content + '<|end|>' -}}
{%- endif -%}
{%- endfor -%}
{{- '<|start|>assistant' -}}"#,
            "harmony-history-contract",
        );
        template.output_protocol = ModelOutputProtocol::HarmonyGptOss;
        for (raw, finish_reason, content, reasoning) in [
            (
                "<|channel|>final<|message|>42<|return|>",
                FinishReason::Stop,
                "42",
                None,
            ),
            (
                "<|channel|>analysis<|message|>Add the numbers.<|end|>\
                 <|start|>assistant<|channel|>final<|message|>42<|return|>",
                FinishReason::EOS,
                "42",
                Some("Add the numbers."),
            ),
            (
                "<|channel|>final<|message|>Partial answer",
                FinishReason::Length,
                "Partial answer",
                None,
            ),
            (
                "<|channel|>analysis<|message|>Still reasoning",
                FinishReason::Length,
                "",
                Some("Still reasoning"),
            ),
            (
                "<|channel|>final<|message|>Partial answer",
                FinishReason::Stop,
                "Partial answer",
                None,
            ),
            (
                "<|channel|>analysis<|message|>Still reasoning",
                FinishReason::Stop,
                "",
                Some("Still reasoning"),
            ),
            (
                "<|channel|>analysis<|message|>Add the numbers.<|end|>\
                 <|start|>assistant<|channel|>final<|message|>Partial answer",
                FinishReason::Stop,
                "Partial answer",
                Some("Add the numbers."),
            ),
        ] {
            let parsed =
                parse_run_model_output(template.output_protocol, raw, false, Some(finish_reason))
                    .unwrap();
            let history = vec![
                RunHistoryMessage::new("user", "17+25?"),
                RunHistoryMessage::assistant(raw, template.output_protocol, &parsed),
            ];
            let plan = build_run_prompt_plan(
                &history,
                "Continue.",
                None,
                "model-with-declared-protocol",
                Some(&template),
                &ChatTemplateOptions::default(),
                &test_run_cmd(),
                &whitespace_budget(8192),
            )
            .expect("validated assistant output must render as history on the next turn");
            assert_eq!(
                plan.prompt,
                format!(
                    "<|start|>user<|message|>17+25?<|end|>\
                     <|start|>assistant<|channel|>final<|message|>{content}<|end|>\
                     <|start|>user<|message|>Continue.<|end|><|start|>assistant"
                )
            );
            assert_eq!(history[1].prompt.reasoning_content.as_deref(), reasoning);
            assert_eq!(history[1].raw_content, raw);
            let original_history = [("user", "17+25?"), ("assistant", raw)];
            assert_eq!(
                history_evidence(&history)["sha256"],
                sha256_text(&serde_json::to_string(&original_history).unwrap()),
                "JSONL history evidence must continue to bind the raw generated output"
            );
            let record = jsonl_assistant_record(
                "session",
                0,
                "request",
                0,
                &parsed.content,
                parsed.reasoning.as_deref(),
                &[],
                Some(finish_reason),
                None,
                1,
                1,
                raw,
                1.0,
            );
            assert_eq!(record["content"], content);
            assert_eq!(record["raw_text_sha256"], sha256_text(raw));
            assert_eq!(record["finish_reason"], finish_reason_str(finish_reason));
        }
    }

    #[test]
    fn gemma_run_history_uses_typed_content_and_keeps_raw_evidence() {
        // Gemma's template omits prior-turn reasoning after a new user turn,
        // while accepting a separate reasoning field for same-turn continuation.
        let mut template = ModelChatTemplate::new(
            concat!(
                "{%- set ns = namespace(last_user=-1) -%}",
                "{%- for message in messages -%}",
                "{%- if message.role == 'user' -%}{%- set ns.last_user = loop.index0 -%}{%- endif -%}",
                "{%- endfor -%}",
                "{%- for message in messages -%}",
                "{{- '<|turn>' + ('model' if message.role == 'assistant' else message.role) + '\n' -}}",
                "{%- if message.reasoning_content and loop.index0 > ns.last_user -%}",
                "{{- '<|channel>thought\n' + message.reasoning_content + '<channel|>' -}}",
                "{%- endif -%}",
                "{{- message.content + '<turn|>\n' -}}",
                "{%- endfor -%}",
                "{{- '<|turn>model\n<|channel>thought\n<channel|>' -}}",
            ),
            "gemma-thought-history-contract",
        );
        template.set_output_protocol(ModelOutputProtocol::GemmaThought);
        for (raw, opened, content, reasoning) in [
            ("<|channel>thought\n<channel|>579", false, "579", None),
            (
                "<|channel>thought\nUse the sum.<channel|>579",
                false,
                "579",
                Some("Use the sum."),
            ),
            (
                "Use the sum.<channel|>579",
                true,
                "579",
                Some("Use the sum."),
            ),
            (
                "<|channel>thought\nStill computing.",
                false,
                "",
                Some("Still computing."),
            ),
        ] {
            let parsed = parse_run_model_output(
                template.output_protocol,
                raw,
                opened,
                Some(FinishReason::Length),
            )
            .unwrap();
            let history = vec![
                RunHistoryMessage::new("user", "123+456?"),
                RunHistoryMessage::assistant(raw, template.output_protocol, &parsed),
            ];
            assert_eq!(history[1].prompt.content, content);
            assert_eq!(history[1].prompt.reasoning_content.as_deref(), reasoning);
            assert_eq!(history[1].raw_content, raw);
            let plan = build_run_prompt_plan(
                &history,
                "Continue.",
                None,
                "loaded-model-alias",
                Some(&template),
                &ChatTemplateOptions::default(),
                &test_run_cmd(),
                &whitespace_budget(8192),
            )
            .unwrap();
            assert_eq!(
                plan.prompt,
                format!(
                    "<|turn>user\n123+456?<turn|>\n<|turn>model\n{content}<turn|>\n\
                     <|turn>user\nContinue.<turn|>\n<|turn>model\n<|channel>thought\n<channel|>"
                )
            );
            let raw_history = [("user", "123+456?"), ("assistant", raw)];
            assert_eq!(
                history_evidence(&history)["sha256"],
                sha256_text(&serde_json::to_string(&raw_history).unwrap())
            );
        }
    }

    #[test]
    fn text_run_history_leaves_reasoning_policy_to_the_template() {
        let template = ModelChatTemplate::new(
            "{% for message in messages %}{{ message.content }}{% endfor %}",
            "verbatim-history-template",
        );
        for raw in [
            "<think>Reason.</think>Answer.",
            "Reason.</think>Answer.",
            "Plain answer.",
        ] {
            let parsed = parse_run_model_output(
                ModelOutputProtocol::Text,
                raw,
                true,
                Some(FinishReason::Stop),
            )
            .unwrap();
            let history = [RunHistoryMessage::assistant(
                raw,
                ModelOutputProtocol::Text,
                &parsed,
            )];
            let prompt = build_chat_prompt(
                &history,
                "Continue.",
                None,
                "text-protocol-model",
                Some(&template),
                &ChatTemplateOptions::default(),
            )
            .unwrap();
            assert_eq!(prompt, format!("{raw}Continue."));
            assert!(history[0].prompt.reasoning_content.is_none());
        }
    }

    #[test]
    fn sibling_repo_strips_gguf_suffix_by_default() {
        assert_eq!(
            tokenizer_sibling_repo("Qwen/Qwen3-0.6B-GGUF").as_deref(),
            Some("Qwen/Qwen3-0.6B")
        );
        assert_eq!(tokenizer_sibling_repo("Qwen/Qwen3-0.6B"), None);
    }

    #[test]
    fn sibling_repo_explicit_mappings_beat_strip_convention() {
        // bartowski/* have no safetensors mirrors; stripping `-GGUF`
        // would point at repos that don't exist.
        assert_eq!(
            tokenizer_sibling_repo("bartowski/Qwen2.5-Coder-32B-Instruct-GGUF").as_deref(),
            Some("Qwen/Qwen2.5-Coder-32B-Instruct")
        );
        // mistralai upstream ships tekken-format tokenizers only.
        assert_eq!(
            tokenizer_sibling_repo("bartowski/mistralai_Mistral-Small-3.2-24B-Instruct-2506-GGUF")
                .as_deref(),
            Some("unsloth/Mistral-Small-3.2-24B-Instruct-2506")
        );
        // meta-llama upstream is gated; unsloth mirror is not.
        assert_eq!(
            tokenizer_sibling_repo("bartowski/Meta-Llama-3.1-8B-Instruct-GGUF").as_deref(),
            Some("unsloth/Meta-Llama-3.1-8B-Instruct")
        );
    }
}