xybrid-sdk 0.1.0

//! Telemetry event bridge - Converts OrchestratorEvent to TelemetryEvent
//!
//! This module bridges events from the orchestrator's event bus to the
//! telemetry stream used by Flutter and other consumers. It also supports
//! exporting telemetry to the Xybrid Platform for analytics and monitoring.
//!
//! # Span Collection
//!
//! This module integrates with `xybrid_core::tracing` to capture execution spans.
//! When a pipeline completes, the span tree is automatically included in the
//! `PipelineComplete` telemetry event and sent to the Platform for visualization.
//!
//! # Resilience Features
//!
//! The HTTP exporter includes production-hardening features:
//! - **Circuit breaker**: Prevents hammering failing endpoints
//! - **Automatic retry**: Exponential backoff with jitter for transient failures
//! - **Failed event queue**: Retries failed events in the background
//!
//! # Orchestrator Bridge Context
//!
//! Producers attach context at orchestrator event publish time; bridge threads
//! must not assume telemetry task-local or thread-local state is available when
//! they drain events later.

use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex, RwLock};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use uuid::Uuid;
pub use xybrid_core::device::DeviceProfile;
use xybrid_core::device::{ResourceMonitor, ResourceTelemetryMode, ResourceUsageSummary};
use xybrid_core::event_bus::{EventContext, OrchestratorEvent};
use xybrid_core::execution::listener::{self as execution_listener, ExecutionEvent};
use xybrid_core::http::{CircuitBreaker, CircuitConfig, RetryPolicy};
use xybrid_core::orchestrator::routing_engine::LocalReliabilityHint;
use xybrid_core::tracing as core_tracing;

/// Telemetry event type (simplified for FFI)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TelemetryEvent {
    /// Event type name
    pub event_type: String,
    /// Stage name (if applicable)
    pub stage_name: Option<String>,
    /// Target (local/cloud/fallback)
    pub target: Option<String>,
    /// Latency in milliseconds (if applicable)
    pub latency_ms: Option<u32>,
    /// Error message (if applicable)
    pub error: Option<String>,
    /// Additional event data as JSON string
    pub data: Option<String>,
    /// Timestamp in milliseconds since epoch
    pub timestamp_ms: u64,
}

/// Global telemetry event channel for streaming
pub type TelemetrySender = mpsc::Sender<TelemetryEvent>;

static TELEMETRY_SENDERS: Mutex<Vec<TelemetrySender>> = Mutex::new(Vec::new());

// ============================================================================
// HTTP Platform Exporter
// ============================================================================

/// Default telemetry ingest URL
pub const DEFAULT_INGEST_URL: &str = "https://ingest.xybrid.dev";

/// Maximum number of events to keep in the failed queue
const MAX_FAILED_QUEUE_SIZE: usize = 1000;

/// Connection timeout for telemetry requests (5 seconds)
const CONNECT_TIMEOUT_MS: u64 = 5000;

/// Request timeout for telemetry requests (10 seconds)
const REQUEST_TIMEOUT_MS: u64 = 10000;

/// Configuration for the HTTP telemetry exporter
#[derive(Debug, Clone)]
pub struct TelemetryConfig {
    /// Platform API endpoint URL (e.g., "https://api.xybrid.dev")
    pub endpoint: String,
    /// API key for authentication
    pub api_key: String,
    /// Session ID for grouping events (generated if not provided)
    pub session_id: Uuid,
    /// Device identifier
    pub device_id: Option<String>,
    /// Platform name (e.g., "ios", "android", "macos")
    pub platform: Option<String>,
    /// App version string
    pub app_version: Option<String>,
    /// Batch size before flushing (default: 10)
    pub batch_size: usize,
    /// Flush interval in seconds (default: 5)
    pub flush_interval_secs: u64,
    /// Maximum retry attempts for failed batches (default: 3)
    pub max_retries: u32,
    /// Enable retry queue for failed events (default: true)
    pub enable_retry_queue: bool,

    /// Human-friendly device label (e.g. "Sami's MacBook Pro"). Shown in the
    /// dashboard alongside the stable `device_id`.
    pub device_label: Option<String>,
    /// Full `DeviceProfile` override. When `Some`, fields win over both the
    /// auto-detected profile and any partial patch.
    pub device_profile_override: Option<DeviceProfile>,
    /// Partial hardware overrides. Merged onto the auto-detected profile.
    pub device_profile_patch: DeviceProfile,
    /// When `true` (default), the exporter probes local hardware at init and
    /// populates the `device` substructure on every event.
    pub auto_hardware_detection: bool,
    /// When `true`, the exporter includes the machine's hostname in the
    /// `device` substructure. Off by default because hostnames are PII.
    pub capture_hostname: bool,
    /// Internal: set to `true` only when the caller supplied `device_id`
    /// explicitly via `with_device(...)`. Distinguishes caller-supplied
    /// identifiers from the auto-wired default so the opt-out path can
    /// suppress the latter without dropping the former.
    #[doc(hidden)]
    pub device_id_explicit: bool,

    /// Resource telemetry mode. Defaults to `Off` so existing callers see no
    /// behavior change. Enable via [`TelemetryConfig::with_resource_telemetry`]
    /// — authenticated deployments typically use
    /// `ResourceTelemetryMode::summary()`. Respected by `init_platform_telemetry`
    /// which pre-warms the process-wide [`xybrid_core::device::ResourceMonitor`]
    /// so the first inference does not pay a sysinfo cold-read cost.
    pub resource_telemetry: ResourceTelemetryMode,
}

impl Default for TelemetryConfig {
    fn default() -> Self {
        let device = crate::device::Device::current();
        Self {
            endpoint: String::new(),
            api_key: String::new(),
            session_id: Uuid::new_v4(),
            device_id: Some(device.id.clone()),
            platform: Some(device.platform.clone()),
            app_version: None,
            batch_size: 10,
            flush_interval_secs: 5,
            max_retries: 3,
            enable_retry_queue: true,
            device_label: None,
            device_profile_override: None,
            device_profile_patch: DeviceProfile::default(),
            auto_hardware_detection: true,
            capture_hostname: false,
            device_id_explicit: false,
            resource_telemetry: ResourceTelemetryMode::Off,
        }
    }
}

impl TelemetryConfig {
    /// Create a new config with endpoint and API key
    pub fn new(endpoint: impl Into<String>, api_key: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            api_key: api_key.into(),
            ..Default::default()
        }
    }

    /// Set the session ID
    pub fn with_session_id(mut self, session_id: Uuid) -> Self {
        self.session_id = session_id;
        self
    }

    /// Set device metadata.
    ///
    /// Sets both the `device_id` (stable identifier) and `platform` (OS family
    /// string). Independent of the hardware profile; auto-detection keeps
    /// running unless you opt out via `with_auto_hardware_detection(false)`.
    pub fn with_device(
        mut self,
        device_id: impl Into<String>,
        platform: impl Into<String>,
    ) -> Self {
        self.device_id = Some(device_id.into());
        self.platform = Some(platform.into());
        self.device_id_explicit = true;
        self
    }

    /// Override only the platform string (device ID remains auto-detected).
    pub fn with_platform(mut self, platform: impl Into<String>) -> Self {
        self.platform = Some(platform.into());
        self
    }

    /// Set app version
    pub fn with_app_version(mut self, version: impl Into<String>) -> Self {
        self.app_version = Some(version.into());
        self
    }

    /// Set batch size
    pub fn with_batch_size(mut self, size: usize) -> Self {
        self.batch_size = size;
        self
    }

    /// Set flush interval
    pub fn with_flush_interval(mut self, secs: u64) -> Self {
        self.flush_interval_secs = secs;
        self
    }

    /// Set a human-friendly label for this device. Example: `"Sami's MacBook Pro"`.
    ///
    /// The label is shown in the console alongside the stable `device_id`.
    pub fn with_device_label(mut self, label: impl Into<String>) -> Self {
        self.device_label = Some(label.into());
        self
    }

    /// Supply the complete `DeviceProfile` emitted on the wire. Disables
    /// automatic hardware detection so any field left as `None` stays `None`
    /// — callers wanting to opt out of leaking OS / chip / RAM just omit
    /// those fields. For partial overlays onto auto-detected values, use the
    /// `with_hardware_*` field-specific builders instead.
    pub fn with_hardware(mut self, profile: DeviceProfile) -> Self {
        self.device_profile_override = Some(profile);
        self.auto_hardware_detection = false;
        self
    }

    /// Override the detected chip family / CPU brand.
    pub fn with_hardware_chip(mut self, chip: impl Into<String>) -> Self {
        self.device_profile_patch.chip_family = Some(chip.into());
        self
    }

    /// Override the detected RAM (gigabytes).
    pub fn with_hardware_ram_gb(mut self, gb: u32) -> Self {
        self.device_profile_patch.ram_gb = Some(gb);
        self
    }

    /// Override the detected OS name and version.
    pub fn with_hardware_os(mut self, os: impl Into<String>, version: impl Into<String>) -> Self {
        self.device_profile_patch.os = Some(os.into());
        self.device_profile_patch.os_version = Some(version.into());
        self
    }

    /// Override the detected CPU architecture (e.g. `"arm64"`, `"x86_64"`).
    pub fn with_hardware_arch(mut self, arch: impl Into<String>) -> Self {
        self.device_profile_patch.arch = Some(arch.into());
        self
    }

    /// Add an arbitrary app-provided attribute, stored under `device.custom`
    /// on the wire event.
    pub fn with_device_attribute(
        mut self,
        key: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.device_profile_patch
            .custom
            .insert(key.into(), value.into());
        self
    }

    /// Opt out of all hardware auto-detection. When `false`, the `device`
    /// substructure only contains fields the app supplies explicitly.
    pub fn with_auto_hardware_detection(mut self, enabled: bool) -> Self {
        self.auto_hardware_detection = enabled;
        self
    }

    /// Opt into hostname capture. Off by default because hostnames like
    /// `Samis-MacBook` are effectively PII and make the payload identify
    /// a person rather than a piece of hardware.
    pub fn with_hostname_capture(mut self, enabled: bool) -> Self {
        self.capture_hostname = enabled;
        self
    }

    /// Enable per-inference resource telemetry (CPU, memory, process RSS,
    /// pressure, thermal, battery). Sibling to the existing cache-token and
    /// LLM-metric fields on the wire; fully documented at
    /// `docs/sdk/resource-telemetry.md`.
    ///
    /// Authenticated deployments typically want
    /// `ResourceTelemetryMode::summary()`. Omitting the call keeps the default
    /// `Off` so existing callers see no behavior change.
    pub fn with_resource_telemetry(mut self, mode: ResourceTelemetryMode) -> Self {
        self.resource_telemetry = mode.normalized();
        self
    }
}

/// Read the `XYBRID_RESOURCE_TELEMETRY` env var, if set. Recognized values:
/// `off`, `boundary`, `summary`, `summary:<ms>`, `debug_local`,
/// `debug_local:<ms>`. Unknown values fall back to `None` so the caller can
/// use their configured default.
fn resource_mode_from_env() -> Option<ResourceTelemetryMode> {
    let raw = std::env::var("XYBRID_RESOURCE_TELEMETRY").ok()?;
    let lower = raw.trim().to_ascii_lowercase();
    let (head, interval) = match lower.split_once(':') {
        Some((h, t)) => (h, t.parse::<u32>().ok()),
        None => (lower.as_str(), None),
    };
    let default_interval = ResourceTelemetryMode::DEFAULT_SUMMARY_INTERVAL_MS;
    let mode = match head {
        "off" => ResourceTelemetryMode::Off,
        "boundary" => ResourceTelemetryMode::Boundary,
        "summary" => ResourceTelemetryMode::Summary {
            interval_ms: interval.unwrap_or(default_interval),
        },
        "debug_local" | "debuglocal" | "debug-local" => ResourceTelemetryMode::DebugLocal {
            interval_ms: interval.unwrap_or(default_interval),
        },
        _ => return None,
    };
    Some(mode.normalized())
}

/// Event payload for platform API (matches IngestTelemetryEvent)
#[derive(Debug, Clone, Serialize)]
struct PlatformEvent {
    session_id: Uuid,
    event_type: String,
    payload: serde_json::Value,
    // `device_id` honors the opt-out contract: when the SDK clears it
    // because the caller opted out of hardware detection without supplying
    // an explicit id, the wire event omits the field entirely rather than
    // emitting `"device_id": null`. Some ingest schemas treat absent vs
    // null-but-present differently, so this matters.
    #[serde(skip_serializing_if = "Option::is_none")]
    device_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    device_label: Option<String>,
    platform: Option<String>,
    app_version: Option<String>,
    /// Hardware + OS snapshot. `None` when the app has opted out of
    /// auto-detection and supplied no explicit overrides.
    #[serde(skip_serializing_if = "Option::is_none")]
    device: Option<DeviceProfile>,
    timestamp: Option<String>,
    pipeline_id: Option<Uuid>,
    trace_id: Option<Uuid>,
    #[serde(skip_serializing_if = "Option::is_none")]
    correlation_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    outcome_category: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    abort_reason: Option<String>,
    stages: Option<serde_json::Value>,
}

/// Batch payload for platform API
#[derive(Debug, Serialize)]
struct PlatformEventBatch {
    events: Vec<PlatformEvent>,
}

const CONTEXT_PIPELINE_ID_KEY: &str = "__xybrid_pipeline_id";
const CONTEXT_TRACE_ID_KEY: &str = "__xybrid_trace_id";

/// HTTP telemetry exporter that sends events to the Xybrid Platform
///
/// # Resilience Features
///
/// - **Circuit breaker**: Opens after 3 consecutive failures, stays open for 30s
/// - **Automatic retry**: Up to 3 attempts with exponential backoff
/// - **Failed event queue**: Stores up to 1000 failed events for later retry
pub struct HttpTelemetryExporter {
    config: TelemetryConfig,
    device_profile: Option<DeviceProfile>,
    buffer: Arc<Mutex<Vec<TelemetryEvent>>>,
    running: Arc<AtomicBool>,
    /// Current pipeline context for enriching events
    pipeline_id: Arc<RwLock<Option<Uuid>>>,
    trace_id: Arc<RwLock<Option<Uuid>>>,
    /// HTTP agent with timeouts configured
    agent: ureq::Agent,
    /// Circuit breaker for the telemetry endpoint
    circuit: Arc<CircuitBreaker>,
    /// Retry policy for batch submissions
    retry_policy: RetryPolicy,
    /// Queue for failed events that will be retried
    failed_queue: Arc<Mutex<VecDeque<PlatformEvent>>>,
    /// Counter for dropped events (when queue is full)
    dropped_count: Arc<AtomicU32>,
}

impl HttpTelemetryExporter {
    /// Create a new HTTP exporter with the given configuration.
    pub fn new(mut config: TelemetryConfig) -> Self {
        let device_profile = resolve_device_profile(&config);
        // Privacy opt-out contract: when the caller disabled hardware
        // auto-detection and did not explicitly supply an identifier via
        // `with_device(...)`, suppress the `device_id` that
        // `TelemetryConfig::default()` auto-wired from `Device::current()`.
        // Explicit non-hardware context (labels, attributes, hostname capture)
        // no longer re-enables the default identifier — the caller must opt
        // back in via `with_device(...)`. `platform` is kept because it's an
        // OS family string, not PII.
        if !config.auto_hardware_detection && !config.device_id_explicit {
            config.device_id = None;
        }

        // Create HTTP agent with timeouts
        let agent = ureq::AgentBuilder::new()
            .timeout_connect(Duration::from_millis(CONNECT_TIMEOUT_MS))
            .timeout(Duration::from_millis(REQUEST_TIMEOUT_MS))
            .build();

        // Circuit breaker: open after 3 failures, stay open for 30s
        let circuit = Arc::new(CircuitBreaker::new(CircuitConfig::default()));

        // Retry policy with configurable max attempts
        let retry_policy = RetryPolicy {
            max_attempts: config.max_retries,
            initial_delay_ms: 500,
            max_delay_ms: 5000,
            jitter_factor: 0.3,
        };

        Self {
            config,
            device_profile,
            buffer: Arc::new(Mutex::new(Vec::new())),
            running: Arc::new(AtomicBool::new(false)),
            pipeline_id: Arc::new(RwLock::new(None)),
            trace_id: Arc::new(RwLock::new(None)),
            agent,
            circuit,
            retry_policy,
            failed_queue: Arc::new(Mutex::new(VecDeque::new())),
            dropped_count: Arc::new(AtomicU32::new(0)),
        }
    }

    /// Create from environment variables
    ///
    /// Reads:
    /// - `XYBRID_API_KEY` - Required API key
    /// - `XYBRID_INGEST_URL` - Ingest endpoint (default: https://ingest.xybrid.dev)
    /// - `XYBRID_PLATFORM_URL` - Legacy fallback (deprecated)
    pub fn from_env() -> Option<Self> {
        let api_key = std::env::var("XYBRID_API_KEY").ok()?;
        // Try new env var first, then legacy, then default
        let endpoint = std::env::var("XYBRID_INGEST_URL")
            .or_else(|_| std::env::var("XYBRID_PLATFORM_URL"))
            .unwrap_or_else(|_| DEFAULT_INGEST_URL.to_string());

        let config = TelemetryConfig::new(endpoint, api_key);
        Some(Self::new(config))
    }

    /// Set the current pipeline context for event enrichment
    pub fn set_pipeline_context(&self, pipeline_id: Option<Uuid>, trace_id: Option<Uuid>) {
        if let Ok(mut pid) = self.pipeline_id.write() {
            *pid = pipeline_id;
        }
        if let Ok(mut tid) = self.trace_id.write() {
            *tid = trace_id;
        }
    }

    /// Check if the circuit breaker is open (blocking requests).
    pub fn is_circuit_open(&self) -> bool {
        self.circuit.is_open()
    }

    /// Reset the circuit breaker to closed state.
    pub fn reset_circuit(&self) {
        self.circuit.reset();
    }

    /// Get the number of events waiting in the failed queue.
    pub fn failed_queue_size(&self) -> usize {
        self.failed_queue.lock().map(|q| q.len()).unwrap_or(0)
    }

    /// Get the number of events that were dropped due to queue overflow.
    pub fn dropped_count(&self) -> u32 {
        self.dropped_count.load(Ordering::Relaxed)
    }

    /// Start the background flush thread
    pub fn start(&self) {
        if self.running.swap(true, Ordering::SeqCst) {
            return; // Already running
        }

        let buffer = Arc::clone(&self.buffer);
        let running = Arc::clone(&self.running);
        let config = self.config.clone();
        let device_profile = self.device_profile.clone();
        let flush_interval = Duration::from_secs(config.flush_interval_secs);
        let pipeline_id = Arc::clone(&self.pipeline_id);
        let trace_id = Arc::clone(&self.trace_id);
        let agent = self.agent.clone();
        let circuit = Arc::clone(&self.circuit);
        let retry_policy = self.retry_policy.clone();
        let failed_queue = Arc::clone(&self.failed_queue);
        let dropped_count = Arc::clone(&self.dropped_count);

        thread::spawn(move || {
            while running.load(Ordering::SeqCst) {
                thread::sleep(flush_interval);

                // First, try to send any failed events from the queue
                if config.enable_retry_queue {
                    retry_failed_events(&failed_queue, &config, &agent, &circuit, &retry_policy);
                }

                // Then flush the current buffer
                flush_buffer_with_retry(
                    &buffer,
                    &config,
                    device_profile.as_ref(),
                    &pipeline_id,
                    &trace_id,
                    &agent,
                    &circuit,
                    &retry_policy,
                    &failed_queue,
                    &dropped_count,
                );
            }
        });
    }

    /// Stop the background flush thread
    pub fn stop(&self) {
        self.running.store(false, Ordering::SeqCst);
        // Final flush with retry
        flush_buffer_with_retry(
            &self.buffer,
            &self.config,
            self.device_profile.as_ref(),
            &self.pipeline_id,
            &self.trace_id,
            &self.agent,
            &self.circuit,
            &self.retry_policy,
            &self.failed_queue,
            &self.dropped_count,
        );
    }

    /// Add an event to the buffer
    pub fn push(&self, event: TelemetryEvent) {
        let mut buffer = self.buffer.lock().unwrap();
        buffer.push(event);

        // Flush if buffer is full
        if buffer.len() >= self.config.batch_size {
            let events: Vec<TelemetryEvent> = buffer.drain(..).collect();
            drop(buffer); // Release lock before HTTP call
            send_batch_with_retry(
                &events,
                &self.config,
                self.device_profile.as_ref(),
                &self.pipeline_id,
                &self.trace_id,
                &self.agent,
                &self.circuit,
                &self.retry_policy,
                &self.failed_queue,
                &self.dropped_count,
            );
        }
    }

    /// Force flush all buffered events
    pub fn flush(&self) {
        flush_buffer_with_retry(
            &self.buffer,
            &self.config,
            self.device_profile.as_ref(),
            &self.pipeline_id,
            &self.trace_id,
            &self.agent,
            &self.circuit,
            &self.retry_policy,
            &self.failed_queue,
            &self.dropped_count,
        );
    }

    /// Create a telemetry sender that feeds into this exporter
    pub fn create_sender(&self) -> TelemetrySender {
        let (tx, rx) = mpsc::channel::<TelemetryEvent>();
        let buffer = Arc::clone(&self.buffer);
        let batch_size = self.config.batch_size;
        let config = self.config.clone();
        let device_profile = self.device_profile.clone();
        let pipeline_id = Arc::clone(&self.pipeline_id);
        let trace_id = Arc::clone(&self.trace_id);
        let agent = self.agent.clone();
        let circuit = Arc::clone(&self.circuit);
        let retry_policy = self.retry_policy.clone();
        let failed_queue = Arc::clone(&self.failed_queue);
        let dropped_count = Arc::clone(&self.dropped_count);

        thread::spawn(move || {
            for event in rx {
                let mut buf = buffer.lock().unwrap();
                buf.push(event);

                if buf.len() >= batch_size {
                    let events: Vec<TelemetryEvent> = buf.drain(..).collect();
                    drop(buf);
                    send_batch_with_retry(
                        &events,
                        &config,
                        device_profile.as_ref(),
                        &pipeline_id,
                        &trace_id,
                        &agent,
                        &circuit,
                        &retry_policy,
                        &failed_queue,
                        &dropped_count,
                    );
                }
            }
        });

        tx
    }
}

impl Drop for HttpTelemetryExporter {
    fn drop(&mut self) {
        self.stop();
    }
}

/// Flush all buffered events to the platform with retry logic.
fn flush_buffer_with_retry(
    buffer: &Arc<Mutex<Vec<TelemetryEvent>>>,
    config: &TelemetryConfig,
    device_profile: Option<&DeviceProfile>,
    pipeline_id: &Arc<RwLock<Option<Uuid>>>,
    trace_id: &Arc<RwLock<Option<Uuid>>>,
    agent: &ureq::Agent,
    circuit: &Arc<CircuitBreaker>,
    retry_policy: &RetryPolicy,
    failed_queue: &Arc<Mutex<VecDeque<PlatformEvent>>>,
    dropped_count: &Arc<AtomicU32>,
) {
    let events: Vec<TelemetryEvent> = {
        let mut buf = buffer.lock().unwrap();
        buf.drain(..).collect()
    };

    if !events.is_empty() {
        send_batch_with_retry(
            &events,
            config,
            device_profile,
            pipeline_id,
            trace_id,
            agent,
            circuit,
            retry_policy,
            failed_queue,
            dropped_count,
        );
    }
}

/// Send a batch of events to the platform API with retry and circuit breaker.
fn send_batch_with_retry(
    events: &[TelemetryEvent],
    config: &TelemetryConfig,
    device_profile: Option<&DeviceProfile>,
    pipeline_id: &Arc<RwLock<Option<Uuid>>>,
    trace_id: &Arc<RwLock<Option<Uuid>>>,
    agent: &ureq::Agent,
    circuit: &Arc<CircuitBreaker>,
    retry_policy: &RetryPolicy,
    failed_queue: &Arc<Mutex<VecDeque<PlatformEvent>>>,
    dropped_count: &Arc<AtomicU32>,
) {
    if events.is_empty() || config.endpoint.is_empty() || config.api_key.is_empty() {
        return;
    }

    // Check circuit breaker
    if !circuit.can_execute() {
        // Circuit is open, queue events for later
        if config.enable_retry_queue {
            let pid = pipeline_id.read().ok().and_then(|g| *g);
            let tid = trace_id.read().ok().and_then(|g| *g);
            let platform_events: Vec<PlatformEvent> = events
                .iter()
                .map(|e| convert_to_platform_event(e, config, device_profile, pid, tid))
                .collect();
            queue_failed_events(platform_events, failed_queue, dropped_count);
        }
        return;
    }

    let pid = pipeline_id.read().ok().and_then(|g| *g);
    let tid = trace_id.read().ok().and_then(|g| *g);

    let platform_events: Vec<PlatformEvent> = events
        .iter()
        .map(|e| convert_to_platform_event(e, config, device_profile, pid, tid))
        .collect();

    // Try to send with retry
    let result = send_batch_inner(&platform_events, config, agent, circuit, retry_policy);

    if let Err(failed_events) = result {
        // Queue failed events for later retry
        if config.enable_retry_queue {
            queue_failed_events(failed_events, failed_queue, dropped_count);
        }
    }
}

/// Inner send function that returns the events on failure for queueing.
fn send_batch_inner(
    events: &[PlatformEvent],
    config: &TelemetryConfig,
    agent: &ureq::Agent,
    circuit: &Arc<CircuitBreaker>,
    retry_policy: &RetryPolicy,
) -> Result<(), Vec<PlatformEvent>> {
    let batch = PlatformEventBatch {
        events: events.to_vec(),
    };

    let url = format!("{}/v1/events/batch", config.endpoint.trim_end_matches('/'));

    for attempt in 0..retry_policy.max_attempts {
        // Calculate delay for this attempt
        let delay = retry_policy.delay_for_attempt(attempt);
        if !delay.is_zero() {
            std::thread::sleep(delay);
        }

        // Check circuit breaker again
        if !circuit.can_execute() {
            return Err(events.to_vec());
        }

        // Send HTTP request
        let result = agent
            .post(&url)
            .set("Authorization", &format!("Bearer {}", config.api_key))
            .set("Content-Type", "application/json")
            .send_json(&batch);

        match result {
            Ok(response) => {
                let status = response.status();
                if status == 200 || status == 201 {
                    circuit.record_success();
                    return Ok(());
                } else if is_retryable_status(status) {
                    circuit.record_failure();
                    // Continue to retry
                } else {
                    // Non-retryable error (4xx client errors)
                    circuit.record_success(); // Don't trip circuit for client errors
                    log::warn!(
                        target: "xybrid_telemetry",
                        "Platform returned status {}",
                        status
                    );
                    return Ok(()); // Don't retry or queue client errors
                }
            }
            Err(ureq::Error::Status(status, _)) => {
                if status == 429 {
                    circuit.record_rate_limited();
                } else if is_retryable_status(status) {
                    circuit.record_failure();
                } else {
                    // Non-retryable status
                    circuit.record_success();
                    log::warn!(
                        target: "xybrid_telemetry",
                        "Platform returned status {}",
                        status
                    );
                    return Ok(());
                }
            }
            Err(ureq::Error::Transport(_)) => {
                circuit.record_failure();
                // Continue to retry
            }
        }
    }

    // All retries exhausted
    Err(events.to_vec())
}

/// Check if an HTTP status code is retryable.
fn is_retryable_status(status: u16) -> bool {
    matches!(status, 429 | 502 | 503 | 504)
}

/// Queue failed events for later retry.
fn queue_failed_events(
    events: Vec<PlatformEvent>,
    failed_queue: &Arc<Mutex<VecDeque<PlatformEvent>>>,
    dropped_count: &Arc<AtomicU32>,
) {
    let mut queue = failed_queue.lock().unwrap();

    for event in events {
        if queue.len() >= MAX_FAILED_QUEUE_SIZE {
            // Queue is full, drop oldest event
            queue.pop_front();
            dropped_count.fetch_add(1, Ordering::Relaxed);
        }
        queue.push_back(event);
    }
}

/// Retry sending failed events from the queue.
fn retry_failed_events(
    failed_queue: &Arc<Mutex<VecDeque<PlatformEvent>>>,
    config: &TelemetryConfig,
    agent: &ureq::Agent,
    circuit: &Arc<CircuitBreaker>,
    retry_policy: &RetryPolicy,
) {
    // Don't retry if circuit is open
    if !circuit.can_execute() {
        return;
    }

    // Take a batch of events from the queue
    let events: Vec<PlatformEvent> = {
        let mut queue = failed_queue.lock().unwrap();
        let batch_size = config.batch_size.min(queue.len());
        queue.drain(..batch_size).collect()
    };

    if events.is_empty() {
        return;
    }

    // Try to send the batch
    if let Err(failed_events) = send_batch_inner(&events, config, agent, circuit, retry_policy) {
        // Put them back at the front of the queue
        let mut queue = failed_queue.lock().unwrap();
        for event in failed_events.into_iter().rev() {
            queue.push_front(event);
        }
    }
}

/// Convert SDK TelemetryEvent to Platform format
fn convert_to_platform_event(
    event: &TelemetryEvent,
    config: &TelemetryConfig,
    device_profile: Option<&DeviceProfile>,
    pipeline_id: Option<Uuid>,
    trace_id: Option<Uuid>,
) -> PlatformEvent {
    // Build payload from event fields
    let mut payload = serde_json::json!({});
    let mut event_pipeline_id = pipeline_id;
    let mut event_trace_id = trace_id;
    let mut correlation_id = None;
    let mut outcome_category = None;
    let mut abort_reason = None;

    if let Some(stage) = &event.stage_name {
        payload["stage_name"] = serde_json::json!(stage);
    }
    if let Some(target) = &event.target {
        payload["target"] = serde_json::json!(target);
    }
    if let Some(latency) = event.latency_ms {
        payload["latency_ms"] = serde_json::json!(latency);
    }
    if let Some(error) = &event.error {
        payload["error"] = serde_json::json!(error);
        payload["status"] = serde_json::json!("error");
    } else {
        payload["status"] = serde_json::json!("success");
    }
    if let Some(data) = &event.data {
        // Try to parse as JSON, otherwise store as string
        if let Ok(mut parsed) = serde_json::from_str::<serde_json::Value>(data) {
            if let Some(obj) = parsed.as_object_mut() {
                if let Some(v) = obj.remove(CONTEXT_PIPELINE_ID_KEY) {
                    event_pipeline_id = v.as_str().and_then(|s| Uuid::parse_str(s).ok());
                }
                if let Some(v) = obj.remove(CONTEXT_TRACE_ID_KEY) {
                    event_trace_id = v.as_str().and_then(|s| Uuid::parse_str(s).ok());
                }
            }
            // Hoist select top-level fields from `data` into `payload` so
            // downstream analytics (the platform's /traces list reads
            // these at the top level of the JSONB payload) can find them
            // without diving into the nested `data` object. Spans further
            // down already populate `tokens_in` / `tokens_out` for
            // local-inference events; this covers the non-span path where
            // a caller publishes usage info manually — e.g. a direct
            // cloud-provider response whose `usage` block wouldn't
            // otherwise reach the dashboard's cost column.
            for key in [
                "tokens_in",
                "tokens_out",
                "model_id",
                // Cache-tier tokens for providers that report them.
                // Canonical names (Anthropic-flavored) so the list page
                // and detail-page receipt work uniformly across
                // DeepSeek / Anthropic / OpenAI / Gemini.
                "cache_read_input_tokens",
                "cache_creation_input_tokens",
                // Per-inference resource summary. Nested JSON stays under
                // `data.resource_summary`; hoisting the object to the payload
                // top level lets the analytics backend column-extract via
                // flat JSON-path selectors without teaching each consumer
                // the nested shape.
                "resource_summary",
                "correlation_id",
                "outcome_category",
                "abort_reason",
                // Per-routing-decision reliability hint (object with
                // `recent_abort_rate` + `sample_size`). Lives in the SDK
                // hoist list so the analytics backend can extract it via
                // `json:$.local_reliability_hint.*` without descending
                // into the nested data object.
                "local_reliability_hint",
                // Streaming flag — `XybridModel::run_streaming` and the
                // streaming-fast-path `ModelComplete` (Pipeline) both
                // stamp `data.streaming = true`. Hoisting to the top
                // level lets the Tinybird datasource pick it up as a
                // typed column for a `streaming` badge / filter on the
                // Traces dashboard, distinguishing chat-flow / REPL
                // turns from batch-style inferences at a glance.
                "streaming",
            ]
            .iter()
            {
                if payload.get(*key).is_none() {
                    if let Some(v) = parsed.get(*key) {
                        payload[*key] = v.clone();
                    }
                }
            }
            correlation_id = parsed
                .get("correlation_id")
                .and_then(serde_json::Value::as_str)
                .map(str::to_string);
            outcome_category = parsed.get("outcome_category").cloned();
            abort_reason = parsed
                .get("abort_reason")
                .and_then(serde_json::Value::as_str)
                .map(str::to_string);
            payload["data"] = parsed;
        } else {
            payload["data"] = serde_json::json!(data);
        }
    }

    // Convert timestamp
    let timestamp = chrono::DateTime::from_timestamp_millis(event.timestamp_ms as i64)
        .map(|dt| dt.to_rfc3339());

    // Capture spans for PipelineComplete and ModelComplete events.
    //
    // Two source paths, in priority order:
    //
    //   1. **Embedded spans in `event.data["spans"]`** — used when the
    //      caller pre-captured the span tree synchronously before
    //      publishing. This sidesteps a race in the default (global
    //      collector) path: `convert_to_platform_event` runs on the
    //      exporter's background thread, but the global tracing
    //      collector is mutated live by whatever the caller's next
    //      unit of work is opening. For bursty emitters (e.g. chunked
    //      document summarization firing many ModelComplete events
    //      back-to-back), this produced empty/misattributed stages.
    //      Callers that need deterministic per-event capture write
    //      `get_stages_json()["spans"]` into `event.data["spans"]`
    //      and reset tracing themselves before publishing.
    //
    //   2. **Global `TemplateExecutor`-populated collector** — the
    //      historical path, preserved for in-process local inference
    //      where only one model call is active at a time and the race
    //      doesn't manifest.
    // Span-bearing event types: completion-family events publish a final
    // ModelComplete/PipelineComplete that drains the collector, AND the
    // cloud-fallback flow publishes LocalAborted/CloudRetry as terminal
    // markers for each leg without ever firing a *Complete, AND
    // ModelWarmup is its own completion category (XybridModel::warmup
    // opens executor spans then publishes a single ModelWarmup —
    // distinct from ModelComplete so the dashboard can filter warmups
    // out of cost-attribution rollups). All four kinds need their
    // flamegraph attached at the wire layer or the dashboard sees an
    // empty trace detail. This list MUST match `snapshot_spans_into_event`
    // above; otherwise the SDK attaches spans to event.data["spans"]
    // but `convert_to_platform_event` strips them again before the wire.
    let stages = if matches!(
        event.event_type.as_str(),
        "PipelineComplete" | "ModelComplete" | "ModelWarmup" | "LocalAborted" | "CloudRetry"
    ) && core_tracing::is_tracing_enabled()
    {
        let embedded_spans: Option<serde_json::Value> = payload
            .get("data")
            .and_then(|d| d.get("spans"))
            .filter(|v| !v.is_null())
            .cloned();

        let spans = if let Some(inner) = embedded_spans {
            // Do NOT reset the global collector here. The caller already
            // did its own synchronous capture+reset on the publishing
            // thread (see `publish_telemetry_event::snapshot_spans_into_event`
            // and Mirage's `summarize.rs::synthesize_chunk_cloud`). Any
            // state currently in the collector at this moment belongs to
            // OTHER in-flight work on other threads — resetting it here
            // would steal their open spans and drop them on the floor,
            // producing empty flamegraphs for those subsequent events.
            // That's the race the earlier reset caused: a classifier's
            // async-processed embedded-spans event clobbered a concurrent
            // cloud synthesize's still-open span stack.
            serde_json::json!({ "spans": inner })
        } else {
            let s = core_tracing::get_stages_json();
            core_tracing::reset_tracing();
            s
        };

        // Hoist LLM token counts from the span metadata into the outer
        // payload so analytics backends that read `tokens_in` /
        // `tokens_out` at the top of the event see them without having
        // to descend into the span tree. No-op when no `llm_inference*`
        // span is present.
        if let Some((tokens_in, tokens_out)) = extract_llm_token_counts(&spans) {
            if let Some(n) = tokens_in {
                if payload.get("tokens_in").is_none() {
                    payload["tokens_in"] = serde_json::json!(n);
                }
            }
            if let Some(n) = tokens_out {
                if payload.get("tokens_out").is_none() {
                    payload["tokens_out"] = serde_json::json!(n);
                }
            }
        }
        // Hoist cost-accounting string attrs onto the payload top level
        // for the billing column.
        //
        // `backend` may live on any span — LLM events set it on the
        // inner `llm_inference` span (via `LlmStrategy` / cloud adapter),
        // non-LLM events (ASR/TTS) set it on the outer `execute:<model>`
        // span via `backend_label_from_template`. Read from any span so
        // both surfaces produce the same wire shape.
        //
        // `provider` is cloud-LLM-only and must stay gated to LLM spans:
        // a non-LLM span carrying a stray `provider` key would emit a
        // phantom value on an ASR/TTS payload.
        if payload.get("backend").is_none() {
            if let Some(v) = extract_string_attr_from_any_span(&spans, "backend") {
                payload["backend"] = serde_json::json!(v);
            }
        }
        if payload.get("provider").is_none() {
            if let Some(v) = extract_llm_inference_string_attr(&spans, "provider") {
                payload["provider"] = serde_json::json!(v);
            }
        }
        // Semantic task label (chat / vlm / asr / tts / embedding / …)
        // sourced from `model_metadata.json`. Lives on the outer
        // `execute:<model>` span so it's always reachable via the
        // any-span hoist regardless of modality. Promotes the field
        // to the wire payload so the console can filter without
        // joining model_id against the registry at render time.
        if payload.get("task").is_none() {
            if let Some(v) = extract_string_attr_from_any_span(&spans, "task") {
                payload["task"] = serde_json::json!(v);
            }
        }
        // Quantization label (q4_0 / q4_k_m / fp16 / …) sourced from
        // `model_metadata.json` first, GGUF filename second. Two runs
        // of "the same model" at different quantizations are now
        // distinguishable in the analytics rollup. Open string —
        // values outside the documented set still flow through.
        if payload.get("quantization").is_none() {
            if let Some(v) = extract_string_attr_from_any_span(&spans, "quantization") {
                payload["quantization"] = serde_json::json!(v);
            }
        }
        // Resolved execution provider — which on-device engine path
        // actually ran (coreml / cpu / metal / cuda / mlx-metal / …).
        // ORT path: harvested from per-session profiling JSON after the
        // first inference (bypasses the API gap where ORT exposes no
        // session-level resolved-EP getter). LLM path: build-flag-derived
        // label keyed on the backend name. Cloud paths omit — `provider`
        // already carries the attribution. Lives on whichever inner span
        // emitted it, hoisted via the any-span lookup so all modalities
        // produce the same wire shape.
        if payload.get("execution_provider").is_none() {
            if let Some(v) = extract_string_attr_from_any_span(&spans, "execution_provider") {
                payload["execution_provider"] = serde_json::json!(v);
            }
        }
        // Local KV cache hits (llama.cpp's multi-turn prefix reuse). The
        // backend only emits this metadata key when the most recent call
        // actually reused a prefix (n > 0), so a present value here is
        // always meaningful — no need to filter out zeros at the SDK
        // layer. Mirror of cloud's `cache_read_input_tokens` so the
        // analytics column can stack local + cloud cache hits.
        if payload.get("prompt_cached_tokens").is_none() {
            if let Some(n) = extract_llm_prompt_cached_tokens(&spans) {
                payload["prompt_cached_tokens"] = serde_json::json!(n);
            }
        }
        Some(spans)
    } else {
        None
    };

    PlatformEvent {
        session_id: config.session_id,
        event_type: event.event_type.clone(),
        payload,
        device_id: config.device_id.clone(),
        device_label: config.device_label.clone(),
        platform: config.platform.clone(),
        app_version: config.app_version.clone(),
        device: device_profile.cloned(),
        timestamp,
        pipeline_id: event_pipeline_id,
        trace_id: event_trace_id,
        correlation_id,
        outcome_category,
        abort_reason,
        stages,
    }
}

/// Resolve the effective device profile for a config: auto-detect (if on),
/// then apply the per-field patch, then apply the full override. Hostname is
/// only filled when `config.capture_hostname` is true.
fn resolve_device_profile(config: &TelemetryConfig) -> Option<DeviceProfile> {
    let mut profile = if config.auto_hardware_detection {
        DeviceProfile::detect()
    } else {
        DeviceProfile::default()
    };
    profile = profile.merged_with(config.device_profile_patch.clone());
    if let Some(override_) = config.device_profile_override.clone() {
        profile = profile.merged_with(override_);
    }
    if config.capture_hostname && profile.hostname.is_none() {
        profile.hostname = detect_hostname();
    }
    if profile.is_empty() {
        None
    } else {
        Some(profile)
    }
}

fn detect_hostname() -> Option<String> {
    std::env::var("HOSTNAME")
        .or_else(|_| std::env::var("COMPUTERNAME"))
        .ok()
        .map(|hostname| hostname.trim().to_string())
        .filter(|hostname| !hostname.is_empty())
}

/// Walk a `stages` JSON (shape: `{"spans":[{"name","metadata":{...}}]}`) and
/// return `(tokens_in, tokens_out)` read from the first span that either
/// carries LLM-style metadata (`ttft_ms`, `tokens_generated`, `tokens_out`,
/// `completion_tokens`) or is named with a known LLM prefix
/// (`llm_inference*` or `inference:*`).
///
/// Looks at both canonical keys (`tokens_in` / `tokens_out`) and OpenAI-ish
/// keys (`prompt_tokens` / `completion_tokens` + `tokens_generated`) so the
/// hoist works regardless of which LLM adapter emitted the span or how the
/// enclosing orchestrator named it.
fn extract_llm_token_counts(stages: &serde_json::Value) -> Option<(Option<u64>, Option<u64>)> {
    let spans = stages.get("spans")?.as_array()?;
    let read = |meta: Option<&serde_json::Value>, keys: &[&str]| -> Option<u64> {
        for k in keys {
            let Some(v) = meta.and_then(|m| m.get(*k)) else {
                continue;
            };
            if let Some(n) = v.as_u64() {
                return Some(n);
            }
            if let Some(s) = v.as_str() {
                if let Ok(n) = s.parse::<u64>() {
                    return Some(n);
                }
            }
        }
        None
    };
    // Token accounting invariant: the authoritative counts live on the LAST
    // LLM span in the trace. A streaming run emits a timing-only span first
    // (ttft_ms, no counts yet) and a final accounting span with the totals;
    // a retried run emits one span per attempt and we want the final attempt.
    // Earlier-span values only win as a fallback when no later span carries
    // the corresponding key at all.
    let mut saw_llm_span = false;
    let mut tokens_in: Option<u64> = None;
    let mut tokens_out: Option<u64> = None;
    for span in spans {
        let name = span.get("name").and_then(|v| v.as_str()).unwrap_or("");
        let meta = span.get("metadata");
        let is_llm_span = name.starts_with("llm_inference")
            || name.starts_with("inference:")
            || meta
                .map(|m| {
                    m.get("ttft_ms").is_some()
                        || m.get("tokens_generated").is_some()
                        || m.get("tokens_out").is_some()
                        || m.get("completion_tokens").is_some()
                })
                .unwrap_or(false);
        if !is_llm_span {
            continue;
        }
        saw_llm_span = true;
        if let Some(v) = read(meta, &["tokens_in", "prompt_tokens"]) {
            tokens_in = Some(v);
        }
        if let Some(v) = read(
            meta,
            &["tokens_out", "completion_tokens", "tokens_generated"],
        ) {
            tokens_out = Some(v);
        }
    }
    if saw_llm_span {
        Some((tokens_in, tokens_out))
    } else {
        None
    }
}

/// Lift the local-LLM `prompt_cached_tokens` value (count of prompt
/// tokens served from the backend's KV cache on the most recent call)
/// off the LAST LLM span. The local mirror of cloud's
/// `cache_read_input_tokens` — emitted only when a backend that tracks
/// prefix reuse (today: llama.cpp) actually reused tokens, so reading
/// `Some(n)` from this helper means the value is meaningful and worth
/// hoisting verbatim. Same last-span-wins / LLM-span detection rule as
/// [`extract_llm_token_counts`] so retried runs and timing-only spans
/// don't fight the count-bearing span.
fn extract_llm_prompt_cached_tokens(stages: &serde_json::Value) -> Option<u64> {
    let spans = stages.get("spans")?.as_array()?;
    let mut latest: Option<u64> = None;
    for span in spans {
        let name = span.get("name").and_then(|v| v.as_str()).unwrap_or("");
        let meta = span.get("metadata");
        let is_llm_span = name.starts_with("llm_inference")
            || name.starts_with("inference:")
            || meta
                .map(|m| {
                    m.get("ttft_ms").is_some()
                        || m.get("tokens_generated").is_some()
                        || m.get("tokens_out").is_some()
                        || m.get("completion_tokens").is_some()
                })
                .unwrap_or(false);
        if !is_llm_span {
            continue;
        }
        let Some(v) = meta.and_then(|m| m.get("prompt_cached_tokens")) else {
            continue;
        };
        if let Some(n) = v.as_u64() {
            latest = Some(n);
        } else if let Some(s) = v.as_str() {
            if let Ok(n) = s.parse::<u64>() {
                latest = Some(n);
            }
        }
    }
    latest
}

/// Walk a `stages` JSON (same shape as [`extract_llm_token_counts`]) and
/// return the value of `key` from the LAST LLM-flavoured span that
/// carries it as a string. Used to lift cost-accounting string attrs
/// (`backend`, `provider`) onto the wire payload's top level alongside
/// the token counts.
///
/// LLM-span detection is identical to [`extract_llm_token_counts`]
/// (name prefix `llm_inference*` / `inference:*` or LLM-style metadata
/// keys) so the two hoists agree on which span is "the" LLM span when
/// a trace contains other inference spans (e.g. ASR/TTS).
fn extract_llm_inference_string_attr(stages: &serde_json::Value, key: &str) -> Option<String> {
    let spans = stages.get("spans")?.as_array()?;
    let mut found: Option<String> = None;
    for span in spans {
        let name = span.get("name").and_then(|v| v.as_str()).unwrap_or("");
        let meta = span.get("metadata");
        let is_llm_span = name.starts_with("llm_inference")
            || name.starts_with("inference:")
            || meta
                .map(|m| {
                    m.get("ttft_ms").is_some()
                        || m.get("tokens_generated").is_some()
                        || m.get("tokens_out").is_some()
                        || m.get("completion_tokens").is_some()
                })
                .unwrap_or(false);
        if !is_llm_span {
            continue;
        }
        if let Some(v) = meta.and_then(|m| m.get(key)).and_then(|v| v.as_str()) {
            // Same last-span-wins rule as the token-count hoist: a retried
            // run emits one span per attempt and we want the final value.
            found = Some(v.to_string());
        }
    }
    found
}

/// Walk a `stages` JSON and return the value of `key` from the LAST span
/// (any flavour) that carries it as a string. Used for cost-accounting
/// scalars that must surface on every inference event regardless of
/// modality — chiefly `backend`, which is set on the outer
/// `execute:<model>` span for non-LLM events (ASR/TTS via
/// `backend_label_from_template`) and on the inner `llm_inference` span
/// for LLM events.
///
/// Intentionally NOT used for `provider`: that field is cloud-LLM-only
/// and a stray match on a non-LLM span would emit a nonsense provider
/// value on an ASR/TTS payload. Use [`extract_llm_inference_string_attr`]
/// for provider.
fn extract_string_attr_from_any_span(stages: &serde_json::Value, key: &str) -> Option<String> {
    let spans = stages.get("spans")?.as_array()?;
    let mut found: Option<String> = None;
    for span in spans {
        let meta = span.get("metadata");
        if let Some(v) = meta.and_then(|m| m.get(key)).and_then(|v| v.as_str()) {
            // Same last-span-wins rule as the LLM-gated hoist.
            found = Some(v.to_string());
        }
    }
    found
}

// ============================================================================
// Global Platform Exporter
// ============================================================================

static PLATFORM_EXPORTER: RwLock<Option<HttpTelemetryExporter>> = RwLock::new(None);

/// Initialize the global platform telemetry exporter
///
/// This also enables span tracing in xybrid-core for detailed execution profiling.
/// Spans are automatically captured and included in `PipelineComplete` events.
///
/// # Example
///
/// ```no_run
/// use xybrid_sdk::telemetry::{init_platform_telemetry, TelemetryConfig};
///
/// let config = TelemetryConfig::new("https://ingest.xybrid.dev", "your-api-key")
///     .with_device("device-123", "ios")
///     .with_app_version("1.0.0");
///
/// init_platform_telemetry(config);
/// ```
pub fn init_platform_telemetry(mut config: TelemetryConfig) {
    // Enable span tracing in xybrid-core for execution profiling
    core_tracing::init_tracing(true);

    // Resource-telemetry mode. Env override wins over the config value
    // so operators can flip `XYBRID_RESOURCE_TELEMETRY=off` during demos
    // without rebuilding; the env-only init path below uses the same
    // helper so behavior is identical.
    config.resource_telemetry = resolve_resource_telemetry_mode(config.resource_telemetry);
    activate_resource_telemetry(config.resource_telemetry);

    // Register automatic execution listener so TemplateExecutor emits
    // ExecutionStarted / ExecutionCompleted / ExecutionFailed events
    register_execution_listener();

    let exporter = HttpTelemetryExporter::new(config);
    exporter.start();

    // Register as a telemetry sender
    let sender = exporter.create_sender();
    register_telemetry_sender(sender);

    if let Ok(mut global) = PLATFORM_EXPORTER.write() {
        *global = Some(exporter);
    }
}

// -- Process-wide resource-telemetry mode ----------------------------------
//
// `XybridModel::run*` / `Pipeline::run*` consult this to decide whether (and
// how) to instrument a run. Stored independently of `TelemetryConfig` so the
// SDK doesn't have to carry a config reference into every run path. Writes
// happen at `init_platform_telemetry`; reads are cheap.
static RESOURCE_TELEMETRY_MODE: RwLock<ResourceTelemetryMode> =
    RwLock::new(ResourceTelemetryMode::Off);

fn set_resource_telemetry_mode(mode: ResourceTelemetryMode) {
    if let Ok(mut guard) = RESOURCE_TELEMETRY_MODE.write() {
        *guard = mode;
    }
}

/// Apply the `XYBRID_RESOURCE_TELEMETRY` env override on top of a configured
/// mode. The env var wins unconditionally when set (and parses to a known
/// variant) so operators can disable resource telemetry at runtime without a
/// rebuild. Both `init_platform_telemetry` and `init_platform_telemetry_from_env`
/// funnel through here so the two init paths can't drift.
fn resolve_resource_telemetry_mode(configured: ResourceTelemetryMode) -> ResourceTelemetryMode {
    match resource_mode_from_env() {
        Some(env_mode) => env_mode,
        None => configured,
    }
}

/// Pre-warm the global `ResourceMonitor` + publish the active mode to the
/// process-wide state read by `begin_resource_run`. `Off` still stores the
/// mode so a later rehydration can read a consistent value; the monitor
/// just sits idle until something flips it on.
fn activate_resource_telemetry(mode: ResourceTelemetryMode) {
    if !mode.is_off() {
        ResourceMonitor::global().prewarm();
    }
    set_resource_telemetry_mode(mode);
}

/// Current resource-telemetry mode for this process. `Off` until
/// `init_platform_telemetry` runs with a mode other than `Off`.
pub fn resource_telemetry_mode() -> ResourceTelemetryMode {
    match RESOURCE_TELEMETRY_MODE.read() {
        Ok(g) => *g,
        Err(poisoned) => *poisoned.into_inner(),
    }
}

/// Begin a resource-telemetry scope for one `Model::run*` / `Pipeline::run*`
/// call. Returns a guard that produces the summary on `finish()`. Callers
/// who want the summary attached to their outgoing `TelemetryEvent` can use
/// [`attach_resource_summary`] at emit time.
pub fn begin_resource_run() -> xybrid_core::device::RunGuard {
    ResourceMonitor::global().begin_run(resource_telemetry_mode())
}

/// Mutate a `TelemetryEvent.data` JSON string to add `resource_summary`.
/// If `data` already parses as an object, the field is inserted alongside
/// existing keys; otherwise the event is left untouched so we never clobber
/// a publisher's bespoke payload. Safe to call with `None` summary (no-op).
pub fn attach_resource_summary(event: &mut TelemetryEvent, summary: Option<ResourceUsageSummary>) {
    let Some(summary) = summary else {
        return;
    };
    let Ok(summary_json) = serde_json::to_value(&summary) else {
        return;
    };
    let mut parsed: serde_json::Value = match event.data.as_deref() {
        Some(s) if !s.is_empty() => match serde_json::from_str(s) {
            Ok(v) => v,
            Err(_) => return,
        },
        _ => serde_json::json!({}),
    };
    if let Some(obj) = parsed.as_object_mut() {
        obj.insert("resource_summary".to_string(), summary_json);
        event.data = serde_json::to_string(&parsed).ok();
    }
}

/// Attach the resource summary from `guard` and publish the event. Run-site
/// shorthand for callers that always want the two steps bundled:
///
/// ```text
/// let guard = begin_resource_run();
/// // ... run inference ...
/// publish_with_resource_summary(event, guard);
/// ```
///
/// Safe to call with a disabled guard; `attach_resource_summary` becomes a
/// no-op and the publish path is unchanged.
pub fn publish_with_resource_summary(
    mut event: TelemetryEvent,
    guard: xybrid_core::device::RunGuard,
) {
    attach_resource_summary(&mut event, guard.finish());
    publish_telemetry_event(event);
}

pub(crate) fn publish_with_resource_summary_in_context(
    mut event: TelemetryEvent,
    guard: xybrid_core::device::RunGuard,
    pipeline_id: Option<Uuid>,
    trace_id: Option<Uuid>,
) {
    attach_resource_summary(&mut event, guard.finish());
    publish_telemetry_event_in_context(event, pipeline_id, trace_id);
}

/// Build a `LocalAborted` telemetry event for a resource-driven cloud-fallback
/// abort.
///
/// The event carries the abort reason, the local leg's latency and partial
/// token count, and a `correlation_id` shared with the paired `CloudRetry`
/// event so analytics joins on the two halves of one logical inference.
/// `correlation_id`, `outcome_category`, and `abort_reason` ride in
/// `event.data`; the platform-event hoist (see `convert_to_platform_event`)
/// promotes them to top-level columns.
pub fn local_aborted_event(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    latency_ms: u32,
    tokens_emitted: u32,
) -> TelemetryEvent {
    local_aborted_event_with_details(
        correlation_id,
        model_id,
        abort_reason,
        latency_ms,
        tokens_emitted,
        None,
        None,
    )
}

/// Build a `LocalAborted` event with resource and reliability context.
///
/// This is the cloud-fallback telemetry shape used by customer-ready routing:
/// analytics can join the local/cloud legs by `correlation_id`, inspect the
/// live resource summary that caused the abort, and feed the local reliability
/// hint back into fleet-aware routing decisions.
pub fn local_aborted_event_with_details(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    latency_ms: u32,
    tokens_emitted: u32,
    resource_summary: Option<ResourceUsageSummary>,
    local_reliability_hint: Option<LocalReliabilityHint>,
) -> TelemetryEvent {
    let mut data = serde_json::json!({
        "model_id": model_id,
        "correlation_id": correlation_id,
        "outcome_category": {
            "kind": "aborted_for_cloud_fallback",
            "reason": abort_reason.as_str(),
        },
        "abort_reason": abort_reason.as_str(),
        "tokens_emitted": tokens_emitted,
    });

    if let Some(obj) = data.as_object_mut() {
        if let Some(summary) =
            resource_summary.and_then(|summary| serde_json::to_value(summary).ok())
        {
            obj.insert("resource_summary".to_string(), summary);
        }
        if let Some(hint) = local_reliability_hint {
            obj.insert(
                "local_reliability_hint".to_string(),
                serde_json::json!({
                    "recent_abort_rate": hint.recent_abort_rate,
                    "sample_size": hint.sample_size,
                }),
            );
        }
    }

    TelemetryEvent {
        event_type: "LocalAborted".to_string(),
        stage_name: Some(model_id.to_string()),
        target: Some("local".to_string()),
        latency_ms: Some(latency_ms),
        error: None,
        data: Some(data.to_string()),
        timestamp_ms: SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0),
    }
}

/// Convenience wrapper: build and publish a `LocalAborted` event.
pub fn publish_local_aborted(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    latency_ms: u32,
    tokens_emitted: u32,
) {
    publish_telemetry_event(local_aborted_event(
        correlation_id,
        model_id,
        abort_reason,
        latency_ms,
        tokens_emitted,
    ));
}

pub(crate) fn publish_local_aborted_with_details(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    latency_ms: u32,
    tokens_emitted: u32,
    resource_summary: Option<ResourceUsageSummary>,
    local_reliability_hint: Option<LocalReliabilityHint>,
) {
    publish_telemetry_event(local_aborted_event_with_details(
        correlation_id,
        model_id,
        abort_reason,
        latency_ms,
        tokens_emitted,
        resource_summary,
        local_reliability_hint,
    ));
}

pub(crate) fn redact_error_for_telemetry(message: &str) -> String {
    let redacted = [
        "Authorization: Bearer ",
        "authorization: Bearer ",
        "Bearer ",
        "api_key=",
        "api-key=",
        "x-api-key: ",
        "X-API-Key: ",
    ]
    .iter()
    .fold(message.to_string(), |current, marker| {
        redact_value_after_marker(&current, marker)
    });

    redact_secret_like_tokens(&redacted)
}

fn redact_value_after_marker(input: &str, marker: &str) -> String {
    let mut output = String::with_capacity(input.len());
    let mut rest = input;

    while let Some(index) = rest.find(marker) {
        let marker_end = index + marker.len();
        output.push_str(&rest[..marker_end]);
        rest = &rest[marker_end..];

        let value_len = rest
            .find(|c: char| c.is_whitespace() || matches!(c, ',' | ';' | ')' | ']'))
            .unwrap_or(rest.len());
        output.push_str("[REDACTED]");
        rest = &rest[value_len..];
    }

    output.push_str(rest);
    output
}

fn redact_secret_like_tokens(input: &str) -> String {
    input
        .split_whitespace()
        .map(redact_secret_like_token)
        .collect::<Vec<_>>()
        .join(" ")
}

fn redact_secret_like_token(token: &str) -> String {
    let trimmed =
        token.trim_matches(|c: char| !(c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.')));
    if trimmed.starts_with("sk_")
        || trimmed.starts_with("sk-")
        || trimmed.starts_with("hf_")
        || trimmed.starts_with("ghp_")
        || trimmed.starts_with("gho_")
        || trimmed.starts_with("xoxb-")
        || trimmed.starts_with("xoxp-")
    {
        token.replacen(trimmed, "[REDACTED]", 1)
    } else {
        token.to_string()
    }
}

/// Build a terminal event for a local abort whose cloud retry was blocked by
/// live policy. This is the one fallback path that remains a hard failure.
pub fn cloud_denied_by_policy_event(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    policy_reason: &str,
    latency_ms: u32,
) -> TelemetryEvent {
    let policy_reason = redact_error_for_telemetry(policy_reason);
    TelemetryEvent {
        event_type: "LocalFailed".to_string(),
        stage_name: Some(model_id.to_string()),
        target: Some("local".to_string()),
        latency_ms: Some(latency_ms),
        error: Some(format!("cloud_denied_by_policy: {}", policy_reason)),
        data: Some(
            serde_json::json!({
                "model_id": model_id,
                "correlation_id": correlation_id,
                "outcome_category": {
                    "kind": "hard_fail",
                    "reason": "cloud_denied_by_policy",
                },
                "terminal_state_tag": {
                    "kind": "cloud_denied_by_policy",
                    "abort_reason": abort_reason.as_str(),
                },
                "abort_reason": abort_reason.as_str(),
                "policy_reason": policy_reason,
                "status": "error",
            })
            .to_string(),
        ),
        timestamp_ms: SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0),
    }
}

pub fn publish_cloud_denied_by_policy(
    correlation_id: &str,
    model_id: &str,
    abort_reason: xybrid_core::abort::AbortReason,
    policy_reason: &str,
    latency_ms: u32,
) {
    publish_telemetry_event(cloud_denied_by_policy_event(
        correlation_id,
        model_id,
        abort_reason,
        policy_reason,
        latency_ms,
    ));
}

/// Build a `CloudRetry` telemetry event for the cloud leg of a fallback run.
///
/// Emits with the shared `correlation_id` and the cloud provider name when
/// known. Pair with [`local_aborted_event`] (same `correlation_id`) to
/// reconstruct one logical inference across two execution targets.
///
/// `error` is `None` when the cloud leg streamed successfully; `Some(msg)`
/// when it failed before producing a final result. The failure branch sets
/// `event.error` and tags the payload with `status = "error"` so traces
/// list views and dashboards can distinguish a cloud-attempt-that-failed
/// from a cloud-attempt-that-succeeded.
pub fn cloud_retry_event(
    correlation_id: &str,
    model_id: &str,
    provider: Option<&str>,
    latency_ms: u32,
    tokens_emitted: u32,
    error: Option<&str>,
) -> TelemetryEvent {
    let provider_value = provider.unwrap_or("xybrid");
    let redacted_error = error.map(redact_error_for_telemetry);
    let (status, outcome_category) = if let Some(reason) = redacted_error.as_deref() {
        (
            "error",
            serde_json::json!({
                "kind": "hard_fail",
                "reason": reason,
            }),
        )
    } else {
        ("ok", serde_json::json!("cloud_success"))
    };
    TelemetryEvent {
        event_type: "CloudRetry".to_string(),
        stage_name: Some(model_id.to_string()),
        target: Some("cloud".to_string()),
        latency_ms: Some(latency_ms),
        error: redacted_error,
        data: Some(
            serde_json::json!({
                "model_id": model_id,
                "correlation_id": correlation_id,
                "provider": provider_value,
                "outcome_category": outcome_category,
                "tokens_emitted": tokens_emitted,
                "status": status,
            })
            .to_string(),
        ),
        timestamp_ms: SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0),
    }
}

/// Convenience wrapper: build and publish a `CloudRetry` event.
pub fn publish_cloud_retry(
    correlation_id: &str,
    model_id: &str,
    provider: Option<&str>,
    latency_ms: u32,
    tokens_emitted: u32,
    error: Option<&str>,
) {
    publish_telemetry_event(cloud_retry_event(
        correlation_id,
        model_id,
        provider,
        latency_ms,
        tokens_emitted,
        error,
    ));
}

/// Initialize platform telemetry from environment variables
///
/// Returns `true` if initialization succeeded, `false` if XYBRID_API_KEY is not set.
/// Also enables span tracing in xybrid-core for detailed execution profiling.
pub fn init_platform_telemetry_from_env() -> bool {
    if let Some(exporter) = HttpTelemetryExporter::from_env() {
        // Enable span tracing in xybrid-core for execution profiling
        core_tracing::init_tracing(true);

        // Resource-telemetry: the env path has no TelemetryConfig, so the
        // env var is the only way in. Apply the same resolver + activation
        // pair `init_platform_telemetry` uses so consumers that rely on
        // `XYBRID_RESOURCE_TELEMETRY=summary` get identical behavior
        // regardless of which init entry point they chose.
        let resource_mode = resolve_resource_telemetry_mode(ResourceTelemetryMode::Off);
        activate_resource_telemetry(resource_mode);

        // Register automatic execution listener
        register_execution_listener();

        exporter.start();
        let sender = exporter.create_sender();
        register_telemetry_sender(sender);

        if let Ok(mut global) = PLATFORM_EXPORTER.write() {
            *global = Some(exporter);
        }
        true
    } else {
        false
    }
}

/// Set pipeline context for event enrichment
pub fn set_telemetry_pipeline_context(pipeline_id: Option<Uuid>, trace_id: Option<Uuid>) {
    let mut event_context = xybrid_core::event_bus::EventContext::current();
    event_context.pipeline_id = pipeline_id;
    event_context.trace_id = trace_id;
    if event_context.is_empty() {
        xybrid_core::event_bus::clear_current_event_context();
    } else {
        xybrid_core::event_bus::set_current_event_context(event_context);
    }

    if let Ok(exporter) = PLATFORM_EXPORTER.read() {
        if let Some(exp) = exporter.as_ref() {
            exp.set_pipeline_context(pipeline_id, trace_id);
        }
    }
}

/// RAII guard that installs a pipeline context on construction and
/// restores the previous context on drop.
///
/// Callers (e.g. `Pipeline::run`, `XybridModel::run_with_context`) need
/// to scope a per-invocation `trace_id` so every telemetry event
/// emitted during the call shares it. Pairing the install with a manual
/// `set_telemetry_pipeline_context(None, None)` on every exit path is
/// error-prone — any `?` between install and clear leaks the
/// `trace_id` onto subsequent unrelated telemetry on the same thread,
/// and a panic skips the cleanup entirely. This guard removes the
/// duplication and closes the panic-leak hole.
///
/// On `Drop` the guard restores whatever context was in place before
/// `install` ran (rather than blanket-clearing), so nested installs
/// compose correctly. The exporter's pipeline context is held behind
/// a process-global `Arc<RwLock<...>>`; for fully race-free per-call
/// scoping under concurrent threads, callers should also rely on the
/// thread-local `EventContext` (see `xybrid-core::event_bus`) which
/// `set_telemetry_pipeline_context` keeps in sync.
pub(crate) struct TelemetryPipelineContextGuard {
    previous_pipeline_id: Option<Uuid>,
    previous_trace_id: Option<Uuid>,
}

impl TelemetryPipelineContextGuard {
    /// Install a pipeline context for the lifetime of the guard.
    pub(crate) fn install(pipeline_id: Option<Uuid>, trace_id: Option<Uuid>) -> Self {
        let (previous_pipeline_id, previous_trace_id) = current_telemetry_pipeline_context();
        set_telemetry_pipeline_context(pipeline_id, trace_id);
        Self {
            previous_pipeline_id,
            previous_trace_id,
        }
    }
}

impl Drop for TelemetryPipelineContextGuard {
    fn drop(&mut self) {
        set_telemetry_pipeline_context(self.previous_pipeline_id, self.previous_trace_id);
    }
}

/// Register the execution listener that converts `ExecutionEvent`s from
/// xybrid-core's `TemplateExecutor` into `TelemetryEvent`s and publishes them.
fn register_execution_listener() {
    execution_listener::set_execution_listener(|event| {
        let timestamp_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;

        let telemetry_event = match event {
            ExecutionEvent::Started { model_id, method } => TelemetryEvent {
                event_type: "ExecutionStarted".to_string(),
                stage_name: Some(method),
                target: Some("device".to_string()),
                latency_ms: None,
                error: None,
                data: Some(serde_json::json!({ "model": model_id }).to_string()),
                timestamp_ms,
            },
            ExecutionEvent::Completed {
                model_id,
                method,
                latency_ms,
            } => TelemetryEvent {
                event_type: "ExecutionCompleted".to_string(),
                stage_name: Some(method),
                target: Some("device".to_string()),
                latency_ms: Some(latency_ms as u32),
                error: None,
                data: Some(serde_json::json!({ "model": model_id }).to_string()),
                timestamp_ms,
            },
            ExecutionEvent::Failed {
                model_id,
                method,
                latency_ms,
                error,
            } => TelemetryEvent {
                event_type: "ExecutionFailed".to_string(),
                // Surface the model_id in the operation column so error
                // rows on the Traces dashboard read like the success rows
                // (`pipeline / <model-id>`) instead of the executor-
                // internal method name. The method is still preserved
                // in `data` for forensics.
                stage_name: Some(model_id.clone()),
                target: Some("device".to_string()),
                latency_ms: Some(latency_ms as u32),
                error: Some(error),
                data: Some(
                    serde_json::json!({
                        "model": model_id,
                        "method": method,
                    })
                    .to_string(),
                ),
                timestamp_ms,
            },
        };

        publish_telemetry_event(telemetry_event);
    });
}

/// Flush all pending telemetry events
pub fn flush_platform_telemetry() {
    if let Ok(exporter) = PLATFORM_EXPORTER.read() {
        if let Some(exp) = exporter.as_ref() {
            exp.flush();
        }
    }
}

/// Shutdown platform telemetry exporter
///
/// This also disables span tracing in xybrid-core.
pub fn shutdown_platform_telemetry() {
    // Disable span tracing
    core_tracing::init_tracing(false);

    // Remove automatic execution listener
    execution_listener::clear_execution_listener();

    if let Ok(mut exporter) = PLATFORM_EXPORTER.write() {
        if let Some(exp) = exporter.take() {
            exp.stop();
        }
    }
}

/// Register a telemetry event sender
pub fn register_telemetry_sender(sender: TelemetrySender) {
    // Use if-let to gracefully handle poisoned mutex
    if let Ok(mut senders) = TELEMETRY_SENDERS.lock() {
        senders.push(sender);
    }
}

/// Convert OrchestratorEvent to TelemetryEvent
pub fn convert_orchestrator_event(event: &OrchestratorEvent) -> TelemetryEvent {
    let timestamp_ms = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64;

    let context = event.context().clone();
    let telemetry_event = match event {
        OrchestratorEvent::PipelineStart { stages, .. } => TelemetryEvent {
            event_type: "PipelineStart".to_string(),
            stage_name: None,
            target: None,
            latency_ms: None,
            error: None,
            data: Some(serde_json::json!({"stages": stages}).to_string()),
            timestamp_ms,
        },
        OrchestratorEvent::PipelineComplete {
            total_latency_ms, ..
        } => TelemetryEvent {
            event_type: "PipelineComplete".to_string(),
            stage_name: None,
            target: None,
            latency_ms: Some(*total_latency_ms),
            error: None,
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::StageStart { stage_name, .. } => TelemetryEvent {
            event_type: "StageStart".to_string(),
            stage_name: Some(stage_name.clone()),
            target: None,
            latency_ms: None,
            error: None,
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::StageComplete {
            stage_name,
            target,
            latency_ms,
            ..
        } => TelemetryEvent {
            event_type: "StageComplete".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: Some(*latency_ms),
            error: None,
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::StageError {
            stage_name, error, ..
        } => TelemetryEvent {
            event_type: "StageError".to_string(),
            stage_name: Some(stage_name.clone()),
            target: None,
            latency_ms: None,
            error: Some(error.clone()),
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::RoutingDecided {
            stage_name,
            target,
            reason,
            recent_abort_rate,
            sample_size,
            ..
        } => TelemetryEvent {
            event_type: "RoutingDecided".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: None,
            error: None,
            // Embed the local reliability hint at the top level of event.data
            // so the `convert_to_platform_event` hoist list can surface it on
            // the payload at the top level (where the platform datasource's
            // JSONPath extractors live, alongside correlation_id / outcome_*).
            // Non-finite f32 values would serialize to JSON `null`, breaking
            // the typed platform column; the authority emits only finite
            // values, but we sanitize at the bridge for defense in depth.
            data: Some(
                serde_json::json!({
                    "reason": reason,
                    "local_reliability_hint": {
                        "recent_abort_rate": if recent_abort_rate.is_finite() {
                            *recent_abort_rate
                        } else {
                            0.0_f32
                        },
                        "sample_size": sample_size,
                    },
                })
                .to_string(),
            ),
            timestamp_ms,
        },
        OrchestratorEvent::ExecutionStarted {
            stage_name, target, ..
        } => TelemetryEvent {
            event_type: "ExecutionStarted".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: None,
            error: None,
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::ExecutionCompleted {
            stage_name,
            target,
            execution_time_ms,
            ..
        } => TelemetryEvent {
            event_type: "ExecutionCompleted".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: Some(*execution_time_ms),
            error: None,
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::ExecutionFailed {
            stage_name,
            target,
            error,
            ..
        } => TelemetryEvent {
            event_type: "ExecutionFailed".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: None,
            error: Some(error.clone()),
            data: None,
            timestamp_ms,
        },
        OrchestratorEvent::PolicyEvaluated {
            stage_name,
            allowed,
            reason,
            ..
        } => TelemetryEvent {
            event_type: "PolicyEvaluated".to_string(),
            stage_name: Some(stage_name.clone()),
            target: None,
            latency_ms: None,
            error: if *allowed {
                None
            } else {
                reason.clone().or(Some("Policy violation".to_string()))
            },
            data: Some(
                serde_json::json!({
                    "allowed": allowed,
                    "reason": reason
                })
                .to_string(),
            ),
            timestamp_ms,
        },
        OrchestratorEvent::LocalAborted {
            stage_name,
            target,
            reason,
            ..
        } => TelemetryEvent {
            event_type: "LocalAborted".to_string(),
            stage_name: Some(stage_name.clone()),
            target: Some(target.clone()),
            latency_ms: None,
            error: Some(reason.clone()),
            data: Some(serde_json::json!({ "reason": reason }).to_string()),
            timestamp_ms,
        },
        _ => TelemetryEvent {
            event_type: format!("{:?}", event),
            stage_name: None,
            target: None,
            latency_ms: None,
            error: None,
            data: Some(format!("{:?}", event)),
            timestamp_ms,
        },
    };

    attach_event_context(telemetry_event, &context)
}

fn attach_event_context(event: TelemetryEvent, context: &EventContext) -> TelemetryEvent {
    if context.is_empty() {
        return event;
    }

    let mut data = match event.data.as_ref() {
        Some(raw) => match serde_json::from_str::<serde_json::Value>(raw) {
            Ok(value) if value.is_object() => value,
            Ok(value) => serde_json::json!({ "value": value }),
            Err(_) => serde_json::json!({ "value": raw }),
        },
        None => serde_json::json!({}),
    };

    if let Some(obj) = data.as_object_mut() {
        if let Some(id) = context.pipeline_id {
            obj.entry(CONTEXT_PIPELINE_ID_KEY.to_string())
                .or_insert_with(|| serde_json::json!(id.to_string()));
        }
        if let Some(id) = context.trace_id {
            obj.entry(CONTEXT_TRACE_ID_KEY.to_string())
                .or_insert_with(|| serde_json::json!(id.to_string()));
        }
        if let Some(id) = context.correlation_id.as_ref() {
            obj.entry("correlation_id".to_string())
                .or_insert_with(|| serde_json::json!(id));
        }
        if let Some(id) = context.request_id.as_ref() {
            obj.entry("request_id".to_string())
                .or_insert_with(|| serde_json::json!(id));
        }
        if let Some(id) = context.model_id.as_ref() {
            obj.entry("model_id".to_string())
                .or_insert_with(|| serde_json::json!(id));
        }
        if let Some(id) = context.span_id.as_ref() {
            obj.entry("span_id".to_string())
                .or_insert_with(|| serde_json::json!(id));
        }
    }

    TelemetryEvent {
        data: Some(data.to_string()),
        ..event
    }
}

/// Synchronous span snapshot for publish. Only acts on completion-family
/// events (PipelineComplete / ModelComplete) where the dashboard wants a
/// flamegraph; skips if the caller already embedded their own spans.
/// When it does act, it drains the global collector so the next unit of
/// work on this thread starts with a clean slate.
fn snapshot_spans_into_event(event: TelemetryEvent) -> TelemetryEvent {
    // Span-bearing event types — see the matching list in
    // `convert_to_platform_event` for why LocalAborted/CloudRetry are here.
    // The cloud-fallback flow never fires a ModelComplete/PipelineComplete,
    // so without these the cloud-leg `SpanGuard`s in
    // `runtime_adapter/cloud/mod.rs` get stranded in the global collector
    // and the dashboard's flamegraph stays empty.
    //
    // `ModelWarmup` follows the same shape: `XybridModel::warmup` opens
    // `execute:<model>` + `llm_inference` spans via the executor before
    // publishing. Without this entry those spans would (a) never reach
    // the warmup event's payload — the dashboard falls back to a
    // synthesized placeholder flamegraph — and (b) leak into the next
    // event's snapshot (typically the first real inference of the
    // session), giving that inference's trace two stray spans it
    // didn't generate.
    let is_span_bearing = matches!(
        event.event_type.as_str(),
        "PipelineComplete" | "ModelComplete" | "ModelWarmup" | "LocalAborted" | "CloudRetry"
    );
    if !is_span_bearing || !core_tracing::is_tracing_enabled() {
        return event;
    }

    // If the caller already has spans in their data blob, leave it
    // alone — they opted into managing their own lifecycle (Mirage
    // does this for cross-chunk isolation).
    let data_already_has_spans = event
        .data
        .as_ref()
        .and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
        .and_then(|v| v.get("spans").cloned())
        .filter(|v| !v.is_null())
        .is_some();
    if data_already_has_spans {
        return event;
    }

    let captured = core_tracing::get_stages_json();
    core_tracing::reset_tracing();

    // Merge captured spans into event.data["spans"].
    let mut merged: serde_json::Value = event
        .data
        .as_ref()
        .and_then(|s| serde_json::from_str(s).ok())
        .unwrap_or_else(|| serde_json::json!({}));
    if !merged.is_object() {
        merged = serde_json::json!({});
    }
    if let Some(spans) = captured.get("spans") {
        merged["spans"] = spans.clone();
    }

    TelemetryEvent {
        data: Some(merged.to_string()),
        ..event
    }
}

fn current_telemetry_pipeline_context() -> (Option<Uuid>, Option<Uuid>) {
    if let Ok(exporter) = PLATFORM_EXPORTER.read() {
        if let Some(exp) = exporter.as_ref() {
            let pipeline_id = exp.pipeline_id.read().ok().and_then(|g| *g);
            let trace_id = exp.trace_id.read().ok().and_then(|g| *g);
            return (pipeline_id, trace_id);
        }
    }
    (None, None)
}

fn snapshot_context_into_event(event: TelemetryEvent) -> TelemetryEvent {
    let (pipeline_id, trace_id) = current_telemetry_pipeline_context();
    snapshot_context_into_event_with(event, pipeline_id, trace_id)
}

fn snapshot_context_into_event_with(
    event: TelemetryEvent,
    pipeline_id: Option<Uuid>,
    trace_id: Option<Uuid>,
) -> TelemetryEvent {
    if pipeline_id.is_none() && trace_id.is_none() {
        return event;
    }

    let mut data = match event.data.as_ref() {
        Some(raw) => match serde_json::from_str::<serde_json::Value>(raw) {
            Ok(value) if value.is_object() => value,
            Ok(value) => serde_json::json!({ "value": value }),
            Err(_) => serde_json::json!({ "value": raw }),
        },
        None => serde_json::json!({}),
    };

    if let Some(obj) = data.as_object_mut() {
        if let Some(id) = pipeline_id {
            obj.insert(CONTEXT_PIPELINE_ID_KEY.to_string(), serde_json::json!(id));
        }
        if let Some(id) = trace_id {
            obj.insert(CONTEXT_TRACE_ID_KEY.to_string(), serde_json::json!(id));
        }
    }

    TelemetryEvent {
        data: Some(data.to_string()),
        ..event
    }
}

/// Build a `ModelDownload` event for a successful registry fetch.
///
/// Carries the cost-attribution fields agreed with the platform schema:
/// `model_id` (registry mask), `bytes_downloaded` (final on-disk size of
/// the model file or .xyb bundle), `source` (canonical download host
/// label — `r2` for Xybrid's R2 mirror, `huggingface` for direct HF
/// pulls; other hosts pass through as-is so a future provider doesn't
/// silently lose attribution), and `duration_ms` (wallclock time spent
/// inside the network download, excluding hash verification and cache
/// extraction so the field reflects bytes-on-the-wire latency).
///
/// The wire `event_type` is the literal string `"ModelDownload"`. The
/// fields land under `payload.data` after `convert_to_platform_event`
/// runs; ingest reads them from there.
///
/// Pure builder — does not publish, allows unit tests to inspect the
/// event shape without standing up the global exporter.
fn build_model_download_event(
    model_id: &str,
    bytes_downloaded: u64,
    source: &str,
    duration_ms: u32,
) -> TelemetryEvent {
    let now_ms = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0);
    let data = serde_json::json!({
        "model_id": model_id,
        "bytes_downloaded": bytes_downloaded,
        "source": source,
        "duration_ms": duration_ms,
    });
    TelemetryEvent {
        event_type: "ModelDownload".to_string(),
        stage_name: None,
        target: None,
        // Mirror duration onto the canonical `latency_ms` field so the
        // platform's existing latency column lights up for downloads
        // without a schema migration.
        latency_ms: Some(duration_ms),
        error: None,
        data: Some(data.to_string()),
        timestamp_ms: now_ms,
    }
}

/// Publish a `ModelDownload` event for a successful registry fetch.
///
/// Honors [`crate::telemetry_optout::is_telemetry_opted_out`]: when the
/// user has set `XYBRID_TELEMETRY_OPTOUT=1` no event is emitted, since
/// the same opt-out already gates the registry call telemetry header
/// and a model-download event would leak the same attribution surface
/// (which model the user pulled, how big it was).
///
/// Cache hits MUST NOT call this — the event represents a real network
/// transfer for cost accounting; emitting it on cache hits would
/// double-count and hide cache-effectiveness metrics.
pub fn publish_model_download(
    model_id: &str,
    bytes_downloaded: u64,
    source: &str,
    duration_ms: u32,
) {
    if crate::telemetry_optout::is_telemetry_opted_out() {
        return;
    }
    let event = build_model_download_event(model_id, bytes_downloaded, source, duration_ms);
    publish_telemetry_event(event);
}

/// Publish a telemetry event to all registered subscribers
pub fn publish_telemetry_event(event: TelemetryEvent) {
    // Span snapshot: capture spans synchronously at publish time.
    //
    // The exporter thread (`convert_to_platform_event`) used to do this
    // asynchronously, which caused a race when a caller published
    // ModelComplete/PipelineComplete events back-to-back: any span opened
    // by the next unit of work (e.g. Mirage's chunked summarize opening
    // `cloud_synthesize` immediately after a local classifier's event
    // published) could be stolen or reset by the exporter while still
    // open, producing empty spans / 0ms durations in the flamegraph.
    //
    // Capturing here instead, on the PUBLISHING thread, anchors each
    // event's span tree to the exact state of the global collector at
    // the moment the caller finished emitting it. Subsequent span work
    // on the same thread is cleanly separated. Callers that already
    // embedded `spans` in `event.data` (e.g. Mirage's `summarize.rs`
    // `synthesize_chunk_cloud`, which captures+resets earlier for
    // composability) are left untouched so they keep full control.
    let event = snapshot_spans_into_event(event);
    let event = snapshot_context_into_event(event);
    dispatch_telemetry_event(event);
}

pub(crate) fn publish_telemetry_event_in_context(
    event: TelemetryEvent,
    pipeline_id: Option<Uuid>,
    trace_id: Option<Uuid>,
) {
    let event = snapshot_spans_into_event(event);
    let event = snapshot_context_into_event_with(event, pipeline_id, trace_id);
    dispatch_telemetry_event(event);
}

fn dispatch_telemetry_event(event: TelemetryEvent) {
    // Use unwrap_or_else to recover from poisoned mutex - this prevents
    // a panic in one component from permanently breaking telemetry
    let Ok(senders) = TELEMETRY_SENDERS.lock() else {
        // Mutex is poisoned, silently skip telemetry rather than crash
        return;
    };
    let mut dead_senders = Vec::new();

    for (idx, sender) in senders.iter().enumerate() {
        if sender.send(event.clone()).is_err() {
            dead_senders.push(idx);
        }
    }

    // Remove dead senders
    drop(senders);
    if !dead_senders.is_empty() {
        if let Ok(mut senders) = TELEMETRY_SENDERS.lock() {
            for idx in dead_senders.iter().rev() {
                senders.remove(*idx);
            }
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum BridgeError {
    #[error("orchestrator event bridge is no longer running")]
    Stopped,
    #[error("orchestrator event bridge flush acknowledgement was dropped")]
    FlushAckDropped,
    #[error("orchestrator event bridge thread panicked")]
    ThreadPanicked,
}

enum BridgeCommand {
    Flush(mpsc::Sender<()>),
}

/// Handle for a scoped orchestrator-to-telemetry event bridge.
#[must_use = "BridgeHandle must be flushed or joined, otherwise queued orchestrator telemetry can be dropped"]
pub struct BridgeHandle {
    join_handle: thread::JoinHandle<()>,
    command_tx: mpsc::Sender<BridgeCommand>,
}

impl BridgeHandle {
    /// Block until all events queued before this call have been delivered to
    /// the SDK telemetry stream.
    pub fn flush(&self) -> Result<(), BridgeError> {
        let (ack_tx, ack_rx) = mpsc::channel();
        self.command_tx
            .send(BridgeCommand::Flush(ack_tx))
            .map_err(|_| BridgeError::Stopped)?;
        ack_rx.recv().map_err(|_| BridgeError::FlushAckDropped)
    }

    /// Compatibility alias for callers that used the previous scoped bridge
    /// drain API. New code should prefer [`Self::flush`] and handle errors.
    pub fn drain(&self) {
        let _ = self.flush();
    }

    /// Wait for the bridge thread to finish draining after the orchestrator's
    /// event bus has been dropped.
    pub fn join(self) -> Result<(), BridgeError> {
        let Self {
            join_handle,
            command_tx,
        } = self;
        drop(command_tx);
        join_handle.join().map_err(|_| BridgeError::ThreadPanicked)
    }
}

pub type OrchestratorEventBridge = BridgeHandle;

const BRIDGE_POLL_INTERVAL: Duration = Duration::from_millis(1);

/// Bridge orchestrator events to telemetry stream.
///
/// This function subscribes to orchestrator events and converts them to
/// telemetry events. The returned handle must be joined after the orchestrator
/// is dropped so queued events are drained before the caller tears down
/// telemetry subscribers or the runtime.
pub fn bridge_orchestrator_events(
    orchestrator: &xybrid_core::orchestrator::Orchestrator,
) -> BridgeHandle {
    let event_bus = orchestrator.event_bus();
    let subscription = event_bus.subscribe();
    let (command_tx, command_rx) = mpsc::channel();

    let join_handle = thread::spawn(move || bridge_loop(subscription, command_rx));

    BridgeHandle {
        join_handle,
        command_tx,
    }
}

/// Subscribe to orchestrator events and return a drainable bridge handle.
pub fn subscribe_orchestrator_events(
    orchestrator: &xybrid_core::orchestrator::Orchestrator,
) -> BridgeHandle {
    bridge_orchestrator_events(orchestrator)
}

pub(crate) fn subscribe_orchestrator_events_in_context(
    orchestrator: &xybrid_core::orchestrator::Orchestrator,
    _pipeline_id: Option<Uuid>,
    _trace_id: Option<Uuid>,
) -> BridgeHandle {
    bridge_orchestrator_events(orchestrator)
}

fn bridge_loop(
    subscription: xybrid_core::event_bus::Subscription,
    command_rx: mpsc::Receiver<BridgeCommand>,
) {
    loop {
        drain_bridge_commands(&subscription, &command_rx);
        match subscription.recv_timeout(BRIDGE_POLL_INTERVAL) {
            Ok(event) => publish_orchestrator_event(event),
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                drain_available_orchestrator_events(&subscription);
                drain_bridge_commands(&subscription, &command_rx);
                break;
            }
        }
    }
}

fn drain_bridge_commands(
    subscription: &xybrid_core::event_bus::Subscription,
    command_rx: &mpsc::Receiver<BridgeCommand>,
) {
    while let Ok(command) = command_rx.try_recv() {
        match command {
            BridgeCommand::Flush(ack_tx) => {
                drain_available_orchestrator_events(subscription);
                let _ = ack_tx.send(());
            }
        }
    }
}

fn drain_available_orchestrator_events(subscription: &xybrid_core::event_bus::Subscription) {
    loop {
        match subscription.try_recv() {
            Ok(event) => publish_orchestrator_event(event),
            Err(mpsc::TryRecvError::Empty | mpsc::TryRecvError::Disconnected) => break,
        }
    }
}

/// Returns `true` when the orchestrator event is wire-noise that the SDK
/// already covers via its own publish sites, and should be dropped before
/// reaching the registered telemetry sender.
///
/// Single-stage local LLM calls were producing 6–9 dashboard rows per
/// turn because the orchestrator emits a full pipeline-execution event
/// chain (`PipelineStart`, `StageStart`, `ExecutionStarted`,
/// `ExecutionCompleted`, `StageComplete`, `PipelineComplete`) on top of
/// the SDK's own `ModelComplete` / `PipelineComplete` — every one of
/// these landed as its own row. The orchestrator-side events are useful
/// for in-process subscribers (other code listening on the event bus)
/// but they're noise at the wire boundary, where the user-facing
/// contract is one row per turn (the SDK's completion event).
///
/// Keep: `PolicyEvaluated`, `RoutingDecided` (routing-decision metadata
/// shown alongside the completion row), `StageError`, `ExecutionFailed`
/// (errors must always reach the wire).
///
/// Drop: pipeline-frame and per-stage success-path events. Errors on
/// these paths still surface via the `*Failed` / `*Error` variants
/// above, which the filter passes through.
fn orchestrator_event_is_wire_noise(event: &OrchestratorEvent) -> bool {
    matches!(
        event,
        OrchestratorEvent::PipelineStart { .. }
            | OrchestratorEvent::PipelineComplete { .. }
            | OrchestratorEvent::StageStart { .. }
            | OrchestratorEvent::StageComplete { .. }
            | OrchestratorEvent::ExecutionStarted { .. }
            | OrchestratorEvent::ExecutionCompleted { .. }
    )
}

fn publish_orchestrator_event(event: OrchestratorEvent) {
    if orchestrator_event_is_wire_noise(&event) {
        return;
    }
    let telemetry_event = convert_orchestrator_event(&event);
    publish_telemetry_event(telemetry_event);
}

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

    static TELEMETRY_SENDER_TEST_LOCK: Mutex<()> = Mutex::new(());

    #[test]
    fn platform_event_payload_has_no_legacy_cache_keys() {
        // End-to-end through convert_to_platform_event so we exercise
        // the real hoist list + serde path. A ModelComplete event
        // carrying a data blob with canonical cache keys must
        // serialize to a payload that contains the canonical names and
        // none of the DeepSeek-specific legacy names. This backs up
        // the source-containment test (`tests/legacy_cache_names.rs`)
        // with a runtime-format assertion: if a field rename is
        // incomplete — e.g., a `#[serde(rename = "cache_hit_tokens")]`
        // is forgotten or a manual json literal writes the old key —
        // this catches it even when the source file reads canonical.
        let data = serde_json::json!({
            "model_id": "deepseek-chat",
            "tokens_in": 1000,
            "tokens_out": 120,
            "cost_usd": 0.00123,
            "cache_read_input_tokens": 800,
            "cache_read_cost_usd": 0.000056,
            "uncached_input_cost_usd": 0.000054,
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("cloud_synthesize".to_string()),
            target: Some("cloud".to_string()),
            latency_ms: Some(420),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform_event = convert_to_platform_event(&event, &config, None, None, None);
        let payload_json = serde_json::to_string(&platform_event.payload).unwrap();

        // Forbidden keys built at runtime so the source of this file
        // doesn't contain the literals (5a would flag them otherwise).
        let forbidden = [
            format!("cache{}hit{}tokens", "_", "_"),
            format!("cache{}miss{}tokens", "_", "_"),
            format!("cache{}hit{}cost{}usd", "_", "_", "_"),
            format!("cache{}miss{}cost{}usd", "_", "_", "_"),
        ];
        for key in &forbidden {
            assert!(
                !payload_json.contains(key),
                "platform event payload leaked legacy key {key}: {payload_json}"
            );
        }
        // Positive assertion — the canonical name IS present. Without
        // this, an empty payload would trivially satisfy the negation
        // above.
        assert!(payload_json.contains("cache_read_input_tokens"));
    }

    #[test]
    fn resource_summary_attaches_and_hoists_through_convert() {
        // End-to-end happy path: attach_resource_summary()
        // edits event.data, then convert_to_platform_event hoists
        // resource_summary to the platform-event payload top level
        // (same hoist mechanism as cache tokens).
        let summary = ResourceUsageSummary {
            cpu_avg_pct: Some(34.1),
            cpu_peak_pct: Some(62.5),
            process_rss_peak_mb: Some(712),
            available_mem_min_mb: Some(4180),
            memory_pressure_peak: xybrid_core::device::MemoryPressure::Normal,
            thermal_state_peak: xybrid_core::device::ThermalState::Normal,
            battery_pct_end: Some(72),
            sample_count: 4,
            sampling_mode: "summary".to_string(),
            sampling_interval_ms: Some(1000),
        };
        let mut event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("qwen2.5-0.5b".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(420),
            error: None,
            data: Some(serde_json::json!({ "model_id": "qwen2.5-0.5b" }).to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        attach_resource_summary(&mut event, Some(summary));

        // Sanity: attach mutated event.data.
        let data_json = event.data.clone().expect("data present");
        let data: serde_json::Value = serde_json::from_str(&data_json).unwrap();
        assert!(data.get("resource_summary").is_some());
        assert_eq!(
            data["resource_summary"]["cpu_peak_pct"].as_f64(),
            Some(62.5)
        );

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let payload_json = serde_json::to_string(&platform.payload).unwrap();

        // Hoisted to payload top level (sibling of tokens_in / cache_*).
        assert!(
            payload_json.contains("\"resource_summary\""),
            "resource_summary must be hoisted to payload top level, got: {}",
            payload_json
        );
        // And reachable via the object root, not just as a nested string.
        let parsed: serde_json::Value = serde_json::from_str(&payload_json).unwrap();
        assert_eq!(
            parsed["resource_summary"]["cpu_peak_pct"].as_f64(),
            Some(62.5),
            "hoisted value should be the same object we attached"
        );
    }

    #[test]
    fn pipeline_complete_publishes_resource_summary_in_payload() {
        // Parallel coverage to `resource_summary_attaches_and_hoists_through_convert`
        // but for the PipelineComplete path — the hoist logic has no branch
        // on event_type, but we assert the shape end-to-end so a future
        // regression that special-cases Model vs Pipeline won't slip.
        let summary = ResourceUsageSummary {
            cpu_avg_pct: Some(12.0),
            cpu_peak_pct: Some(45.0),
            process_rss_peak_mb: Some(380),
            available_mem_min_mb: Some(6000),
            memory_pressure_peak: xybrid_core::device::MemoryPressure::Normal,
            thermal_state_peak: xybrid_core::device::ThermalState::Normal,
            battery_pct_end: None,
            sample_count: 3,
            sampling_mode: "summary".to_string(),
            sampling_interval_ms: Some(1000),
        };
        let mut event = TelemetryEvent {
            event_type: "PipelineComplete".to_string(),
            stage_name: Some("mirage-document-insights".to_string()),
            target: None,
            latency_ms: Some(1_200),
            error: None,
            data: Some("{\"stages\":[],\"output_type\":\"Text\"}".to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        attach_resource_summary(&mut event, Some(summary));

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let payload_json = serde_json::to_string(&platform.payload).unwrap();

        assert!(
            payload_json.contains("\"resource_summary\""),
            "PipelineComplete should carry resource_summary on payload top level, got: {payload_json}"
        );
        let parsed: serde_json::Value = serde_json::from_str(&payload_json).unwrap();
        assert_eq!(parsed["resource_summary"]["sample_count"].as_i64(), Some(3));
        assert_eq!(
            parsed["resource_summary"]["memory_pressure_peak"].as_str(),
            Some("normal")
        );
    }

    #[test]
    fn routing_outcome_fields_hoist_to_platform_event_top_level() {
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("qwen2.5-0.5b".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(420),
            error: None,
            data: Some(
                serde_json::json!({
                    "model_id": "qwen2.5-0.5b",
                    "correlation_id": "run-string-123",
                    "outcome_category": {
                        "kind": "aborted_for_cloud_fallback",
                        "reason": "stress_memory"
                    },
                    "abort_reason": "stress_memory"
                })
                .to_string(),
            ),
            timestamp_ms: 1_700_000_000_000,
        };

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);

        assert_eq!(platform.correlation_id.as_deref(), Some("run-string-123"));
        assert_eq!(
            platform
                .outcome_category
                .as_ref()
                .and_then(|v| v.get("kind")),
            Some(&serde_json::json!("aborted_for_cloud_fallback"))
        );
        assert_eq!(platform.abort_reason.as_deref(), Some("stress_memory"));
        assert_eq!(platform.payload["correlation_id"], "run-string-123");
        assert_eq!(
            platform.payload["outcome_category"]["kind"],
            "aborted_for_cloud_fallback"
        );
        assert_eq!(platform.payload["abort_reason"], "stress_memory");
    }

    #[test]
    fn local_reliability_hint_hoists_to_platform_event_top_level() {
        // RoutingDecision events serialize the hint into event.data via
        // `Telemetry::log_routing_decision`. The hoist list in
        // convert_to_platform_event must copy that nested object to the
        // payload top level so the analytics backend can column-extract
        // `json:$.local_reliability_hint.recent_abort_rate` and
        // `.sample_size` without descending into `data`.
        let event = TelemetryEvent {
            event_type: "RoutingDecision".to_string(),
            stage_name: Some("stage-1".to_string()),
            target: Some("cloud".to_string()),
            latency_ms: None,
            error: None,
            data: Some(
                serde_json::json!({
                    "stage": "stage-1",
                    "target": "cloud",
                    "reason": "history_bias",
                    "local_reliability_hint": {
                        "recent_abort_rate": 0.75,
                        "sample_size": 4_u32,
                    }
                })
                .to_string(),
            ),
            timestamp_ms: 1_700_000_000_000,
        };

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);

        assert_eq!(
            platform.payload["local_reliability_hint"]["recent_abort_rate"]
                .as_f64()
                .unwrap_or(-1.0),
            0.75
        );
        assert_eq!(
            platform.payload["local_reliability_hint"]["sample_size"].as_i64(),
            Some(4)
        );
    }

    #[test]
    fn empty_local_reliability_hint_still_hoists_with_sample_size_zero() {
        // Empty-window case: window has no entries yet. The SDK still
        // emits (0.0, 0) so the platform can distinguish "no data" from
        // "field missing because the SDK is older than the schema".
        let event = TelemetryEvent {
            event_type: "RoutingDecision".to_string(),
            stage_name: Some("stage-1".to_string()),
            target: Some("local".to_string()),
            latency_ms: None,
            error: None,
            data: Some(
                serde_json::json!({
                    "stage": "stage-1",
                    "target": "local",
                    "reason": "default_local",
                    "local_reliability_hint": {
                        "recent_abort_rate": 0.0,
                        "sample_size": 0_u32,
                    }
                })
                .to_string(),
            ),
            timestamp_ms: 1_700_000_000_000,
        };

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);

        assert_eq!(
            platform.payload["local_reliability_hint"]["sample_size"].as_i64(),
            Some(0)
        );
    }

    #[test]
    fn local_aborted_and_cloud_retry_events_share_correlation_id() {
        let local = local_aborted_event(
            "run-abc-123",
            "qwen2.5-0.5b",
            xybrid_core::abort::AbortReason::StressMemory,
            180,
            4,
        );
        let cloud = cloud_retry_event("run-abc-123", "qwen2.5-0.5b", Some("openai"), 920, 38, None);

        assert_eq!(local.event_type, "LocalAborted");
        assert_eq!(local.target.as_deref(), Some("local"));
        assert_eq!(local.latency_ms, Some(180));

        assert_eq!(cloud.event_type, "CloudRetry");
        assert_eq!(cloud.target.as_deref(), Some("cloud"));
        assert_eq!(cloud.latency_ms, Some(920));

        let local_data: serde_json::Value =
            serde_json::from_str(local.data.as_ref().unwrap()).unwrap();
        let cloud_data: serde_json::Value =
            serde_json::from_str(cloud.data.as_ref().unwrap()).unwrap();

        assert_eq!(local_data["correlation_id"], "run-abc-123");
        assert_eq!(cloud_data["correlation_id"], "run-abc-123");
        assert_eq!(local_data["abort_reason"], "stress_memory");
        assert_eq!(
            local_data["outcome_category"]["kind"],
            "aborted_for_cloud_fallback"
        );
        assert_eq!(local_data["outcome_category"]["reason"], "stress_memory");
        assert_eq!(local_data["tokens_emitted"], 4);
        assert_eq!(cloud_data["provider"], "openai");
        assert_eq!(cloud_data["outcome_category"], "cloud_success");
        assert_eq!(cloud_data["tokens_emitted"], 38);

        // The platform-event hoist promotes `correlation_id` and `abort_reason`
        // out of `event.data` so analytics can join on top-level columns.
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform_local = convert_to_platform_event(&local, &config, None, None, None);
        let platform_cloud = convert_to_platform_event(&cloud, &config, None, None, None);

        assert_eq!(
            platform_local.correlation_id.as_deref(),
            Some("run-abc-123")
        );
        assert_eq!(
            platform_cloud.correlation_id.as_deref(),
            Some("run-abc-123")
        );
        assert_eq!(
            platform_cloud.outcome_category.as_ref(),
            Some(&serde_json::json!("cloud_success"))
        );
        assert_eq!(
            platform_cloud.payload["outcome_category"],
            serde_json::json!("cloud_success")
        );
        assert_eq!(
            platform_local.abort_reason.as_deref(),
            Some("stress_memory")
        );
    }

    #[test]
    fn local_aborted_event_carries_resource_summary_and_reliability_hint() {
        let summary = ResourceUsageSummary {
            cpu_avg_pct: Some(61.0),
            cpu_peak_pct: Some(88.5),
            process_rss_peak_mb: Some(2048),
            available_mem_min_mb: Some(512),
            memory_pressure_peak: xybrid_core::device::MemoryPressure::Critical,
            thermal_state_peak: xybrid_core::device::ThermalState::Hot,
            battery_pct_end: Some(51),
            sample_count: 3,
            sampling_mode: "debug_local".to_string(),
            sampling_interval_ms: Some(100),
        };
        let local = local_aborted_event_with_details(
            "run-rich-123",
            "qwen2.5-0.5b",
            xybrid_core::abort::AbortReason::StressMemory,
            180,
            4,
            Some(summary),
            Some(LocalReliabilityHint {
                recent_abort_rate: 0.5,
                sample_size: 2,
            }),
        );

        let data: serde_json::Value =
            serde_json::from_str(local.data.as_ref().expect("data present")).unwrap();
        assert_eq!(
            data["resource_summary"]["memory_pressure_peak"].as_str(),
            Some("critical")
        );
        assert_eq!(
            data["resource_summary"]["sampling_interval_ms"].as_i64(),
            Some(100)
        );
        assert_eq!(
            data["local_reliability_hint"]["recent_abort_rate"].as_f64(),
            Some(0.5)
        );
        assert_eq!(
            data["local_reliability_hint"]["sample_size"].as_i64(),
            Some(2)
        );

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&local, &config, None, None, None);
        assert_eq!(
            platform.payload["resource_summary"]["sample_count"].as_i64(),
            Some(3)
        );
        assert_eq!(
            platform.payload["local_reliability_hint"]["sample_size"].as_i64(),
            Some(2)
        );
    }

    #[test]
    fn cloud_retry_event_defaults_provider_when_unspecified() {
        let cloud = cloud_retry_event("run-xyz", "model-x", None, 50, 10, None);
        let cloud_data: serde_json::Value =
            serde_json::from_str(cloud.data.as_ref().unwrap()).unwrap();
        assert_eq!(cloud_data["provider"], "xybrid");
        assert_eq!(cloud_data["outcome_category"], "cloud_success");
        assert_eq!(cloud_data["status"], "ok");
        assert!(cloud.error.is_none());
    }

    #[test]
    fn cloud_retry_event_marks_failure_branch() {
        let cloud = cloud_retry_event(
            "run-fail-1",
            "deepseek-chat",
            Some("deepseek"),
            842,
            0,
            Some("Gateway returned 502: Provider error"),
        );
        let cloud_data: serde_json::Value =
            serde_json::from_str(cloud.data.as_ref().unwrap()).unwrap();
        assert_eq!(cloud.event_type, "CloudRetry");
        assert_eq!(cloud.target.as_deref(), Some("cloud"));
        assert_eq!(cloud.latency_ms, Some(842));
        assert_eq!(
            cloud.error.as_deref(),
            Some("Gateway returned 502: Provider error")
        );
        assert_eq!(cloud_data["status"], "error");
        assert_eq!(cloud_data["outcome_category"]["kind"], "hard_fail");
        assert_eq!(
            cloud_data["outcome_category"]["reason"],
            "Gateway returned 502: Provider error"
        );
        assert_eq!(cloud_data["tokens_emitted"], 0);

        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&cloud, &config, None, None, None);
        assert_eq!(
            platform.outcome_category.as_ref(),
            Some(&serde_json::json!({
                "kind": "hard_fail",
                "reason": "Gateway returned 502: Provider error"
            }))
        );
    }

    #[test]
    fn telemetry_error_redaction_removes_api_keys_and_bearer_tokens() {
        let redacted = redact_error_for_telemetry(
            "Gateway returned 401: Authorization: Bearer sk_test_abc123 and api_key=hf_secret_xyz",
        );

        assert!(redacted.contains("Authorization: Bearer [REDACTED]"));
        assert!(redacted.contains("api_key=[REDACTED]"));
        assert!(!redacted.contains("sk_test_abc123"));
        assert!(!redacted.contains("hf_secret_xyz"));
    }

    #[test]
    fn attach_resource_summary_with_none_is_noop() {
        let original = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: None,
            target: None,
            latency_ms: None,
            error: None,
            data: Some("{\"model_id\":\"x\"}".to_string()),
            timestamp_ms: 0,
        };
        let mut event = original.clone();
        attach_resource_summary(&mut event, None);
        assert_eq!(event.data, original.data);
    }

    #[test]
    fn env_resource_mode_parses_all_variants() {
        // Helper drives the parsed variants without touching the real env.
        let cases: &[(&str, ResourceTelemetryMode)] = &[
            ("off", ResourceTelemetryMode::Off),
            ("boundary", ResourceTelemetryMode::Boundary),
            (
                "summary",
                ResourceTelemetryMode::Summary {
                    interval_ms: ResourceTelemetryMode::DEFAULT_SUMMARY_INTERVAL_MS,
                },
            ),
            (
                "summary:500",
                ResourceTelemetryMode::Summary { interval_ms: 500 },
            ),
            (
                "debug_local:250",
                ResourceTelemetryMode::DebugLocal { interval_ms: 250 },
            ),
        ];
        for (raw, expected) in cases {
            std::env::set_var("XYBRID_RESOURCE_TELEMETRY", raw);
            let parsed = resource_mode_from_env().expect(raw);
            assert_eq!(&parsed, expected, "parsing `{}`", raw);
        }
        std::env::remove_var("XYBRID_RESOURCE_TELEMETRY");
        assert!(resource_mode_from_env().is_none());
    }

    #[test]
    fn test_convert_stage_start_event() {
        let event = OrchestratorEvent::StageStart {
            stage_name: "asr".to_string(),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "StageStart");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert!(telemetry.target.is_none());
        assert!(telemetry.latency_ms.is_none());
        assert!(telemetry.error.is_none());
        assert!(telemetry.timestamp_ms > 0);
    }

    #[test]
    fn test_convert_stage_complete_event() {
        let event = OrchestratorEvent::StageComplete {
            stage_name: "tts".to_string(),
            target: "local".to_string(),
            latency_ms: 150,
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "StageComplete");
        assert_eq!(telemetry.stage_name, Some("tts".to_string()));
        assert_eq!(telemetry.target, Some("local".to_string()));
        assert_eq!(telemetry.latency_ms, Some(150));
        assert!(telemetry.error.is_none());
    }

    #[test]
    fn test_convert_stage_error_event() {
        let event = OrchestratorEvent::StageError {
            stage_name: "asr".to_string(),
            error: "Model not found".to_string(),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "StageError");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert_eq!(telemetry.error, Some("Model not found".to_string()));
    }

    #[test]
    fn test_convert_pipeline_start_event() {
        let event = OrchestratorEvent::PipelineStart {
            stages: vec!["asr".to_string(), "llm".to_string(), "tts".to_string()],
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "PipelineStart");
        assert!(telemetry.stage_name.is_none());
        assert!(telemetry.data.is_some());
        let data = telemetry.data.unwrap();
        assert!(data.contains("asr"));
        assert!(data.contains("llm"));
        assert!(data.contains("tts"));
    }

    #[test]
    fn test_convert_pipeline_complete_event() {
        let event = OrchestratorEvent::PipelineComplete {
            total_latency_ms: 500,
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "PipelineComplete");
        assert_eq!(telemetry.latency_ms, Some(500));
    }

    #[test]
    fn test_convert_routing_decided_event() {
        let event = OrchestratorEvent::RoutingDecided {
            stage_name: "asr".to_string(),
            target: "cloud".to_string(),
            reason: "network_optimal".to_string(),
            recent_abort_rate: 0.0,
            sample_size: 0,
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "RoutingDecided");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert_eq!(telemetry.target, Some("cloud".to_string()));
        assert!(telemetry.data.is_some());
        let data = telemetry.data.unwrap();
        assert!(data.contains("network_optimal"));
    }

    #[test]
    fn routing_decided_event_carries_local_reliability_hint_end_to_end() {
        // Production-shape regression test: walk the full
        // OrchestratorEvent -> TelemetryEvent -> PlatformEvent pipeline
        // so the hint flows through every seam the previous PR draft
        // missed.
        let event = OrchestratorEvent::RoutingDecided {
            stage_name: "stage-1".to_string(),
            target: "cloud".to_string(),
            reason: "history_bias".to_string(),
            recent_abort_rate: 0.75,
            sample_size: 4,
            context: Default::default(),
        };
        let telemetry_event = convert_orchestrator_event(&event);

        // Bridge embeds the hint in event.data so the hoist list picks it up.
        let data_str = telemetry_event.data.as_ref().expect("data must be present");
        let parsed_data: serde_json::Value = serde_json::from_str(data_str).unwrap();
        assert_eq!(
            parsed_data["local_reliability_hint"]["recent_abort_rate"]
                .as_f64()
                .unwrap_or(-1.0),
            0.75
        );
        assert_eq!(
            parsed_data["local_reliability_hint"]["sample_size"].as_i64(),
            Some(4)
        );

        // And the platform-event payload surfaces the hint at the top level.
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&telemetry_event, &config, None, None, None);
        assert_eq!(
            platform.payload["local_reliability_hint"]["recent_abort_rate"]
                .as_f64()
                .unwrap_or(-1.0),
            0.75
        );
        assert_eq!(
            platform.payload["local_reliability_hint"]["sample_size"].as_i64(),
            Some(4)
        );
    }

    #[test]
    fn routing_decided_event_sanitizes_non_finite_recent_abort_rate() {
        // Defense-in-depth: the authority emits only finite rates, but
        // the bridge clamps NaN/Infinity to 0.0 so the platform's typed
        // Float32 column never receives a JSON null. The clamp lives in
        // the bridge so any future direct OrchestratorEvent caller (test,
        // example, FFI) cannot poison the telemetry stream.
        for bad_rate in [f32::NAN, f32::INFINITY, f32::NEG_INFINITY] {
            let event = OrchestratorEvent::RoutingDecided {
                stage_name: "stage-1".to_string(),
                target: "cloud".to_string(),
                reason: "history_bias".to_string(),
                recent_abort_rate: bad_rate,
                sample_size: 1,
                context: Default::default(),
            };
            let telemetry_event = convert_orchestrator_event(&event);
            let parsed_data: serde_json::Value =
                serde_json::from_str(telemetry_event.data.as_ref().unwrap()).unwrap();
            assert_eq!(
                parsed_data["local_reliability_hint"]["recent_abort_rate"].as_f64(),
                Some(0.0),
                "non-finite rate {bad_rate} must be sanitized to 0.0"
            );
        }
    }

    #[test]
    fn scoped_orchestrator_bridge_drains_queued_events_with_captured_context() {
        let _guard = TelemetrySenderTestGuard::acquire();
        let (tx, rx) = mpsc::channel();
        register_telemetry_sender(tx);

        let pipeline_id = Uuid::new_v4();
        let trace_id = Uuid::new_v4();
        let _event_context = xybrid_core::event_bus::EventContextGuard::install(
            xybrid_core::event_bus::EventContext::default()
                .with_pipeline_id(pipeline_id)
                .with_trace_id(trace_id),
        );
        let orchestrator = xybrid_core::orchestrator::Orchestrator::new();
        let bridge = bridge_orchestrator_events(&orchestrator);

        orchestrator
            .event_bus()
            .publish(OrchestratorEvent::RoutingDecided {
                stage_name: "scoped-bridge-context".to_string(),
                target: "cloud".to_string(),
                reason: "history_bias".to_string(),
                recent_abort_rate: 0.5,
                sample_size: 2,
                context: Default::default(),
            });
        bridge.drain();

        let mut received = None;
        for _ in 0..20 {
            match rx.recv_timeout(std::time::Duration::from_millis(50)) {
                Ok(event)
                    if event.event_type == "RoutingDecided"
                        && event.stage_name.as_deref() == Some("scoped-bridge-context") =>
                {
                    received = Some(event);
                    break;
                }
                Ok(_) | Err(mpsc::RecvTimeoutError::Timeout) => {}
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    panic!("telemetry receiver disconnected before RoutingDecided arrived")
                }
            }
        }
        let received = received.expect("drained bridge should publish queued orchestrator event");

        let data: serde_json::Value =
            serde_json::from_str(received.data.as_ref().expect("context-bearing data")).unwrap();
        assert_eq!(
            data[CONTEXT_PIPELINE_ID_KEY],
            serde_json::json!(pipeline_id)
        );
        assert_eq!(data[CONTEXT_TRACE_ID_KEY], serde_json::json!(trace_id));
        assert_eq!(
            data["local_reliability_hint"]["recent_abort_rate"].as_f64(),
            Some(0.5)
        );
        assert_eq!(
            data["local_reliability_hint"]["sample_size"].as_i64(),
            Some(2)
        );
    }

    #[test]
    fn test_convert_execution_started_event() {
        let event = OrchestratorEvent::ExecutionStarted {
            stage_name: "asr".to_string(),
            target: "local".to_string(),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "ExecutionStarted");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert_eq!(telemetry.target, Some("local".to_string()));
    }

    #[test]
    fn test_convert_execution_completed_event() {
        let event = OrchestratorEvent::ExecutionCompleted {
            stage_name: "asr".to_string(),
            target: "local".to_string(),
            execution_time_ms: 75,
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "ExecutionCompleted");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert_eq!(telemetry.target, Some("local".to_string()));
        assert_eq!(telemetry.latency_ms, Some(75));
    }

    #[test]
    fn test_convert_execution_failed_event() {
        let event = OrchestratorEvent::ExecutionFailed {
            stage_name: "tts".to_string(),
            target: "cloud".to_string(),
            error: "Timeout".to_string(),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "ExecutionFailed");
        assert_eq!(telemetry.stage_name, Some("tts".to_string()));
        assert_eq!(telemetry.target, Some("cloud".to_string()));
        assert_eq!(telemetry.error, Some("Timeout".to_string()));
    }

    #[test]
    fn test_convert_policy_evaluated_allowed() {
        let event = OrchestratorEvent::PolicyEvaluated {
            stage_name: "asr".to_string(),
            allowed: true,
            reason: Some("All conditions met".to_string()),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "PolicyEvaluated");
        assert_eq!(telemetry.stage_name, Some("asr".to_string()));
        assert!(telemetry.error.is_none()); // No error when allowed
        assert!(telemetry.data.is_some());
    }

    #[test]
    fn test_convert_policy_evaluated_denied() {
        let event = OrchestratorEvent::PolicyEvaluated {
            stage_name: "llm".to_string(),
            allowed: false,
            reason: Some("Privacy policy violation".to_string()),
            context: Default::default(),
        };
        let telemetry = convert_orchestrator_event(&event);

        assert_eq!(telemetry.event_type, "PolicyEvaluated");
        assert_eq!(telemetry.stage_name, Some("llm".to_string()));
        assert_eq!(
            telemetry.error,
            Some("Privacy policy violation".to_string())
        );
    }

    fn routing_decided_event(stage_name: impl Into<String>) -> OrchestratorEvent {
        OrchestratorEvent::RoutingDecided {
            stage_name: stage_name.into(),
            target: "local".to_string(),
            reason: "test_route".to_string(),
            recent_abort_rate: 0.0,
            sample_size: 0,
            context: xybrid_core::event_bus::EventContext::default(),
        }
    }

    fn clear_registered_telemetry_senders() {
        TELEMETRY_SENDERS
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clear();
    }

    struct TelemetrySenderTestGuard {
        _guard: MutexGuard<'static, ()>,
    }

    impl TelemetrySenderTestGuard {
        fn acquire() -> Self {
            let guard = TELEMETRY_SENDER_TEST_LOCK
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            clear_registered_telemetry_senders();
            Self { _guard: guard }
        }
    }

    impl Drop for TelemetrySenderTestGuard {
        fn drop(&mut self) {
            clear_registered_telemetry_senders();
        }
    }

    #[test]
    fn bridge_drains_all_events_on_orchestrator_drop() {
        let _guard = TelemetrySenderTestGuard::acquire();
        let (tx, rx) = mpsc::channel();
        register_telemetry_sender(tx);

        let orchestrator = xybrid_core::orchestrator::Orchestrator::new();
        let bridge = bridge_orchestrator_events(&orchestrator);
        for idx in 0..100 {
            orchestrator
                .event_bus()
                .publish(routing_decided_event(format!("bridge-drain-{idx:03}")));
        }

        drop(orchestrator);
        bridge.join().expect("bridge should drain cleanly");

        let events: Vec<_> = rx
            .try_iter()
            .filter(|event| {
                event.event_type == "RoutingDecided"
                    && event
                        .stage_name
                        .as_deref()
                        .is_some_and(|stage| stage.starts_with("bridge-drain-"))
            })
            .collect();
        assert_eq!(events.len(), 100);
        for idx in 0..100 {
            assert!(
                events
                    .iter()
                    .any(|event| event.stage_name.as_deref()
                        == Some(&format!("bridge-drain-{idx:03}"))),
                "missing routed event bridge-drain-{idx:03}; got {events:?}"
            );
        }
    }

    #[test]
    fn bridge_preserves_correlation_id_across_task_boundary() {
        let _guard = TelemetrySenderTestGuard::acquire();
        let (tx, rx) = mpsc::channel();
        register_telemetry_sender(tx);

        let publishers: Vec<_> = ["A", "B"]
            .into_iter()
            .map(|correlation_id| {
                thread::spawn(move || {
                    let _context = xybrid_core::event_bus::EventContextGuard::install(
                        xybrid_core::event_bus::EventContext::default()
                            .with_correlation_id(correlation_id),
                    );
                    let orchestrator = xybrid_core::orchestrator::Orchestrator::new();
                    let bridge = bridge_orchestrator_events(&orchestrator);
                    for idx in 0..10 {
                        orchestrator
                            .event_bus()
                            .publish(routing_decided_event(format!("{correlation_id}-{idx}")));
                        thread::yield_now();
                    }
                    drop(orchestrator);
                    bridge.join().expect("bridge should drain cleanly");
                })
            })
            .collect();

        for publisher in publishers {
            publisher.join().expect("publisher thread should not panic");
        }

        let events: Vec<_> = rx
            .try_iter()
            .filter(|event| {
                event.event_type == "RoutingDecided"
                    && event
                        .stage_name
                        .as_deref()
                        .is_some_and(|stage| stage.starts_with("A-") || stage.starts_with("B-"))
            })
            .collect();
        assert_eq!(events.len(), 20);
        for event in events {
            let stage = event.stage_name.as_deref().expect("stage name");
            let data: serde_json::Value =
                serde_json::from_str(event.data.as_deref().expect("event data")).unwrap();
            let expected = stage.split('-').next().unwrap();
            assert_eq!(
                data["correlation_id"].as_str(),
                Some(expected),
                "event {stage} carried wrong context: {data}"
            );
        }
    }

    #[test]
    fn flush_blocks_until_in_flight_events_delivered() {
        let _guard = TelemetrySenderTestGuard::acquire();
        let (tx, rx) = mpsc::channel();
        register_telemetry_sender(tx);

        let orchestrator = xybrid_core::orchestrator::Orchestrator::new();
        let bridge = bridge_orchestrator_events(&orchestrator);
        for idx in 0..5 {
            orchestrator
                .event_bus()
                .publish(routing_decided_event(format!("flush-{idx}")));
        }

        bridge.flush().expect("flush should complete");

        let mut delivered = Vec::new();
        let deadline = std::time::Instant::now() + Duration::from_secs(1);
        while delivered.len() < 5 && std::time::Instant::now() < deadline {
            let event = rx
                .recv_timeout(Duration::from_millis(50))
                .expect("flush should deliver event before returning");
            if event.event_type == "RoutingDecided"
                && event
                    .stage_name
                    .as_deref()
                    .is_some_and(|stage| stage.starts_with("flush-"))
            {
                delivered.push(event);
            }
        }
        assert_eq!(delivered.len(), 5);

        drop(orchestrator);
        bridge.join().expect("bridge should drain cleanly");
    }

    #[test]
    fn test_telemetry_event_serialization() {
        let event = TelemetryEvent {
            event_type: "StageStart".to_string(),
            stage_name: Some("asr".to_string()),
            target: None,
            latency_ms: None,
            error: None,
            data: None,
            timestamp_ms: 1234567890,
        };

        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("StageStart"));
        assert!(json.contains("asr"));

        let deserialized: TelemetryEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(deserialized.event_type, "StageStart");
        assert_eq!(deserialized.stage_name, Some("asr".to_string()));
    }

    #[test]
    fn test_register_and_publish() {
        let _guard = TelemetrySenderTestGuard::acquire();
        let (tx, rx) = mpsc::channel();
        register_telemetry_sender(tx);

        let event = TelemetryEvent {
            event_type: "TestEvent".to_string(),
            stage_name: Some("test".to_string()),
            target: None,
            latency_ms: None,
            error: None,
            data: None,
            timestamp_ms: 0,
        };

        publish_telemetry_event(event.clone());

        // Should receive the event
        let received = rx.recv_timeout(std::time::Duration::from_millis(100));
        assert!(received.is_ok());
        let received_event = received.unwrap();
        assert_eq!(received_event.event_type, "TestEvent");
    }

    #[test]
    fn test_telemetry_config_defaults() {
        let config = TelemetryConfig::default();
        assert_eq!(config.batch_size, 10);
        assert_eq!(config.flush_interval_secs, 5);
        assert_eq!(config.max_retries, 3);
        assert!(config.enable_retry_queue);
        assert!(config.auto_hardware_detection);
        assert!(!config.capture_hostname);
    }

    #[test]
    fn test_http_exporter_circuit_breaker_initial_state() {
        let config = TelemetryConfig::new("https://example.com", "test-key")
            .with_device("test-device", "test-platform");
        let exporter = HttpTelemetryExporter::new(config);
        assert!(!exporter.is_circuit_open());
    }

    #[test]
    fn test_http_exporter_circuit_breaker_reset() {
        let config = TelemetryConfig::new("https://example.com", "test-key")
            .with_device("test-device", "test-platform");
        let exporter = HttpTelemetryExporter::new(config);

        // Manually trigger failures to open the circuit
        for _ in 0..3 {
            exporter.circuit.record_failure();
        }
        assert!(exporter.is_circuit_open());

        // Reset should close it
        exporter.reset_circuit();
        assert!(!exporter.is_circuit_open());
    }

    #[test]
    fn test_http_exporter_failed_queue_initial_empty() {
        let config = TelemetryConfig::new("https://example.com", "test-key")
            .with_device("test-device", "test-platform");
        let exporter = HttpTelemetryExporter::new(config);
        assert_eq!(exporter.failed_queue_size(), 0);
        assert_eq!(exporter.dropped_count(), 0);
    }

    #[test]
    fn test_queue_failed_events() {
        let queue = Arc::new(Mutex::new(VecDeque::new()));
        let dropped = Arc::new(AtomicU32::new(0));

        let events = vec![PlatformEvent {
            session_id: Uuid::new_v4(),
            event_type: "Test".to_string(),
            payload: serde_json::json!({}),
            device_id: None,
            device_label: None,
            platform: None,
            app_version: None,
            device: None,
            timestamp: None,
            pipeline_id: None,
            trace_id: None,
            correlation_id: None,
            outcome_category: None,
            abort_reason: None,
            stages: None,
        }];

        queue_failed_events(events, &queue, &dropped);
        assert_eq!(queue.lock().unwrap().len(), 1);
        assert_eq!(dropped.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn test_is_retryable_status() {
        assert!(is_retryable_status(429));
        assert!(is_retryable_status(502));
        assert!(is_retryable_status(503));
        assert!(is_retryable_status(504));
        assert!(!is_retryable_status(200));
        assert!(!is_retryable_status(400));
        assert!(!is_retryable_status(401));
        assert!(!is_retryable_status(404));
    }

    // ------------------------------------------------------------------
    // Device telemetry — opt-out contract + random-UUID device_id
    //
    // These tests cover the adversarial-review findings: auto-detection
    // opt-out must omit both the `device` object and the auto-generated
    // `device_id`; the default device_id must not be a hardware fingerprint.
    // ------------------------------------------------------------------

    fn sample_event() -> TelemetryEvent {
        TelemetryEvent {
            event_type: "TestEvent".to_string(),
            stage_name: None,
            target: None,
            latency_ms: None,
            error: None,
            data: None,
            timestamp_ms: 0,
        }
    }

    #[test]
    fn opt_out_emits_no_device_and_no_auto_id() {
        // Start with a default config (which auto-wires `device_id` from
        // `Device::current()`), then opt out of hardware detection without
        // providing a label or overrides. This mirrors the `HttpTelemetryExporter::new`
        // merge path where the strict opt-out contract clears the
        // default-injected id.
        let mut config =
            TelemetryConfig::new("http://example.invalid", "k").with_auto_hardware_detection(false);
        let profile = resolve_device_profile(&config);
        assert!(profile.is_none(), "strict opt-out must yield None profile");

        // Replicate the exporter's strict-opt-out clearing so the test covers
        // the full emission path, not just `convert_to_platform_event`.
        if !config.auto_hardware_detection && !config.device_id_explicit {
            config.device_id = None;
        }

        let event =
            convert_to_platform_event(&sample_event(), &config, profile.as_ref(), None, None);
        let json = serde_json::to_value(&event).unwrap();
        assert!(
            json.get("device").is_none(),
            "no `device` key on strict opt-out, got: {json}"
        );
        assert!(
            json.get("device_id").is_none(),
            "strict opt-out must omit `device_id` entirely, not emit null. got: {json}"
        );
    }

    #[test]
    fn embedded_context_overrides_flush_context() {
        let config = TelemetryConfig::new("http://example.invalid", "k");
        let event_pipeline_id = Uuid::new_v4();
        let event_trace_id = Uuid::new_v4();
        let flush_pipeline_id = Uuid::new_v4();
        let flush_trace_id = Uuid::new_v4();
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: None,
            target: Some("cloud".to_string()),
            latency_ms: Some(42),
            error: None,
            data: Some(
                serde_json::json!({
                    CONTEXT_PIPELINE_ID_KEY: event_pipeline_id,
                    CONTEXT_TRACE_ID_KEY: event_trace_id,
                    "model_id": "deepseek-chat",
                    "tokens_in": 100,
                    "tokens_out": 10,
                })
                .to_string(),
            ),
            timestamp_ms: 0,
        };

        let converted = convert_to_platform_event(
            &event,
            &config,
            None,
            Some(flush_pipeline_id),
            Some(flush_trace_id),
        );

        assert_eq!(converted.pipeline_id, Some(event_pipeline_id));
        assert_eq!(converted.trace_id, Some(event_trace_id));
        assert_eq!(converted.payload["model_id"], "deepseek-chat");
        assert!(converted.payload["data"]
            .get(CONTEXT_PIPELINE_ID_KEY)
            .is_none());
        assert!(converted.payload["data"]
            .get(CONTEXT_TRACE_ID_KEY)
            .is_none());
    }

    #[test]
    fn extract_llm_token_counts_reads_canonical_keys() {
        let stages = serde_json::json!({
            "spans": [
                { "name": "execute:preprocessing", "metadata": {} },
                {
                    "name": "llm_inference_with_messages",
                    "metadata": {
                        "tokens_in": "128",
                        "tokens_out": "42",
                        "tokens_generated": "42"
                    }
                }
            ]
        });
        let (tin, tout) = extract_llm_token_counts(&stages).expect("should find llm span");
        assert_eq!(tin, Some(128));
        assert_eq!(tout, Some(42));
    }

    #[test]
    fn extract_llm_token_counts_falls_back_to_openai_style_keys() {
        let stages = serde_json::json!({
            "spans": [{
                "name": "llm_inference_streaming",
                "metadata": {
                    "prompt_tokens": "16",
                    "completion_tokens": "64"
                }
            }]
        });
        let (tin, tout) = extract_llm_token_counts(&stages).expect("should find llm span");
        assert_eq!(tin, Some(16));
        assert_eq!(tout, Some(64));
    }

    #[test]
    fn extract_llm_token_counts_returns_none_without_llm_span() {
        let stages = serde_json::json!({
            "spans": [
                { "name": "execute:asr.whisper-tiny", "metadata": {} }
            ]
        });
        assert!(extract_llm_token_counts(&stages).is_none());
    }

    #[test]
    fn extract_llm_prompt_cached_tokens_picks_last_llm_span_with_value() {
        // Mirrors the token-count extractor's last-span-wins rule: a
        // streaming run emits a timing-only span first, then a final
        // span with the accounting numbers. The cached-tokens hoist
        // must read from the final span — earlier values should be
        // shadowed when a later span has its own.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "ttft_ms": 120, "prompt_cached_tokens": "0" }
                },
                {
                    "name": "llm_inference_with_messages",
                    "metadata": { "tokens_in": 200, "prompt_cached_tokens": "150" }
                }
            ]
        });
        assert_eq!(extract_llm_prompt_cached_tokens(&stages), Some(150));
    }

    #[test]
    fn extract_llm_prompt_cached_tokens_handles_numeric_value() {
        // The runtime emits via add_metadata which always stringifies,
        // but a future emit path or external producer might supply a
        // raw number. Both shapes must extract correctly.
        let stages = serde_json::json!({
            "spans": [{
                "name": "llm_inference_with_messages",
                "metadata": { "prompt_cached_tokens": 96 }
            }]
        });
        assert_eq!(extract_llm_prompt_cached_tokens(&stages), Some(96));
    }

    #[test]
    fn extract_llm_prompt_cached_tokens_returns_none_when_absent() {
        // Backends that don't track prefix reuse never emit the key —
        // the SDK hoist must produce None so the wire payload omits the
        // field entirely (rather than emitting a misleading 0).
        let stages = serde_json::json!({
            "spans": [{
                "name": "llm_inference_with_messages",
                "metadata": { "tokens_in": 32, "tokens_out": 96 }
            }]
        });
        assert_eq!(extract_llm_prompt_cached_tokens(&stages), None);
    }

    #[test]
    fn extract_llm_prompt_cached_tokens_ignores_non_llm_spans() {
        // An ASR/TTS span carrying a stray `prompt_cached_tokens` would
        // be a real bug to flag, but the extractor must not let that
        // value leak onto the LLM hoist axis. Same LLM-span detection
        // as the token-count extractor.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "execute:asr.whisper-tiny",
                    "metadata": { "prompt_cached_tokens": "999" }
                }
            ]
        });
        assert_eq!(extract_llm_prompt_cached_tokens(&stages), None);
    }

    #[test]
    fn extract_llm_token_counts_scans_across_llm_spans() {
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "ttft_ms": 120 }
                },
                {
                    "name": "inference:qwen2.5-0.5b",
                    "metadata": { "tokens_in": 32, "tokens_out": 96 }
                }
            ]
        });
        let (tin, tout) = extract_llm_token_counts(&stages).expect("should find llm spans");
        assert_eq!(tin, Some(32));
        assert_eq!(tout, Some(96));
    }

    #[test]
    fn extract_llm_token_counts_prefers_last_authoritative_span() {
        // Retry-like trace: three spans, each with progressively larger
        // token counts. The authoritative totals live on the final attempt.
        // We must NOT keep the first-seen values from span #1.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "tokens_in": 10, "tokens_out": 5 }
                },
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "ttft_ms": 50 }
                },
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "tokens_in": 128, "tokens_out": 42 }
                }
            ]
        });
        let (tin, tout) = extract_llm_token_counts(&stages).expect("should find llm spans");
        assert_eq!(tin, Some(128), "must take last-span tokens_in, not first");
        assert_eq!(tout, Some(42), "must take last-span tokens_out, not first");
    }

    #[test]
    fn extract_llm_token_counts_falls_back_across_partial_spans() {
        // Streaming-with-partial: the final span has no tokens_in but an
        // earlier span did. The fallback must carry the earlier value
        // through rather than drop it, since nothing later overrides it.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "tokens_in": 64 }
                },
                {
                    "name": "llm_inference_streaming",
                    "metadata": { "tokens_out": 256 }
                }
            ]
        });
        let (tin, tout) = extract_llm_token_counts(&stages).expect("should find llm spans");
        assert_eq!(tin, Some(64));
        assert_eq!(tout, Some(256));
    }

    #[test]
    fn extract_llm_inference_string_attr_reads_backend_and_provider() {
        // The cloud adapter writes both keys onto the inner
        // `llm_inference` span (see `runtime_adapter::cloud::mod`).
        // The hoist must read them through the same span-detection
        // rule used by the token-count hoist so the two stay in sync.
        let stages = serde_json::json!({
            "spans": [
                { "name": "execute:gpt-4o-mini", "metadata": {} },
                {
                    "name": "llm_inference",
                    "metadata": {
                        "backend": "cloud",
                        "provider": "openai",
                        "tokens_in": "120",
                        "tokens_out": "32"
                    }
                }
            ]
        });
        assert_eq!(
            extract_llm_inference_string_attr(&stages, "backend").as_deref(),
            Some("cloud")
        );
        assert_eq!(
            extract_llm_inference_string_attr(&stages, "provider").as_deref(),
            Some("openai")
        );
    }

    #[test]
    fn extract_llm_inference_string_attr_skips_non_llm_spans() {
        // A non-LLM span (e.g., the outer execute span or an ASR
        // span) carrying a `backend` key MUST NOT win — the hoist's
        // contract is "read from the LLM span, the same one the
        // token-count hoist reads". Otherwise a TTS/ASR backend
        // string could spuriously land on an LLM event's payload.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "execute:asr-whisper-tiny",
                    "metadata": { "backend": "ort" }
                },
                {
                    "name": "execute:gpt-4o-mini",
                    "metadata": { "provider": "should-not-win" }
                }
            ]
        });
        assert!(extract_llm_inference_string_attr(&stages, "backend").is_none());
        assert!(extract_llm_inference_string_attr(&stages, "provider").is_none());
    }

    #[test]
    fn extract_llm_inference_string_attr_prefers_last_llm_span() {
        // Mirror of `extract_llm_token_counts_prefers_last_authoritative_span`:
        // a retried run emits one LLM span per attempt and the final
        // attempt is the source of truth. Fields from earlier attempts
        // must not leak through.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "llm_inference",
                    "metadata": { "backend": "cloud", "provider": "anthropic" }
                },
                {
                    "name": "llm_inference",
                    "metadata": { "backend": "cloud", "provider": "openai" }
                }
            ]
        });
        assert_eq!(
            extract_llm_inference_string_attr(&stages, "provider").as_deref(),
            Some("openai"),
            "must take last-span provider, not first"
        );
    }

    #[test]
    fn extract_string_attr_from_any_span_reads_outer_execute_span() {
        // Non-LLM events (ASR/TTS) carry `backend` on the outer
        // `execute:<model>` span via `backend_label_from_template` —
        // the LLM-gated hoist would skip it, dropping backend
        // attribution from the wire payload. The any-span hoist must
        // pick it up so ASR/TTS rows aren't blank in the billing column.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "execute:wav2vec2-base-960h",
                    "metadata": { "backend": "ort" }
                }
            ]
        });
        assert_eq!(
            extract_string_attr_from_any_span(&stages, "backend").as_deref(),
            Some("ort")
        );
    }

    #[test]
    fn extract_string_attr_from_any_span_prefers_last() {
        // Same last-wins rule as the LLM-gated hoist, applied across
        // any span. If both inner and outer spans carry `backend` (an
        // LLM run that also annotates the outer `execute:` span), the
        // last one in the array wins — matching the existing
        // token-count hoist behaviour for retried runs.
        let stages = serde_json::json!({
            "spans": [
                {
                    "name": "execute:gpt-4o-mini",
                    "metadata": { "backend": "first" }
                },
                {
                    "name": "llm_inference",
                    "metadata": { "backend": "second" }
                }
            ]
        });
        assert_eq!(
            extract_string_attr_from_any_span(&stages, "backend").as_deref(),
            Some("second")
        );
    }

    #[test]
    fn build_model_download_event_has_canonical_shape() {
        // ModelDownload is the cost-attribution event for a successful
        // registry fetch. The wire shape is fixed by the platform's
        // billing pipeline: top-level `event_type` literal
        // "ModelDownload", `latency_ms` mirrors `duration_ms`, and the
        // four cost fields land under `data` as a JSON object.
        let event = build_model_download_event("kokoro-82m", 1_234_567, "huggingface", 5_432);

        assert_eq!(event.event_type, "ModelDownload");
        assert_eq!(
            event.latency_ms,
            Some(5_432),
            "latency_ms must mirror duration_ms so the existing latency column lights up without a schema migration"
        );
        assert!(event.error.is_none());
        assert!(event.stage_name.is_none());
        assert!(event.target.is_none());
        assert!(event.timestamp_ms > 0, "timestamp must be populated");

        let data: serde_json::Value =
            serde_json::from_str(event.data.as_deref().expect("data present"))
                .expect("data is valid JSON");
        assert_eq!(data["model_id"].as_str(), Some("kokoro-82m"));
        assert_eq!(data["bytes_downloaded"].as_u64(), Some(1_234_567));
        assert_eq!(data["source"].as_str(), Some("huggingface"));
        assert_eq!(data["duration_ms"].as_u64(), Some(5_432));
    }

    #[test]
    fn build_model_download_event_carries_r2_source_label() {
        // Sanity that the closed-set source label is passed through as-is.
        // The classifier in `registry_client::classify_download_source`
        // is responsible for mapping URLs to labels; this test guards
        // the builder's pass-through contract.
        let event = build_model_download_event("test-model", 42, "r2", 1);
        let data: serde_json::Value = serde_json::from_str(event.data.as_deref().unwrap()).unwrap();
        assert_eq!(data["source"].as_str(), Some("r2"));
    }

    #[test]
    fn task_field_hoists_to_payload_top_level() {
        // INF-93: `task` is a semantic label sourced from the model
        // bundle's `metadata.task`, written by `TemplateExecutor` onto
        // the outer `execute:<model>` span. This test exercises the
        // full convert path so an accidental future regression (e.g.,
        // dropping the hoist branch) is caught without depending on
        // the executor.
        //
        // `convert_to_platform_event` only runs the hoist block when
        // global tracing is on (it would otherwise have nothing to
        // capture from the global collector); flip it on here so the
        // embedded-spans branch fires.
        xybrid_core::tracing::init_tracing(true);
        let data = serde_json::json!({
            "spans": [
                {
                    "name": "execute:wav2vec2-base-960h",
                    "metadata": { "task": "asr", "backend": "ort" }
                }
            ]
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("transcribe".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(420),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let payload_json = serde_json::to_string(&platform.payload).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&payload_json).unwrap();
        assert_eq!(
            parsed["task"].as_str(),
            Some("asr"),
            "task must be hoisted to payload top level: {}",
            payload_json
        );
    }

    #[test]
    fn quantization_field_hoists_to_payload_top_level() {
        // INF-91: `quantization` is sourced by `TemplateExecutor` from
        // `metadata.quantization` (or GGUF filename inference) and
        // written onto the outer `execute:<model>` span. The hoist
        // must surface it on the wire payload alongside `task` and
        // `backend` so analytics rolls up
        // `model_id × quantization` as distinct rows.
        xybrid_core::tracing::init_tracing(true);
        let data = serde_json::json!({
            "spans": [
                {
                    "name": "execute:qwen2.5-0.5b-instruct",
                    "metadata": { "quantization": "q4_k_m", "backend": "llamacpp" }
                }
            ]
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("chat".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(1_200),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let parsed: serde_json::Value =
            serde_json::from_value(serde_json::to_value(&platform.payload).unwrap()).unwrap();
        assert_eq!(
            parsed["quantization"].as_str(),
            Some("q4_k_m"),
            "quantization must be hoisted: {}",
            serde_json::to_string(&parsed).unwrap()
        );
    }

    #[test]
    fn streaming_field_hoists_to_payload_top_level() {
        // `XybridModel::run_streaming` and the streaming-fast-path
        // `ModelComplete` (Pipeline) both stamp `data.streaming = true`.
        // The hoist must surface it on the wire payload's top level so
        // the platform's Tinybird datasource can read it as a typed
        // column for a `streaming` badge / filter on the Traces
        // dashboard.
        //
        // Unlike `task` / `quantization` (which are sourced from spans),
        // `streaming` is published directly on the event's `data` blob
        // — no span involvement. Test exercises the convert path with a
        // minimal `data` payload mirroring the production publish shape.
        let data = serde_json::json!({
            "model_id": "qwen2.5-0.5b-instruct",
            "version": "1.0",
            "output_type": "Text",
            "streaming": true,
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("qwen2.5-0.5b-instruct".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(1_200),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let parsed: serde_json::Value =
            serde_json::from_value(serde_json::to_value(&platform.payload).unwrap()).unwrap();
        assert_eq!(
            parsed["streaming"].as_bool(),
            Some(true),
            "streaming must be hoisted to payload top level: {}",
            serde_json::to_string(&parsed).unwrap()
        );
    }

    #[test]
    fn streaming_field_omitted_when_data_does_not_carry_it() {
        // Non-streaming inference events (`XybridModel::run`,
        // non-streaming pipeline runs) don't stamp `data.streaming`.
        // The hoist must leave the top-level key absent in that case so
        // the platform-side column reads `NULL` (not `false`) for batch
        // calls — preserves the three-valued logic the Tinybird
        // datasource encodes via `Nullable(UInt8)`.
        let data = serde_json::json!({
            "model_id": "qwen2.5-0.5b-instruct",
            "version": "1.0",
            "output_type": "Text",
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("qwen2.5-0.5b-instruct".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(1_200),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let parsed: serde_json::Value =
            serde_json::from_value(serde_json::to_value(&platform.payload).unwrap()).unwrap();
        assert!(
            parsed.get("streaming").is_none(),
            "streaming must be omitted from top-level payload when absent in data: {}",
            serde_json::to_string(&parsed).unwrap()
        );
    }

    #[test]
    fn task_field_omitted_when_span_does_not_carry_it() {
        // Contract: `task` is optional / additive. A bundle without a
        // `task` declaration must produce a payload that omits the key
        // entirely (not an empty string). Guards against a future
        // refactor that emits `Some("")` or `Some("unknown")`.
        //
        // Tracing must be enabled for the hoist branch to fire;
        // otherwise this test would pass trivially even if the hoist
        // were emitting a stale empty-string default.
        xybrid_core::tracing::init_tracing(true);
        let data = serde_json::json!({
            "spans": [
                {
                    "name": "execute:custom-model",
                    "metadata": { "backend": "ort" }
                }
            ]
        });
        let event = TelemetryEvent {
            event_type: "ModelComplete".to_string(),
            stage_name: Some("custom".to_string()),
            target: Some("local".to_string()),
            latency_ms: Some(10),
            error: None,
            data: Some(data.to_string()),
            timestamp_ms: 1_700_000_000_000,
        };
        let config = TelemetryConfig::new("https://ingest.example.test", "sk_test_abc");
        let platform = convert_to_platform_event(&event, &config, None, None, None);
        let parsed: serde_json::Value =
            serde_json::from_value(serde_json::to_value(&platform.payload).unwrap()).unwrap();
        assert!(
            parsed.get("task").is_none(),
            "task must be absent (not empty string) when bundle didn't declare it: {}",
            serde_json::to_string(&parsed).unwrap()
        );
    }

    #[test]
    fn opt_out_with_explicit_attribute_suppresses_device_id() {
        // Privacy contract: opting out of hardware detection must suppress
        // the auto-wired `device_id` even when explicit non-hardware context
        // (labels, attributes, hostname) puts the profile back in "has
        // context" mode. Callers who want a stable id must opt back in via
        // `with_device(...)`.
        let config = TelemetryConfig::new("http://example.invalid", "k")
            .with_auto_hardware_detection(false)
            .with_device_attribute("tailnet", "production");
        let exporter = HttpTelemetryExporter::new(config);
        assert!(
            exporter.config.device_id.is_none(),
            "opt-out + explicit attribute must suppress auto-wired device_id, got: {:?}",
            exporter.config.device_id
        );
    }

    #[test]
    fn opt_out_with_explicit_with_device_preserves_id() {
        // Inverse of the suppression test: when the caller explicitly opts
        // back in via `with_device(...)`, the identifier must survive the
        // opt-out clear.
        let config = TelemetryConfig::new("http://example.invalid", "k")
            .with_auto_hardware_detection(false)
            .with_device("caller-supplied-id", "linux");
        let exporter = HttpTelemetryExporter::new(config);
        assert_eq!(
            exporter.config.device_id.as_deref(),
            Some("caller-supplied-id"),
            "explicit with_device must survive opt-out clear"
        );
    }

    #[test]
    fn with_hardware_disables_auto_detection() {
        let profile = DeviceProfile {
            chip_family: Some("supplied-chip".into()),
            ram_gb: Some(16),
            ..Default::default()
        };
        let config = TelemetryConfig::new("http://example.invalid", "k").with_hardware(profile);
        assert!(
            !config.auto_hardware_detection,
            "with_hardware must disable auto-detection"
        );
        let resolved = resolve_device_profile(&config).expect("profile present");
        assert_eq!(resolved.chip_family.as_deref(), Some("supplied-chip"));
        assert_eq!(resolved.ram_gb, Some(16));
        assert!(resolved.os.is_none(), "os must stay None (opt-out honored)");
        assert!(
            resolved.arch.is_none(),
            "arch must stay None (opt-out honored)"
        );
    }

    #[test]
    fn opt_out_with_explicit_attribute_still_emits_device() {
        let config = TelemetryConfig::new("http://example.invalid", "k")
            .with_auto_hardware_detection(false)
            .with_device_attribute("tailnet", "production");
        let profile = resolve_device_profile(&config);
        let profile = profile.expect("explicit attribute must surface a profile");
        assert!(profile.chip_family.is_none());
        assert!(profile.ram_gb.is_none());
        assert_eq!(
            profile.custom.get("tailnet").map(String::as_str),
            Some("production")
        );
    }
}