tokio_prompt_orchestrator/config/

mod.rs

//! # Stage: Declarative Pipeline Configuration
//!
//! ## Responsibility
//! Parse, validate, and hot-reload TOML pipeline configuration files.
//! Users define an entire pipeline topology declaratively and run it with:
//! ```text
//! cargo run -- --config pipeline.toml
//! ```
//!
//! ## Guarantees
//! - Deterministic: same TOML input always produces the same `PipelineConfig`
//! - Validated: all semantic constraints are checked before a config is accepted
//! - Type-safe: invalid field combinations are caught at parse time via serde
//! - Hot-reloadable: file changes are detected and validated before applying
//! - Schema-exportable: JSON Schema output enables IDE autocomplete
//!
//! ## NOT Responsible For
//! - Building the runtime pipeline from config (that belongs to `stages`)
//! - Managing worker connections (that belongs to `worker`)
//! - Metrics collection (that belongs to `metrics`)

pub mod loader;
pub mod validation;
pub mod watcher;

#[cfg(feature = "schema")]
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

// ── Default value functions ──────────────────────────────────────────────

/// Default RAG stage timeout: 5000 ms.
///
/// **Why 5 s?** Covers P99 of most retrieval backends (vector DBs, BM25 indexes)
/// under normal load while still bounding worst-case pipeline latency.  Increase
/// for slow cross-region retrieval or when using large embedding models for
/// re-ranking.
fn default_timeout_ms() -> u64 {
    5000
}

/// Default maximum context tokens for the RAG stage: 2048 tokens.
///
/// **Why 2048?** Fits comfortably in the context window of all major models
/// (≥4K context) while leaving ample room for the instruction, user prompt,
/// and model response.  Increase to 4096–8192 for RAG-heavy workloads where
/// retrieved passages tend to be long.
fn default_max_context_tokens() -> usize {
    2048
}

/// Default retry base delay: 100 ms.
///
/// **Why 100 ms?** With exponential back-off (2×) this reaches ~3 s by
/// attempt 5, covering most transient LLM API failures (rate-limit windows,
/// brief network blips) without introducing multi-second latency on the first
/// retry.  Reduce to 50 ms for latency-sensitive workflows; increase for
/// providers with aggressive rate limits.
fn default_retry_base_ms() -> u64 {
    100
}

/// Default retry maximum delay: 5000 ms.
///
/// **Why 5 s?** Caps the exponential back-off ceiling so no single retry ever
/// blocks a pipeline stage for more than 5 s.  This balances giving slow
/// providers time to recover with keeping end-to-end P99 latency predictable.
fn default_retry_max_ms() -> u64 {
    5000
}

/// Default deduplication window: 300 seconds (5 minutes).
///
/// **Why 5 minutes?** Covers typical LLM retry storms — clients that retry on
/// error within a few minutes will hit the in-flight dedup cache rather than
/// issuing a duplicate API call.  Longer windows reduce redundant API costs
/// but proportionally increase memory use; at 10K entries × 5 min the overhead
/// is negligible.
fn default_dedup_window_s() -> u64 {
    300
}

/// Default deduplication max entries: 10 000.
///
/// **Why 10 000?** At ~256 bytes per entry (key hash + response pointer) this
/// costs ~2.5 MB of heap, which is acceptable even in memory-constrained
/// deployments.  For high-cardinality workloads (many unique sessions) increase
/// to 50K–100K; for embedded deployments reduce to 1K–5K.
fn default_dedup_max_entries() -> usize {
    10_000
}

/// Default channel capacity for pipeline stages: 512 entries.
///
/// **Why 512?** Sized for ~50 ms of burst traffic at 10 000 req/s while
/// keeping per-stage memory under ~4 MB (each `PromptRequest` ≈ 8 KB).
/// This gives downstream stages enough headroom to absorb a slow GC pause or
/// brief processing spike without shedding load.  Reduce to 128–256 for
/// memory-constrained deployments; increase to 1024–2048 for very high
/// throughput or bursty workloads.
fn default_channel_capacity() -> usize {
    512
}

/// Default enabled state: `true`.
///
/// **Why true?** Most stages should be on by default so that a minimal config
/// file (with only required fields) produces a fully-functional pipeline.
/// Stages that should be off by default (rate limiting) have their own
/// `default_*_enabled` functions that return `false`.
fn default_true() -> bool {
    true
}

// ── Channel sizes ────────────────────────────────────────────────────────

/// Per-channel capacity overrides for the five inter-stage pipeline channels.
///
/// All fields are optional; when absent the stage default is used
/// (512, 512, 1024, 512, 256 respectively).
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
pub struct ChannelSizes {
    /// Capacity of the RAG → Assemble channel (default: 512).
    pub rag_to_assemble: Option<usize>,
    /// Capacity of the Assemble → Inference channel (default: 512).
    pub assemble_to_inference: Option<usize>,
    /// Capacity of the Inference → Post channel (default: 1024).
    pub inference_to_post: Option<usize>,
    /// Capacity of the Post → Stream channel (default: 512).
    pub post_to_stream: Option<usize>,
    /// Capacity of the Stream output channel (default: 256).
    pub stream_output: Option<usize>,
}

// ── Top-level config ─────────────────────────────────────────────────────

/// Root configuration for a pipeline instance.
///
/// Deserialized from a TOML file and validated before use.
/// Every field has either a required value or a documented default.
///
/// # Example
///
/// ```toml
/// [pipeline]
/// name = "production"
/// version = "1.0"
///
/// [stages.inference]
/// worker = "open_ai"
/// model = "gpt-4"
/// ```
///
/// # Panics
///
/// This type never panics during construction or access.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct PipelineConfig {
    /// Pipeline identity and version metadata.
    pub pipeline: PipelineSection,
    /// Per-stage configuration for the five pipeline stages.
    pub stages: StagesConfig,
    /// Resilience settings: retries, circuit breaker, backpressure.
    pub resilience: ResilienceConfig,
    /// Rate limiting settings to control request throughput.
    #[serde(default)]
    pub rate_limits: RateLimitConfig,
    /// Deduplication settings for request coalescing.
    pub deduplication: DeduplicationConfig,
    /// Observability: logging, metrics, tracing.
    pub observability: ObservabilityConfig,
    /// Optional per-channel capacity overrides.  When absent, each stage uses
    /// its compiled-in default (512 / 512 / 1024 / 512 / 256).
    #[serde(default)]
    pub channel_sizes: Option<ChannelSizes>,
    /// Optional per-worker configuration overrides.
    ///
    /// Allows setting model, API base URL, and connection parameters for each
    /// worker type directly in the pipeline TOML without modifying worker code.
    /// Workers read these values at construction time in `spawn_pipeline_with_config`.
    #[serde(default)]
    pub workers: WorkersConfig,
}

// ── Pipeline identity ────────────────────────────────────────────────────

/// Pipeline identity and version metadata.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct PipelineSection {
    /// Human-readable pipeline name (e.g., "production", "staging").
    pub name: String,
    /// Semantic version of this configuration (e.g., "1.0").
    pub version: String,
    /// Optional description for documentation purposes.
    pub description: Option<String>,
}

// ── Stage configs ────────────────────────────────────────────────────────

/// Configuration for all five pipeline stages.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct StagesConfig {
    /// RAG (retrieval-augmented generation) stage settings.
    pub rag: RagStageConfig,
    /// Prompt assembly stage settings.
    pub assemble: AssembleStageConfig,
    /// Model inference stage settings.
    pub inference: InferenceStageConfig,
    /// Post-processing stage settings.
    pub post_process: PostProcessStageConfig,
    /// Output streaming stage settings.
    pub stream: StreamStageConfig,
}

/// RAG stage configuration.
///
/// Controls retrieval timeout, context limits, and channel sizing.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct RagStageConfig {
    /// Whether the RAG stage is enabled. Disabled stages pass-through.
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Maximum time (ms) to wait for retrieval results.
    #[serde(default = "default_timeout_ms")]
    pub timeout_ms: u64,
    /// Maximum context tokens to prepend from retrieval.
    #[serde(default = "default_max_context_tokens")]
    pub max_context_tokens: usize,
    /// Channel buffer capacity for this stage. `None` uses the pipeline default (512).
    pub channel_capacity: Option<usize>,
}

/// Assembly stage configuration.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct AssembleStageConfig {
    /// Whether the assembly stage is enabled.
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Channel buffer capacity for this stage.
    #[serde(default = "default_channel_capacity")]
    pub channel_capacity: usize,
}

/// Default number of concurrent inference workers: 1 (single worker).
///
/// **Why 1?** Preserves backward compatibility with existing deployments that
/// assume a single ordered inference queue.  Increase to the number of
/// parallel in-flight requests you want to sustain (e.g. 4–8 for high-
/// throughput deployments with async LLM APIs that support concurrent calls).
fn default_inference_workers() -> usize {
    1
}

/// Default adaptive timeout minimum floor: 500 ms.
///
/// **Why 500 ms?** This is a conservative lower bound that prevents the
/// adaptive algorithm from producing a timeout so short that it triggers
/// spurious failures during brief periods of low-latency traffic.  Cloud LLM
/// APIs often have a non-trivial connection establishment cost (~100–300 ms)
/// that would be mis-classified as a timeout at a sub-500 ms floor.  Operators
/// running local inference servers with sub-100 ms P99 should lower this to
/// match their actual minimum latency.
fn default_adaptive_timeout_min_ms() -> u64 {
    500
}

/// Inference stage configuration.
///
/// Specifies the worker backend, model, and generation parameters.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct InferenceStageConfig {
    /// Which model worker backend to use.
    pub worker: WorkerKind,
    /// Model name or identifier (e.g., "gpt-4", "claude-3-opus").
    pub model: String,
    /// Maximum tokens to generate. `None` uses the worker default.
    pub max_tokens: Option<u32>,
    /// Sampling temperature. `None` uses the worker default.
    pub temperature: Option<f32>,
    /// Inference timeout in milliseconds. `None` uses no explicit timeout.
    pub timeout_ms: Option<u64>,
    /// Number of concurrent inference worker tasks reading from the same
    /// input channel.  Default 1 (backward-compatible).  Increase to add
    /// parallelism for high-throughput deployments.
    #[serde(default = "default_inference_workers")]
    pub inference_workers: usize,
    /// Enable adaptive timeout based on P95 of recent inferences. Overrides timeout_ms if set.
    #[serde(default)]
    pub adaptive_timeout_enabled: bool,
    /// Minimum adaptive timeout floor in milliseconds (only used when adaptive_timeout_enabled = true).
    #[serde(default = "default_adaptive_timeout_min_ms")]
    pub adaptive_timeout_min_ms: u64,
}

/// Supported model worker backends.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
#[serde(rename_all = "snake_case")]
pub enum WorkerKind {
    /// OpenAI-compatible API (GPT-4, GPT-3.5, etc.).
    OpenAi,
    /// Anthropic Claude API.
    Anthropic,
    /// Local llama.cpp server.
    LlamaCpp,
    /// vLLM inference server.
    Vllm,
    /// Echo worker for testing — returns the prompt as tokens.
    Echo,
}

/// Provider identifier enum used in per-worker configuration.
///
/// Accepts the following string values in TOML (case-insensitive via custom
/// `Deserialize`): `"openai"`, `"anthropic"`, `"llama_cpp"` / `"llama-cpp"`,
/// `"vllm"`, `"echo"`.  Unknown values produce a clear serde error at
/// parse time.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
pub enum Provider {
    /// OpenAI-compatible API.
    OpenAi,
    /// Anthropic Claude API.
    Anthropic,
    /// Local llama.cpp server.
    LlamaCpp,
    /// vLLM inference server.
    Vllm,
    /// Echo (test) worker.
    Echo,
}

impl<'de> serde::Deserialize<'de> for Provider {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        match s.to_lowercase().replace('-', "_").as_str() {
            "openai" => Ok(Provider::OpenAi),
            "anthropic" => Ok(Provider::Anthropic),
            "llama_cpp" => Ok(Provider::LlamaCpp),
            "vllm" => Ok(Provider::Vllm),
            "echo" => Ok(Provider::Echo),
            other => Err(serde::de::Error::custom(format!(
                "unknown provider {other:?}; expected one of: openai, anthropic, llama_cpp, vllm, echo"
            ))),
        }
    }
}

/// Per-worker connection and model configuration.
///
/// Optional overrides for a named worker. All fields are optional; when absent
/// the worker falls back to its built-in defaults or environment variables.
///
/// Workers prefer TOML config over defaults but environment variables still
/// override TOML values (e.g. `OPENAI_API_KEY` always takes precedence over
/// any key stored in the config file).
///
/// # Example
///
/// ```toml
/// [workers.openai]
/// model = "gpt-4o"
/// api_base_url = "https://api.openai.com/v1"
///
/// [workers.anthropic]
/// model = "claude-opus-4-6"
/// max_tokens = 4096
///
/// [workers.llama_cpp]
/// host = "localhost"
/// port = 8080
/// ```
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[derive(Default)]
pub struct WorkerConfig {
    /// Model name to use (e.g. `"gpt-4o"`, `"claude-opus-4-6"`).
    pub model: Option<String>,
    /// API base URL (e.g. `"https://api.openai.com/v1"`).
    /// Overrides the worker's built-in default endpoint.
    pub api_base_url: Option<String>,
    /// Maximum tokens to generate.  `None` uses the worker default.
    pub max_tokens: Option<u32>,
    /// Sampling temperature.  `None` uses the worker default.
    pub temperature: Option<f32>,
    /// Hostname for local servers (llama.cpp / vLLM).
    pub host: Option<String>,
    /// Port for local servers (llama.cpp / vLLM).
    pub port: Option<u16>,
}


/// Per-worker configuration map, keyed by provider name.
///
/// Used in the `[workers.*]` TOML sections.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
pub struct WorkersConfig {
    /// OpenAI worker configuration overrides.
    #[serde(default)]
    pub openai: WorkerConfig,
    /// Anthropic worker configuration overrides.
    #[serde(default)]
    pub anthropic: WorkerConfig,
    /// llama.cpp worker configuration overrides.
    #[serde(default)]
    pub llama_cpp: WorkerConfig,
    /// vLLM worker configuration overrides.
    #[serde(default)]
    pub vllm: WorkerConfig,
    /// Echo worker configuration overrides (mostly useful for testing).
    #[serde(default)]
    pub echo: WorkerConfig,
}

/// Post-processing stage configuration.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct PostProcessStageConfig {
    /// Whether post-processing is enabled.
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Channel buffer capacity for this stage.
    #[serde(default = "default_channel_capacity")]
    pub channel_capacity: usize,
}

/// Output streaming stage configuration.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct StreamStageConfig {
    /// Whether output streaming is enabled.
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Channel buffer capacity for this stage.
    #[serde(default = "default_channel_capacity")]
    pub channel_capacity: usize,
}

// ── Rate limiting ────────────────────────────────────────────────────────

/// Default requests-per-second limit: 100 req/s.
///
/// **Why 100?** A conservative starting point that prevents accidental
/// saturation of the downstream LLM provider (most have free-tier limits of
/// ~60 req/min ≈ 1 req/s).  Operators deploying against higher-tier API
/// accounts should raise this to their actual quota.
fn default_rps() -> u32 {
    100
}

/// Default burst capacity above the sustained rate: 20 requests.
///
/// **Why 20?** Allows a short burst (e.g. a web page loading many parallel
/// requests) without immediately triggering the rate limiter, while still
/// preventing sustained abuse.  Equivalent to ~200 ms of sustained quota at
/// the 100 req/s default.
fn default_burst() -> u32 {
    20
}

/// Default enabled state for rate limiting: `false`.
///
/// **Why false?** Rate limiting has no sensible universal default — the right
/// limit depends entirely on the provider contract.  Enabling it by default
/// with wrong values would silently reject legitimate traffic.  Operators must
/// explicitly set `enabled = true` and configure `requests_per_second`.
fn default_rate_limit_enabled() -> bool {
    false
}

/// Rate limiting configuration.
///
/// Controls the token-bucket rate limiter that caps inbound request throughput.
/// When enabled, requests exceeding the allowed rate are rejected with HTTP 429.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct RateLimitConfig {
    /// Whether rate limiting is enabled.
    #[serde(default = "default_rate_limit_enabled")]
    pub enabled: bool,
    /// Maximum sustained requests per second (token refill rate).
    #[serde(default = "default_rps")]
    pub requests_per_second: u32,
    /// Maximum burst capacity above the sustained rate.
    #[serde(default = "default_burst")]
    pub burst_capacity: u32,
}

impl Default for RateLimitConfig {
    fn default() -> Self {
        Self {
            enabled: default_rate_limit_enabled(),
            requests_per_second: default_rps(),
            burst_capacity: default_burst(),
        }
    }
}

// ── Resilience ───────────────────────────────────────────────────────────

/// Resilience configuration for retries and circuit breaking.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct ResilienceConfig {
    /// Maximum number of retry attempts before failing a request.
    pub retry_attempts: u32,
    /// Base delay (ms) for exponential backoff. Must be ≤ `retry_max_ms`.
    #[serde(default = "default_retry_base_ms")]
    pub retry_base_ms: u64,
    /// Maximum delay (ms) cap for exponential backoff.
    #[serde(default = "default_retry_max_ms")]
    pub retry_max_ms: u64,
    /// Number of consecutive failures before the circuit breaker opens.
    pub circuit_breaker_threshold: u32,
    /// Seconds to keep the circuit breaker open before allowing a probe.
    pub circuit_breaker_timeout_s: u64,
    /// Required success rate (0.0–1.0) over the sliding window.
    pub circuit_breaker_success_rate: f64,
}

// ── Deduplication ────────────────────────────────────────────────────────

/// Deduplication configuration for request coalescing.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct DeduplicationConfig {
    /// Whether deduplication is enabled.
    pub enabled: bool,
    /// Time window (seconds) within which duplicate requests are coalesced.
    #[serde(default = "default_dedup_window_s")]
    pub window_s: u64,
    /// Maximum number of dedup entries held in memory.
    #[serde(default = "default_dedup_max_entries")]
    pub max_entries: usize,
}

// ── Observability ────────────────────────────────────────────────────────

/// Observability configuration: logging, metrics endpoint, and tracing.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
pub struct ObservabilityConfig {
    /// Log output format.
    pub log_format: LogFormat,
    /// Port for the Prometheus metrics HTTP endpoint. `None` disables it.
    pub metrics_port: Option<u16>,
    /// OpenTelemetry tracing collector endpoint. `None` disables distributed tracing.
    pub tracing_endpoint: Option<String>,
}

/// Log output format.
///
/// # Panics
///
/// This type never panics.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(JsonSchema))]
#[serde(deny_unknown_fields)]
#[serde(rename_all = "snake_case")]
pub enum LogFormat {
    /// Human-readable, colorized log output.
    Pretty,
    /// Structured JSON log output for machine consumption.
    Json,
}

/// Export the JSON Schema for `PipelineConfig`.
///
/// This enables IDE autocomplete when editing TOML config files.
///
/// # Errors
///
/// Returns `serde_json::Error` if schema serialization fails (should not
/// happen with well-formed derive macros).
///
/// # Panics
///
/// This function never panics.
#[cfg(feature = "schema")]
pub fn export_schema() -> Result<String, serde_json::Error> {
    let schema = schemars::schema_for!(PipelineConfig);
    serde_json::to_string_pretty(&schema)
}

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

    #[test]
    fn test_default_timeout_ms_returns_5000() {
        assert_eq!(default_timeout_ms(), 5000);
    }

    #[test]
    fn test_default_max_context_tokens_returns_2048() {
        assert_eq!(default_max_context_tokens(), 2048);
    }

    #[test]
    fn test_default_retry_base_ms_returns_100() {
        assert_eq!(default_retry_base_ms(), 100);
    }

    #[test]
    fn test_default_retry_max_ms_returns_5000() {
        assert_eq!(default_retry_max_ms(), 5000);
    }

    #[test]
    fn test_default_dedup_window_s_returns_300() {
        assert_eq!(default_dedup_window_s(), 300);
    }

    #[test]
    fn test_default_dedup_max_entries_returns_10000() {
        assert_eq!(default_dedup_max_entries(), 10_000);
    }

    #[test]
    fn test_default_channel_capacity_returns_512() {
        assert_eq!(default_channel_capacity(), 512);
    }

    #[test]
    fn test_default_true_returns_true() {
        assert!(default_true());
    }

    #[test]
    fn test_worker_kind_serializes_to_snake_case() {
        let json = serde_json::to_string(&WorkerKind::OpenAi).expect("test: serialization");
        assert_eq!(json, "\"open_ai\"");
    }

    #[test]
    fn test_worker_kind_deserializes_from_snake_case() {
        let kind: WorkerKind =
            serde_json::from_str("\"llama_cpp\"").expect("test: deserialization");
        assert_eq!(kind, WorkerKind::LlamaCpp);
    }

    #[test]
    fn test_log_format_serializes_to_snake_case() {
        let json = serde_json::to_string(&LogFormat::Pretty).expect("test: serialization");
        assert_eq!(json, "\"pretty\"");
    }

    #[test]
    fn test_log_format_deserializes_from_snake_case() {
        let fmt: LogFormat = serde_json::from_str("\"json\"").expect("test: deserialization");
        assert_eq!(fmt, LogFormat::Json);
    }

    #[cfg(feature = "schema")]
    #[test]
    fn test_export_schema_produces_valid_json() {
        let schema = export_schema().expect("test: schema export");
        let parsed: serde_json::Value =
            serde_json::from_str(&schema).expect("test: schema is valid JSON");
        // Should contain top-level properties
        assert!(parsed.get("properties").is_some() || parsed.get("$ref").is_some());
    }

    #[test]
    fn test_pipeline_config_minimal_toml_parses() {
        let toml_str = r#"
[pipeline]
name = "test"
version = "1.0"

[stages.rag]
enabled = true

[stages.assemble]
enabled = true

[stages.inference]
worker = "echo"
model = "test-model"

[stages.post_process]
enabled = true

[stages.stream]
enabled = true

[resilience]
retry_attempts = 3
circuit_breaker_threshold = 5
circuit_breaker_timeout_s = 60
circuit_breaker_success_rate = 0.8

[deduplication]
enabled = false

[observability]
log_format = "pretty"
"#;
        let config: PipelineConfig = toml::from_str(toml_str).expect("test: minimal TOML parses");
        assert_eq!(config.pipeline.name, "test");
        assert_eq!(config.stages.inference.worker, WorkerKind::Echo);
        assert_eq!(config.resilience.retry_base_ms, 100); // default applied
        assert!(!config.deduplication.enabled);
    }

    #[test]
    fn test_pipeline_config_full_toml_parses() {
        let toml_str = r#"
[pipeline]
name = "production"
version = "1.0"
description = "Production pipeline with OpenAI GPT-4"

[stages.rag]
enabled = true
timeout_ms = 5000
max_context_tokens = 2048

[stages.assemble]
enabled = true
channel_capacity = 256

[stages.inference]
worker = "open_ai"
model = "gpt-4"
max_tokens = 1024
temperature = 0.7
timeout_ms = 30000

[stages.post_process]
enabled = true

[stages.stream]
enabled = true

[resilience]
retry_attempts = 3
retry_base_ms = 100
retry_max_ms = 5000
circuit_breaker_threshold = 5
circuit_breaker_timeout_s = 60
circuit_breaker_success_rate = 0.8

[deduplication]
enabled = true
window_s = 300
max_entries = 10000

[observability]
log_format = "json"
metrics_port = 9090
"#;
        let config: PipelineConfig = toml::from_str(toml_str).expect("test: full TOML parses");
        assert_eq!(config.pipeline.name, "production");
        assert_eq!(config.stages.inference.worker, WorkerKind::OpenAi);
        assert_eq!(config.stages.inference.temperature, Some(0.7));
        assert_eq!(config.stages.assemble.channel_capacity, 256);
        assert_eq!(config.observability.metrics_port, Some(9090));
    }

    #[test]
    fn test_pipeline_config_serialize_deserialize_roundtrip() {
        let config = PipelineConfig {
            pipeline: PipelineSection {
                name: "roundtrip".into(),
                version: "2.0".into(),
                description: Some("Roundtrip test".into()),
            },
            stages: StagesConfig {
                rag: RagStageConfig {
                    enabled: true,
                    timeout_ms: 3000,
                    max_context_tokens: 1024,
                    channel_capacity: Some(256),
                },
                assemble: AssembleStageConfig {
                    enabled: true,
                    channel_capacity: 512,
                },
                inference: InferenceStageConfig {
                    worker: WorkerKind::Anthropic,
                    model: "claude-3-opus".into(),
                    max_tokens: Some(2048),
                    temperature: Some(0.5),
                    timeout_ms: Some(60000),
                    inference_workers: 1,
                    adaptive_timeout_enabled: false,
                    adaptive_timeout_min_ms: 500,
                },
                post_process: PostProcessStageConfig {
                    enabled: true,
                    channel_capacity: 512,
                },
                stream: StreamStageConfig {
                    enabled: false,
                    channel_capacity: 128,
                },
            },
            resilience: ResilienceConfig {
                retry_attempts: 5,
                retry_base_ms: 200,
                retry_max_ms: 10000,
                circuit_breaker_threshold: 10,
                circuit_breaker_timeout_s: 120,
                circuit_breaker_success_rate: 0.9,
            },
            deduplication: DeduplicationConfig {
                enabled: true,
                window_s: 600,
                max_entries: 20000,
            },
            observability: ObservabilityConfig {
                log_format: LogFormat::Json,
                metrics_port: Some(8080),
                tracing_endpoint: Some("http://jaeger:14268".into()),
            },
            rate_limits: RateLimitConfig::default(),
            channel_sizes: None,
            workers: WorkersConfig::default(),
        };

        let toml_str = toml::to_string_pretty(&config).expect("test: serialize to TOML");
        let deserialized: PipelineConfig =
            toml::from_str(&toml_str).expect("test: deserialize from TOML");
        assert_eq!(config, deserialized);
    }

    #[test]
    fn test_pipeline_config_json_roundtrip() {
        let config = PipelineConfig {
            pipeline: PipelineSection {
                name: "json-rt".into(),
                version: "1.0".into(),
                description: None,
            },
            stages: StagesConfig {
                rag: RagStageConfig {
                    enabled: true,
                    timeout_ms: 5000,
                    max_context_tokens: 2048,
                    channel_capacity: None,
                },
                assemble: AssembleStageConfig {
                    enabled: true,
                    channel_capacity: 512,
                },
                inference: InferenceStageConfig {
                    worker: WorkerKind::Echo,
                    model: "echo".into(),
                    max_tokens: None,
                    temperature: None,
                    timeout_ms: None,
                    inference_workers: 1,
                    adaptive_timeout_enabled: false,
                    adaptive_timeout_min_ms: 500,
                },
                post_process: PostProcessStageConfig {
                    enabled: true,
                    channel_capacity: 512,
                },
                stream: StreamStageConfig {
                    enabled: true,
                    channel_capacity: 512,
                },
            },
            resilience: ResilienceConfig {
                retry_attempts: 3,
                retry_base_ms: 100,
                retry_max_ms: 5000,
                circuit_breaker_threshold: 5,
                circuit_breaker_timeout_s: 60,
                circuit_breaker_success_rate: 0.8,
            },
            deduplication: DeduplicationConfig {
                enabled: false,
                window_s: 300,
                max_entries: 10000,
            },
            observability: ObservabilityConfig {
                log_format: LogFormat::Pretty,
                metrics_port: None,
                tracing_endpoint: None,
            },
            rate_limits: RateLimitConfig::default(),
            channel_sizes: None,
            workers: WorkersConfig::default(),
        };

        let json = serde_json::to_string(&config).expect("test: serialize to JSON");
        let deserialized: PipelineConfig =
            serde_json::from_str(&json).expect("test: deserialize from JSON");
        assert_eq!(config, deserialized);
    }

    #[test]
    fn test_all_worker_kinds_roundtrip_toml() {
        // TOML requires a table wrapper for enum serialization
        #[derive(Debug, Serialize, Deserialize, PartialEq)]
        struct Wrapper {
            kind: WorkerKind,
        }

        let kinds = vec![
            WorkerKind::OpenAi,
            WorkerKind::Anthropic,
            WorkerKind::LlamaCpp,
            WorkerKind::Vllm,
            WorkerKind::Echo,
        ];
        for kind in kinds {
            let w = Wrapper { kind: kind.clone() };
            let s = toml::to_string(&w).expect("test: serialize worker kind");
            let deserialized: Wrapper = toml::from_str(&s).expect("test: deserialize worker kind");
            assert_eq!(w, deserialized);
        }
    }

    #[test]
    fn test_pipeline_section_optional_description_omitted() {
        let toml_str = r#"
name = "no-desc"
version = "1.0"
"#;
        let section: PipelineSection =
            toml::from_str(toml_str).expect("test: parse without description");
        assert!(section.description.is_none());
    }

    #[test]
    fn test_rag_stage_defaults_applied_when_omitted() {
        let toml_str = r#"
enabled = true
"#;
        let rag: RagStageConfig = toml::from_str(toml_str).expect("test: parse with defaults");
        assert_eq!(rag.timeout_ms, 5000);
        assert_eq!(rag.max_context_tokens, 2048);
        assert!(rag.channel_capacity.is_none());
    }

    #[test]
    fn test_resilience_defaults_applied_when_omitted() {
        let toml_str = r#"
retry_attempts = 3
circuit_breaker_threshold = 5
circuit_breaker_timeout_s = 60
circuit_breaker_success_rate = 0.8
"#;
        let resilience: ResilienceConfig =
            toml::from_str(toml_str).expect("test: parse with defaults");
        assert_eq!(resilience.retry_base_ms, 100);
        assert_eq!(resilience.retry_max_ms, 5000);
    }

    #[test]
    fn test_deduplication_defaults_applied_when_omitted() {
        let toml_str = r#"
enabled = true
"#;
        let dedup: DeduplicationConfig =
            toml::from_str(toml_str).expect("test: parse with defaults");
        assert_eq!(dedup.window_s, 300);
        assert_eq!(dedup.max_entries, 10_000);
    }

    #[test]
    fn test_unknown_field_in_resilience_is_rejected() {
        // deny_unknown_fields must fire so typos in TOML are caught at parse time
        let bad_toml = r#"
retry_attempts = 3
circuit_breaker_threshold = 5
circuit_breaker_timeout_s = 60
circuit_breaker_success_rate = 0.8
typo_field = "oops"
"#;
        let result = toml::from_str::<ResilienceConfig>(bad_toml);
        assert!(result.is_err(), "unknown fields must be rejected");
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("typo_field") || msg.contains("unknown"),
            "error should mention the unknown field: {msg}"
        );
    }

    #[test]
    fn test_unknown_field_in_deduplication_is_rejected() {
        let bad_toml = r#"
enabled = true
unknwon_key = 42
"#;
        let result = toml::from_str::<DeduplicationConfig>(bad_toml);
        assert!(result.is_err(), "unknown fields must be rejected");
    }
}