tenuo 0.1.0-beta.16

//! Heartbeat module for control plane integration.
//!
//! This module provides automatic registration, heartbeat, SRL synchronization,
//! and audit event streaming for authorizers connecting to a control plane.
//!
//! # Usage
//!
//! The heartbeat is enabled when all three environment variables are set:
//! - `TENUO_CONTROL_PLANE_URL`: The control plane API endpoint
//! - `TENUO_API_KEY`: API key for authentication
//! - `TENUO_AUTHORIZER_NAME`: Unique name for this authorizer instance
//!
//! # Behavior
//!
//! 1. On startup, registers with the control plane to get an `authorizer_id`
//! 2. Spawns a background task that sends heartbeats at the configured interval
//! 3. If registration fails after 3 retries, continues in standalone mode
//! 4. If heartbeats fail, logs warnings and continues retrying
//! 5. On each heartbeat, checks if SRL update is needed (version or urgent flag)
//! 6. Fetches and applies new SRL when needed
//! 7. Flushes buffered audit events to the control plane (signed if key configured)
//!
//! # Signed Events (Receipts)
//!
//! When `signing_key` is configured, events become cryptographic receipts:
//! - Each event is signed by the authorizer using Ed25519
//! - The control plane verifies the signature and stores as a receipt
//! - This provides non-repudiation: the authorizer (witness) signed the event

use crate::crypto::SigningKey;
use crate::planes::Authorizer;
use crate::revocation::SignedRevocationList;
use crate::PublicKey;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio::time::interval;
use tracing::{debug, info, warn};

// ============================================================================
// Audit Event Streaming
// ============================================================================

/// A verified approval record included in authorization events.
#[derive(Clone, Debug, Serialize)]
pub struct ApprovalRecord {
    /// Hex-encoded Ed25519 public key of the approver
    pub approver_key: String,
    /// External identity reference (e.g., "arn:aws:iam::123:user/admin")
    pub external_id: String,
    /// When the approval was granted (Unix seconds)
    pub approved_at: u64,
    /// When the approval expires (Unix seconds)
    pub expires_at: u64,
    /// Hex-encoded hash binding the approval to (warrant_id, tool, args, holder)
    pub request_hash: String,
}

/// An authorization event to be sent to the control plane for dashboard/analytics.
#[derive(Clone, Debug, Serialize)]
pub struct AuthorizationEvent {
    /// ISO 8601 timestamp
    pub timestamp: String,
    /// Authorizer instance ID (assigned by control plane on registration)
    pub authorizer_id: String,
    /// Warrant ID being authorized (leaf warrant)
    pub warrant_id: String,
    /// Decision: "allow" or "deny"
    pub decision: &'static str,
    /// Reason for denial (if denied)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub deny_reason: Option<String>,
    /// Tool being authorized
    pub tool: String,
    /// Specific constraint that failed (if denied)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub failed_constraint: Option<String>,
    /// Delegation chain depth
    pub chain_depth: u8,
    /// Root principal (issuer of root warrant in chain)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub root_principal: Option<String>,
    /// Full warrant chain (base64-encoded CBOR WarrantStack)
    /// Contains all warrants from root to leaf for chain reconstruction
    #[serde(skip_serializing_if = "Option::is_none")]
    pub warrant_stack: Option<String>,
    /// Authorization latency in microseconds
    pub latency_us: u64,
    /// Unique request ID for tracing
    pub request_id: String,
    /// JSON string of the tool arguments passed by the agent
    #[serde(skip_serializing_if = "Option::is_none")]
    pub arguments: Option<String>,
    /// Verified approval records (present when human-in-the-loop approvals were required)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub approvals: Option<Vec<ApprovalRecord>>,
}

impl AuthorizationEvent {
    /// Create a new "allow" event
    #[allow(clippy::too_many_arguments)]
    pub fn allow(
        authorizer_id: String,
        warrant_id: String,
        tool: String,
        chain_depth: u8,
        root_principal: Option<String>,
        warrant_stack: Option<String>,
        latency_us: u64,
        request_id: String,
        arguments: Option<String>,
        approvals: Option<Vec<ApprovalRecord>>,
    ) -> Self {
        Self {
            timestamp: chrono::Utc::now().to_rfc3339(),
            authorizer_id,
            warrant_id,
            decision: "allow",
            deny_reason: None,
            tool,
            failed_constraint: None,
            chain_depth,
            root_principal,
            warrant_stack,
            latency_us,
            request_id,
            arguments,
            approvals,
        }
    }

    /// Create a new "deny" event
    #[allow(clippy::too_many_arguments)]
    pub fn deny(
        authorizer_id: String,
        warrant_id: String,
        tool: String,
        deny_reason: String,
        failed_constraint: Option<String>,
        chain_depth: u8,
        root_principal: Option<String>,
        warrant_stack: Option<String>,
        latency_us: u64,
        request_id: String,
        arguments: Option<String>,
        approvals: Option<Vec<ApprovalRecord>>,
    ) -> Self {
        Self {
            timestamp: chrono::Utc::now().to_rfc3339(),
            authorizer_id,
            warrant_id,
            decision: "deny",
            deny_reason: Some(deny_reason),
            tool,
            failed_constraint,
            chain_depth,
            root_principal,
            warrant_stack,
            latency_us,
            request_id,
            arguments,
            approvals,
        }
    }
}

/// Signed event envelope for cryptographic receipts.
///
/// When the authorizer has a signing key configured, events are wrapped in this
/// envelope with an Ed25519 signature. The control plane verifies the signature
/// and stores the event as a non-repudiable receipt.
#[derive(Clone, Debug, Serialize)]
pub struct SignedEvent {
    /// The authorization event data
    #[serde(flatten)]
    pub event: AuthorizationEvent,
    /// Base64-encoded Ed25519 signature over the canonical signing payload.
    /// Payload format: CBOR({1: authorizer_id, 2: warrant_chain, 3: action, 4: outcome, 5: timestamp})
    pub signature: String,
    /// Base64-encoded CBOR bytes that were signed (envelope pattern).
    /// When present, the control plane verifies these exact bytes instead of
    /// reconstructing the payload, eliminating CBOR encoding mismatches.
    pub signing_payload: String,
    /// The action that was authorized (for receipt indexing)
    pub action: String,
    /// Session ID for grouping related actions (optional)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Action category for filtering (http, file, shell, etc.)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub action_category: Option<String>,
}

/// CBOR signing payload format (matches Go control plane's AuthorizerReceiptPayload).
/// Uses string keys "1", "2", etc. via serde(rename).
/// IMPORTANT: warrant_chain uses serde_bytes to encode as CBOR byte string (not array).
#[derive(serde::Serialize)]
struct ReceiptSigningPayload<'a> {
    #[serde(rename = "1")]
    authorizer_id: &'a str,
    #[serde(rename = "2", with = "serde_bytes")]
    warrant_chain: &'a [u8],
    #[serde(rename = "3")]
    action: &'a str,
    #[serde(rename = "4")]
    outcome: &'a str,
    #[serde(rename = "5")]
    timestamp: i64,
    #[serde(rename = "6", skip_serializing_if = "Option::is_none")]
    root_principal: Option<&'a str>,
}

/// Channel-based sender for audit events.
/// Clone this and pass to request handlers.
pub type AuditEventSender = mpsc::Sender<AuthorizationEvent>;

/// Create an audit event channel with the specified buffer size.
/// Returns (sender, receiver). The sender can be cloned for multiple handlers.
pub fn create_audit_channel(
    buffer_size: usize,
) -> (AuditEventSender, mpsc::Receiver<AuthorizationEvent>) {
    mpsc::channel(buffer_size)
}

// ============================================================================
// Runtime Metrics (sent with each heartbeat)
// ============================================================================

/// Runtime metrics collected and sent with each heartbeat.
#[derive(Clone, Debug, Default, Serialize)]
pub struct RuntimeMetrics {
    /// Uptime in seconds since authorizer started
    pub uptime_seconds: u64,
    /// Total requests since startup
    pub requests_total: u64,
    /// Requests since last heartbeat
    pub requests_since_last: u64,
    /// Average latency in microseconds (since last heartbeat)
    pub avg_latency_us: u64,
    /// P99 latency in microseconds (since last heartbeat)
    pub p99_latency_us: u64,
    /// Current memory usage in bytes (approximate)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub memory_bytes: Option<u64>,
}

/// Aggregate statistics sent with each heartbeat (reduces event volume).
#[derive(Clone, Debug, Default, Serialize)]
pub struct HeartbeatStats {
    /// Number of allowed requests since last heartbeat
    pub allow_count: u64,
    /// Number of denied requests since last heartbeat
    pub deny_count: u64,
    /// Top denial reasons with counts (max 10)
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub top_deny_reasons: Vec<(String, u64)>,
    /// Top tools/actions with counts (max 10)
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub top_actions: Vec<(String, u64)>,
    /// Number of unique principals seen since last heartbeat
    pub unique_principals: u64,
    /// Number of unique warrants seen since last heartbeat
    pub unique_warrants: u64,
}

/// SRL synchronization health status.
#[derive(Clone, Debug, Default, Serialize)]
pub struct SrlHealth {
    /// Last successful SRL fetch timestamp (ISO 8601)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_fetch_at: Option<String>,
    /// Whether the last fetch attempt succeeded
    pub last_fetch_success: bool,
    /// Total fetch failures since startup
    pub fetch_failures_total: u64,
    /// Total SRL verification failures since startup
    pub verification_failures_total: u64,
    /// Current SRL version (None if no SRL loaded)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub current_srl_version: Option<u64>,
}

/// Environment information sent during registration.
#[derive(Clone, Debug, Default, Serialize)]
pub struct EnvironmentInfo {
    // Kubernetes context
    #[serde(skip_serializing_if = "Option::is_none")]
    pub k8s_namespace: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub k8s_pod_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub k8s_node_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub k8s_cluster: Option<String>,

    // Cloud context
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cloud_provider: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cloud_region: Option<String>,

    // Deployment context
    #[serde(skip_serializing_if = "Option::is_none")]
    pub environment: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub deploy_id: Option<String>,

    /// SDK and Framework context for Fleet Health UI
    #[serde(skip_serializing_if = "HashMap::is_empty", default)]
    pub metadata: HashMap<String, String>,
}

impl EnvironmentInfo {
    /// Create environment info from standard environment variables.
    ///
    /// # Security Note
    ///
    /// This function ONLY reads non-sensitive metadata variables:
    /// - Kubernetes identifiers (namespace, pod name, node name)
    /// - Cloud region names (not credentials)
    /// - Deployment identifiers (environment name, build ID)
    ///
    /// It does NOT read any secrets, tokens, keys, passwords, or credentials.
    /// If you add new variables here, ensure they are safe to transmit to
    /// the control plane.
    pub fn from_env() -> Self {
        Self {
            // Kubernetes (standard downward API env vars - non-sensitive identifiers)
            k8s_namespace: std::env::var("TENUO_K8S_NAMESPACE")
                .or_else(|_| std::env::var("POD_NAMESPACE"))
                .ok(),
            k8s_pod_name: std::env::var("TENUO_K8S_POD_NAME")
                .or_else(|_| std::env::var("POD_NAME"))
                .or_else(|_| std::env::var("HOSTNAME"))
                .ok(),
            k8s_node_name: std::env::var("TENUO_K8S_NODE_NAME")
                .or_else(|_| std::env::var("NODE_NAME"))
                .ok(),
            k8s_cluster: std::env::var("TENUO_K8S_CLUSTER").ok(),

            // Cloud context (region names only - NOT credentials)
            cloud_provider: std::env::var("TENUO_CLOUD_PROVIDER").ok(),
            cloud_region: std::env::var("TENUO_CLOUD_REGION")
                .or_else(|_| std::env::var("AWS_REGION"))
                .or_else(|_| std::env::var("GOOGLE_CLOUD_REGION"))
                .ok(),

            // Deployment context (identifiers only - NOT secrets)
            environment: std::env::var("TENUO_ENVIRONMENT")
                .or_else(|_| std::env::var("ENV"))
                .or_else(|_| std::env::var("ENVIRONMENT"))
                .ok(),
            deploy_id: std::env::var("TENUO_DEPLOY_ID")
                .or_else(|_| std::env::var("BUILD_ID"))
                .or_else(|_| std::env::var("CI_COMMIT_SHA"))
                .ok(),
            metadata: HashMap::new(),
        }
    }
}

// ============================================================================
// Shared Metrics State (updated by request handlers)
// ============================================================================

/// Shared metrics state that can be updated by request handlers.
/// Clone and pass to handlers; updates are thread-safe.
#[derive(Clone)]
pub struct MetricsCollector {
    inner: Arc<MetricsCollectorInner>,
}

struct MetricsCollectorInner {
    start_time: Instant,
    requests_total: AtomicU64,
    requests_since_last: AtomicU64,
    allow_count: AtomicU64,
    deny_count: AtomicU64,

    // Latency tracking (circular buffer for p99)
    latencies: Mutex<LatencyTracker>,

    // Aggregation maps (protected by mutex)
    deny_reasons: Mutex<HashMap<String, u64>>,
    actions: Mutex<HashMap<String, u64>>,
    principals: Mutex<std::collections::HashSet<String>>,
    warrants: Mutex<std::collections::HashSet<String>>,

    // SRL health
    srl_last_fetch_at: Mutex<Option<String>>,
    srl_last_fetch_success: AtomicU64, // 0 = false, 1 = true
    srl_fetch_failures: AtomicU64,
    srl_verification_failures: AtomicU64,
    srl_current_version: AtomicU64,
}

/// Simple latency tracker with circular buffer for percentile estimation.
struct LatencyTracker {
    buffer: Vec<u64>,
    index: usize,
    total_sum: u64,
    total_count: u64,
}

impl LatencyTracker {
    fn new(capacity: usize) -> Self {
        Self {
            buffer: Vec::with_capacity(capacity),
            index: 0,
            total_sum: 0,
            total_count: 0,
        }
    }

    fn record(&mut self, latency_us: u64) {
        self.total_sum += latency_us;
        self.total_count += 1;

        if self.buffer.len() < self.buffer.capacity() {
            self.buffer.push(latency_us);
        } else {
            self.buffer[self.index] = latency_us;
            self.index = (self.index + 1) % self.buffer.capacity();
        }
    }

    fn avg(&self) -> u64 {
        if self.total_count == 0 {
            0
        } else {
            self.total_sum / self.total_count
        }
    }

    fn p99(&self) -> u64 {
        if self.buffer.is_empty() {
            return 0;
        }
        let mut sorted = self.buffer.clone();
        sorted.sort_unstable();
        let idx = ((sorted.len() as f64) * 0.99).ceil() as usize;
        sorted[idx.min(sorted.len() - 1)]
    }

    fn reset_interval(&mut self) {
        self.total_sum = 0;
        self.total_count = 0;
        // Keep buffer for p99 continuity, just reset sum/count
    }
}

impl MetricsCollector {
    /// Create a new metrics collector.
    pub fn new() -> Self {
        Self {
            inner: Arc::new(MetricsCollectorInner {
                start_time: Instant::now(),
                requests_total: AtomicU64::new(0),
                requests_since_last: AtomicU64::new(0),
                allow_count: AtomicU64::new(0),
                deny_count: AtomicU64::new(0),
                latencies: Mutex::new(LatencyTracker::new(1000)), // Keep last 1000 for p99
                deny_reasons: Mutex::new(HashMap::new()),
                actions: Mutex::new(HashMap::new()),
                principals: Mutex::new(std::collections::HashSet::new()),
                warrants: Mutex::new(std::collections::HashSet::new()),
                srl_last_fetch_at: Mutex::new(None),
                srl_last_fetch_success: AtomicU64::new(0),
                srl_fetch_failures: AtomicU64::new(0),
                srl_verification_failures: AtomicU64::new(0),
                srl_current_version: AtomicU64::new(0),
            }),
        }
    }

    /// Record an authorization decision.
    pub async fn record_authorization(
        &self,
        allowed: bool,
        tool: &str,
        latency_us: u64,
        warrant_id: &str,
        principal: Option<&str>,
        deny_reason: Option<&str>,
    ) {
        self.inner.requests_total.fetch_add(1, Ordering::Relaxed);
        self.inner
            .requests_since_last
            .fetch_add(1, Ordering::Relaxed);

        if allowed {
            self.inner.allow_count.fetch_add(1, Ordering::Relaxed);
        } else {
            self.inner.deny_count.fetch_add(1, Ordering::Relaxed);
            if let Some(reason) = deny_reason {
                let mut reasons = self.inner.deny_reasons.lock().await;
                *reasons.entry(reason.to_string()).or_insert(0) += 1;
            }
        }

        // Record latency
        {
            let mut latencies = self.inner.latencies.lock().await;
            latencies.record(latency_us);
        }

        // Track unique warrants and principals
        {
            let mut warrants = self.inner.warrants.lock().await;
            warrants.insert(warrant_id.to_string());
        }
        if let Some(p) = principal {
            let mut principals = self.inner.principals.lock().await;
            principals.insert(p.to_string());
        }

        // Track tool usage
        {
            let mut actions = self.inner.actions.lock().await;
            *actions.entry(tool.to_string()).or_insert(0) += 1;
        }
    }

    /// Record SRL fetch result.
    pub async fn record_srl_fetch(&self, success: bool, version: Option<u64>) {
        if success {
            self.inner
                .srl_last_fetch_success
                .store(1, Ordering::Relaxed);
            if let Some(v) = version {
                self.inner.srl_current_version.store(v, Ordering::Relaxed);
            }
            let mut last_fetch = self.inner.srl_last_fetch_at.lock().await;
            *last_fetch = Some(chrono::Utc::now().to_rfc3339());
        } else {
            self.inner
                .srl_last_fetch_success
                .store(0, Ordering::Relaxed);
            self.inner
                .srl_fetch_failures
                .fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Record SRL verification failure.
    pub fn record_srl_verification_failure(&self) {
        self.inner
            .srl_verification_failures
            .fetch_add(1, Ordering::Relaxed);
    }

    /// Collect runtime metrics for heartbeat.
    pub async fn collect_runtime_metrics(&self) -> RuntimeMetrics {
        let latencies = self.inner.latencies.lock().await;
        RuntimeMetrics {
            uptime_seconds: self.inner.start_time.elapsed().as_secs(),
            requests_total: self.inner.requests_total.load(Ordering::Relaxed),
            requests_since_last: self.inner.requests_since_last.load(Ordering::Relaxed),
            avg_latency_us: latencies.avg(),
            p99_latency_us: latencies.p99(),
            memory_bytes: get_memory_usage(),
        }
    }

    /// Collect heartbeat stats and reset interval counters.
    pub async fn collect_and_reset_stats(&self) -> HeartbeatStats {
        // Collect deny reasons (top 10)
        let top_deny_reasons = {
            let mut reasons = self.inner.deny_reasons.lock().await;
            let mut sorted: Vec<_> = reasons.drain().collect();
            sorted.sort_by(|a, b| b.1.cmp(&a.1));
            sorted.truncate(10);
            sorted
        };

        // Collect top actions (top 10)
        let top_actions = {
            let mut actions = self.inner.actions.lock().await;
            let mut sorted: Vec<_> = actions.drain().collect();
            sorted.sort_by(|a, b| b.1.cmp(&a.1));
            sorted.truncate(10);
            sorted
        };

        // Collect unique counts and reset
        let unique_principals = {
            let mut principals = self.inner.principals.lock().await;
            let count = principals.len() as u64;
            principals.clear();
            count
        };

        let unique_warrants = {
            let mut warrants = self.inner.warrants.lock().await;
            let count = warrants.len() as u64;
            warrants.clear();
            count
        };

        // Get counts and reset interval counters
        let allow_count = self.inner.allow_count.swap(0, Ordering::Relaxed);
        let deny_count = self.inner.deny_count.swap(0, Ordering::Relaxed);
        self.inner.requests_since_last.store(0, Ordering::Relaxed);

        // Reset latency interval stats
        {
            let mut latencies = self.inner.latencies.lock().await;
            latencies.reset_interval();
        }

        HeartbeatStats {
            allow_count,
            deny_count,
            top_deny_reasons,
            top_actions,
            unique_principals,
            unique_warrants,
        }
    }

    /// Collect SRL health status.
    pub async fn collect_srl_health(&self) -> SrlHealth {
        let last_fetch_at = self.inner.srl_last_fetch_at.lock().await.clone();
        let version = self.inner.srl_current_version.load(Ordering::Relaxed);
        SrlHealth {
            last_fetch_at,
            last_fetch_success: self.inner.srl_last_fetch_success.load(Ordering::Relaxed) == 1,
            fetch_failures_total: self.inner.srl_fetch_failures.load(Ordering::Relaxed),
            verification_failures_total: self
                .inner
                .srl_verification_failures
                .load(Ordering::Relaxed),
            // Convert 0 to None (0 is internal sentinel for "no SRL")
            current_srl_version: if version == 0 { None } else { Some(version) },
        }
    }
}

impl Default for MetricsCollector {
    fn default() -> Self {
        Self::new()
    }
}

/// Get current process memory usage (platform-dependent).
fn get_memory_usage() -> Option<u64> {
    #[cfg(target_os = "linux")]
    {
        // Read from /proc/self/statm
        if let Ok(statm) = std::fs::read_to_string("/proc/self/statm") {
            if let Some(rss_pages) = statm.split_whitespace().nth(1) {
                if let Ok(pages) = rss_pages.parse::<u64>() {
                    // Page size is typically 4096
                    return Some(pages * 4096);
                }
            }
        }
        None
    }

    #[cfg(target_os = "macos")]
    {
        // On macOS, we'd need mach APIs which are complex
        // Return None for now; could add via libc later
        None
    }

    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    {
        None
    }
}

// ============================================================================
// Heartbeat Configuration
// ============================================================================

/// Configuration for the heartbeat client.
#[derive(Clone)]
pub struct HeartbeatConfig {
    /// Control plane URL
    pub control_plane_url: String,
    /// API key for authentication
    pub api_key: String,
    /// Human-readable name for this authorizer
    pub authorizer_name: String,
    /// Type of authorizer (e.g., "sidecar", "gateway")
    pub authorizer_type: String,
    /// Full version string (e.g., "0.1.0-beta.7+authz.1")
    pub version: String,
    /// Interval between heartbeats in seconds
    pub interval_secs: u64,
    /// Shared authorizer for SRL updates (optional)
    pub authorizer: Option<Arc<RwLock<Authorizer>>>,
    /// Trusted root public key for SRL verification
    pub trusted_root: Option<PublicKey>,
    /// Maximum events to batch before flushing (default: 100)
    pub audit_batch_size: usize,
    /// Flush interval for audit events in seconds (default: 10)
    pub audit_flush_interval_secs: u64,
    /// Environment information for registration
    pub environment: EnvironmentInfo,
    /// Shared metrics collector (optional, for metrics reporting)
    pub metrics: Option<MetricsCollector>,
    /// Signing key for cryptographic receipts.
    /// Events are signed by the authorizer and become non-repudiable receipts.
    /// The public key is sent during registration for verification on the control plane.
    pub signing_key: SigningKey,
    /// Optional watch channel sender notified immediately after the authorizer
    /// ID is assigned by the control plane. Python bindings use this to avoid
    /// polling; all other callers leave this as `None`.
    pub id_notify: Option<tokio::sync::watch::Sender<Option<String>>>,
    /// Optional agent ID to bind this authorizer to a registered agent.
    /// Set automatically when constructed via [`HeartbeatConfig::from_connect_token`].
    pub agent_id: Option<String>,
    /// Optional connect token for auto-claiming the agent before the first
    /// heartbeat. Stored so the heartbeat loop can call `claim_agent` at startup.
    pub connect_token: Option<crate::connect_token::ConnectToken>,
}

impl Default for HeartbeatConfig {
    fn default() -> Self {
        let signing_key = SigningKey::generate();
        Self {
            control_plane_url: String::new(),
            api_key: String::new(),
            authorizer_name: String::new(),
            authorizer_type: "sidecar".to_string(),
            version: String::new(),
            interval_secs: 30,
            authorizer: None,
            trusted_root: None,
            audit_batch_size: 100,
            audit_flush_interval_secs: 10,
            environment: EnvironmentInfo::default(),
            metrics: None,
            signing_key,
            id_notify: None,
            agent_id: None,
            connect_token: None,
        }
    }
}

impl HeartbeatConfig {
    /// Build a [`HeartbeatConfig`] from a connect token string.
    ///
    /// Parses the token, generates a fresh signing key, and populates
    /// `control_plane_url`, `api_key`, and `agent_id` from the token payload.
    /// The caller still needs to set `authorizer_name` and `authorizer_type`.
    pub fn from_connect_token(
        raw_token: &str,
        authorizer_name: &str,
        authorizer_type: &str,
    ) -> Result<Self, crate::connect_token::ConnectTokenError> {
        let ct = crate::connect_token::ConnectToken::parse(raw_token)?;
        let signing_key = SigningKey::generate();

        Ok(Self {
            control_plane_url: ct.endpoint.clone(),
            api_key: ct.api_key.clone(),
            authorizer_name: authorizer_name.to_string(),
            authorizer_type: authorizer_type.to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
            signing_key,
            agent_id: ct.agent_id.clone(),
            connect_token: Some(ct),
            ..Default::default()
        })
    }
}

/// Request body for authorizer registration.
/// Fields are flattened to match the Go control plane's expected format.
#[derive(Serialize)]
struct RegisterRequest<'a> {
    name: &'a str,
    #[serde(rename = "type")]
    authorizer_type: &'a str,
    version: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    public_key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    agent_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<&'a HashMap<String, String>>,
    // Flattened environment fields (Go control plane expects these at top level)
    #[serde(skip_serializing_if = "Option::is_none")]
    environment: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    k8s_namespace: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    k8s_pod_name: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    k8s_cluster: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cloud_provider: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cloud_region: Option<&'a str>,
}

/// Response from authorizer registration.
#[derive(Deserialize)]
struct RegisterResponse {
    id: String,
}

/// Request body for heartbeat with metrics.
#[derive(Serialize)]
struct HeartbeatRequest {
    /// Current SRL version (None if no SRL loaded)
    #[serde(skip_serializing_if = "Option::is_none")]
    srl_version: Option<u64>,
    /// Runtime metrics
    #[serde(skip_serializing_if = "Option::is_none")]
    metrics: Option<RuntimeMetrics>,
    /// Aggregate stats since last heartbeat
    #[serde(skip_serializing_if = "Option::is_none")]
    stats: Option<HeartbeatStats>,
    /// SRL synchronization health
    #[serde(skip_serializing_if = "Option::is_none")]
    srl_health: Option<SrlHealth>,
}

/// Response from heartbeat endpoint.
#[derive(Deserialize)]
struct HeartbeatResponse {
    /// Status of the authorizer (e.g., "active")
    #[allow(dead_code)]
    status: String,
    /// Latest SRL version available on the control plane
    #[serde(default)]
    latest_srl_version: Option<u64>,
    /// Urgent refresh flag - immediate SRL update needed
    #[serde(default)]
    refresh_required: bool,
}

/// Response from SRL endpoint (CBOR or JSON wrapper).
#[derive(Deserialize)]
struct SrlResponse {
    /// Base64-encoded CBOR SRL
    srl: String,
    /// SRL version
    version: u64,
}

/// Start the heartbeat loop in the background (legacy signature for backwards compatibility).
///
/// This function will:
/// 1. Attempt to register with the control plane (with retries)
/// 2. If successful, fetch initial SRL
/// 3. Start sending periodic heartbeats
/// 4. If registration fails, log a warning and return (authorizer runs standalone)
/// 5. On each heartbeat, check if SRL needs updating and fetch if needed
///
/// This function is designed to be spawned as a tokio task and will run
/// indefinitely until the process exits.
pub async fn start_heartbeat_loop(config: HeartbeatConfig) {
    start_heartbeat_loop_with_audit(config, None).await;
}

/// Start the heartbeat and audit event loops in the background.
///
/// This function will:
/// 1. Attempt to register with the control plane (with retries)
/// 2. If successful, fetch initial SRL and spawn audit event flush task
/// 3. Start sending periodic heartbeats
/// 4. If registration fails, log a warning and return (authorizer runs standalone)
/// 5. On each heartbeat, check if SRL needs updating and fetch if needed
///
/// If `audit_rx` is provided, audit events will be batched and sent to the control plane.
///
/// This function is designed to be spawned as a tokio task and will run
/// indefinitely until the process exits.
pub async fn start_heartbeat_loop_with_audit(
    config: HeartbeatConfig,
    audit_rx: Option<mpsc::Receiver<AuthorizationEvent>>,
) {
    start_heartbeat_loop_with_audit_and_id(config, audit_rx, Arc::new(RwLock::new(None))).await;
}

/// Start the heartbeat and audit event loops in the background, with shared authorizer_id.
///
/// Same as `start_heartbeat_loop_with_audit`, but writes the authorizer_id to the provided
/// shared state after registration. This allows request handlers to include the authorizer_id
/// in audit events.
pub async fn start_heartbeat_loop_with_audit_and_id(
    config: HeartbeatConfig,
    audit_rx: Option<mpsc::Receiver<AuthorizationEvent>>,
    shared_authorizer_id: Arc<RwLock<Option<String>>>,
) {
    let client = Client::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .expect("Failed to create HTTP client");

    // Auto-claim agent if a connect token with agent binding is present.
    // This is idempotent — repeated claims by other authorizer instances
    // sharing the same token are silently accepted.
    if let Some(ref ct) = config.connect_token {
        if let Err(e) = ct.claim_agent(&config.signing_key).await {
            warn!(error = %e, "connect token agent claim failed (non-fatal)");
        }
    }

    // Register with retry
    let authorizer_id = match register_with_retry(&client, &config).await {
        Some(id) => id,
        None => {
            warn!(
                "Failed to register with control plane after 3 attempts. \
                 Authorizer will run in standalone mode without heartbeats."
            );
            return;
        }
    };

    info!(
        authorizer_id = %authorizer_id,
        name = %config.authorizer_name,
        "Registered with control plane"
    );

    // Store authorizer_id in shared state for request handlers
    {
        let mut id_guard = shared_authorizer_id.write().await;
        *id_guard = Some(authorizer_id.clone());
    }

    // Notify any watch-based subscriber (e.g. Python bindings) immediately,
    // eliminating the need for a polling watcher.
    if let Some(ref tx) = config.id_notify {
        let _ = tx.send(Some(authorizer_id.clone()));
    }

    // Track local SRL version (0 = no SRL loaded)
    let mut local_srl_version: u64 = 0;

    // Fetch initial SRL
    if let (Some(ref authorizer), Some(ref trusted_root)) =
        (&config.authorizer, &config.trusted_root)
    {
        match fetch_and_apply_srl(&client, &config, authorizer, trusted_root).await {
            Ok(version) => {
                local_srl_version = version;
                info!(srl_version = version, "Initial SRL fetched");
                // Record successful fetch
                if let Some(ref metrics) = config.metrics {
                    metrics.record_srl_fetch(true, Some(version)).await;
                }
            }
            Err(e) => {
                warn!(error = %e, "Failed to fetch initial SRL, will retry on heartbeat");
                // Record failed fetch
                if let Some(ref metrics) = config.metrics {
                    metrics.record_srl_fetch(false, None).await;
                }
            }
        }
    }

    // Spawn audit event flush task if receiver provided
    if let Some(rx) = audit_rx {
        let audit_client = client.clone();
        let audit_config = config.clone();
        let audit_authorizer_id = authorizer_id.clone();
        tokio::spawn(async move {
            run_audit_flush_loop(audit_client, audit_config, audit_authorizer_id, rx).await;
        });
        info!("Audit event streaming enabled");
    }

    // Heartbeat loop
    let mut ticker = interval(Duration::from_secs(config.interval_secs));

    // Skip the first immediate tick
    ticker.tick().await;

    loop {
        ticker.tick().await;

        match send_heartbeat(&client, &config, &authorizer_id).await {
            Ok(response) => {
                debug!(
                    authorizer_id = %authorizer_id,
                    "Heartbeat sent successfully"
                );

                // Check if SRL update is needed
                let needs_update = response.refresh_required
                    || response
                        .latest_srl_version
                        .map(|v| v > local_srl_version)
                        .unwrap_or(false);

                if needs_update {
                    if let Some(ref authorizer) = config.authorizer {
                        if let Some(ref trusted_root) = config.trusted_root {
                            match fetch_and_apply_srl(&client, &config, authorizer, trusted_root)
                                .await
                            {
                                Ok(new_version) => {
                                    local_srl_version = new_version;
                                    info!(
                                        srl_version = new_version,
                                        refresh_required = response.refresh_required,
                                        "SRL updated from control plane"
                                    );
                                    // Record successful fetch
                                    if let Some(ref metrics) = config.metrics {
                                        metrics.record_srl_fetch(true, Some(new_version)).await;
                                    }
                                }
                                Err(e) => {
                                    warn!(
                                        error = %e,
                                        "Failed to fetch SRL from control plane"
                                    );
                                    // Record failed fetch
                                    if let Some(ref metrics) = config.metrics {
                                        metrics.record_srl_fetch(false, None).await;
                                    }
                                }
                            }
                        } else {
                            warn!("SRL update needed but no trusted root configured");
                        }
                    }
                }
            }
            Err(e) => {
                warn!(
                    authorizer_id = %authorizer_id,
                    error = %e,
                    "Heartbeat failed, will retry on next interval"
                );
            }
        }
    }
}

/// Run the audit event flush loop.
/// Collects events from the channel and flushes them in batches.
async fn run_audit_flush_loop(
    client: Client,
    config: HeartbeatConfig,
    authorizer_id: String,
    mut rx: mpsc::Receiver<AuthorizationEvent>,
) {
    let mut buffer: Vec<AuthorizationEvent> = Vec::with_capacity(config.audit_batch_size);
    let mut flush_ticker = interval(Duration::from_secs(config.audit_flush_interval_secs));

    // Skip the first immediate tick
    flush_ticker.tick().await;

    loop {
        tokio::select! {
            // Receive events from channel
            event = rx.recv() => {
                match event {
                    Some(e) => {
                        buffer.push(e);

                        // Flush if batch is full
                        if buffer.len() >= config.audit_batch_size {
                            flush_audit_events(&client, &config, &authorizer_id, &mut buffer).await;
                        }
                    }
                    None => {
                        // Channel closed (sender dropped), flush remaining events and exit
                        if !buffer.is_empty() {
                            info!(
                                authorizer_id = %authorizer_id,
                                remaining_events = buffer.len(),
                                "Flushing remaining audit events before shutdown"
                            );
                            flush_audit_events(&client, &config, &authorizer_id, &mut buffer).await;
                        }
                        info!("Audit channel closed, exiting flush loop");
                        break;
                    }
                }
            }
            // Periodic flush
            _ = flush_ticker.tick() => {
                if !buffer.is_empty() {
                    flush_audit_events(&client, &config, &authorizer_id, &mut buffer).await;
                }
            }
        }
    }
}

/// Sign an authorization event to create a cryptographic receipt.
///
/// Creates a CBOR signing payload and signs it with the authorizer's key.
/// The signature covers: (authorizer_id, warrant_chain, action, outcome, timestamp, root_principal)
///
/// `authorizer_id_override` ensures the signing payload uses the real registered
/// authorizer ID even if the event was created before registration completed.
fn sign_event(
    event: &AuthorizationEvent,
    signing_key: &SigningKey,
    authorizer_id_override: &str,
) -> SignedEvent {
    // Decode warrant stack if present (base64 -> bytes)
    let warrant_chain_bytes = event
        .warrant_stack
        .as_ref()
        .and_then(|ws| base64::Engine::decode(&base64::engine::general_purpose::STANDARD, ws).ok())
        .unwrap_or_default();

    // Build action string from tool
    let action = format!("tool:{}", event.tool);

    // Parse timestamp to Unix epoch
    let timestamp = chrono::DateTime::parse_from_rfc3339(&event.timestamp)
        .map(|dt| dt.timestamp())
        .unwrap_or_else(|_| chrono::Utc::now().timestamp());

    // Build signing payload using the canonical authorizer ID (not the
    // potentially-stale value the event was created with).
    let payload = ReceiptSigningPayload {
        authorizer_id: authorizer_id_override,
        warrant_chain: &warrant_chain_bytes,
        action: &action,
        outcome: event.decision,
        timestamp,
        root_principal: event.root_principal.as_deref(),
    };

    // Encode to CBOR
    let mut payload_buf = Vec::new();
    if ciborium::into_writer(&payload, &mut payload_buf).is_err() {
        payload_buf.clear();
    }

    // Sign the payload
    let signature = signing_key.sign(&payload_buf);
    let signature_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        signature.to_bytes(),
    );

    // Include the exact signed bytes so the control plane can verify
    // without reconstructing the CBOR (envelope pattern).
    let signing_payload_b64 =
        base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &payload_buf);

    // Patch the event's authorizer_id to match what was signed
    let mut patched_event = event.clone();
    patched_event.authorizer_id = authorizer_id_override.to_string();

    SignedEvent {
        event: patched_event,
        signature: signature_b64,
        signing_payload: signing_payload_b64,
        action,
        session_id: None,
        action_category: Some("tool".to_string()),
    }
}

/// Flush buffered audit events to the control plane.
/// If a signing key is configured, events are signed before sending.
async fn flush_audit_events(
    client: &Client,
    config: &HeartbeatConfig,
    authorizer_id: &str,
    buffer: &mut Vec<AuthorizationEvent>,
) {
    if buffer.is_empty() {
        return;
    }

    let event_count = buffer.len();
    let url = format!(
        "{}/v1/authorizers/{}/events",
        config.control_plane_url, authorizer_id
    );

    // Sign all events, using the canonical authorizer_id from registration
    // (events may carry a stale "pending" ID if emitted before registration completed)
    let signed_events: Vec<SignedEvent> = buffer
        .iter()
        .map(|event| sign_event(event, &config.signing_key, authorizer_id))
        .collect();

    let result = client
        .post(&url)
        .header("Authorization", format!("Bearer {}", config.api_key))
        .header("Content-Type", "application/json")
        .json(&signed_events)
        .send()
        .await;

    match result {
        Ok(response) if response.status().is_success() => {
            debug!(
                authorizer_id = %authorizer_id,
                event_count = %event_count,
                "Flushed audit events to control plane"
            );
            eprintln!(
                "[tenuo] flushed {} audit events for {}",
                event_count, authorizer_id
            );
            buffer.clear();
        }
        Ok(response) => {
            let status = response.status();
            let body = response.text().await.unwrap_or_default();
            warn!(
                authorizer_id = %authorizer_id,
                event_count = %event_count,
                status = %status,
                "Failed to flush audit events, will retry"
            );
            eprintln!(
                "[tenuo] WARN: flush failed status={} body={} (authorizer={})",
                status,
                &body[..body.len().min(200)],
                authorizer_id
            );
            // Keep events in buffer for retry, but cap size to prevent unbounded growth
            if buffer.len() > config.audit_batch_size * 10 {
                let drain_count = buffer.len() - config.audit_batch_size;
                warn!(
                    dropped_events = %drain_count,
                    "Dropping oldest audit events due to buffer overflow"
                );
                buffer.drain(0..drain_count);
            }
        }
        Err(e) => {
            warn!(
                authorizer_id = %authorizer_id,
                event_count = %event_count,
                error = %e,
                "Network error flushing audit events, will retry"
            );
            // Same overflow protection
            if buffer.len() > config.audit_batch_size * 10 {
                let drain_count = buffer.len() - config.audit_batch_size;
                warn!(
                    dropped_events = %drain_count,
                    "Dropping oldest audit events due to buffer overflow"
                );
                buffer.drain(0..drain_count);
            }
        }
    }
}

/// Attempt to register with the control plane, retrying up to 3 times.
async fn register_with_retry(client: &Client, config: &HeartbeatConfig) -> Option<String> {
    const MAX_ATTEMPTS: u32 = 3;

    for attempt in 1..=MAX_ATTEMPTS {
        match register(client, config).await {
            Ok(id) => return Some(id),
            Err(e) => {
                let backoff = Duration::from_secs(2u64.pow(attempt));
                warn!(
                    attempt = attempt,
                    max_attempts = MAX_ATTEMPTS,
                    error = %e,
                    backoff_secs = backoff.as_secs(),
                    "Registration attempt failed, retrying..."
                );
                eprintln!(
                    "[tenuo] registration attempt {}/{} failed: {} (retry in {}s)",
                    attempt,
                    MAX_ATTEMPTS,
                    e,
                    backoff.as_secs(),
                );

                if attempt < MAX_ATTEMPTS {
                    tokio::time::sleep(backoff).await;
                }
            }
        }
    }

    eprintln!(
        "[tenuo] WARN: failed to register '{}' with {} after {} attempts. \
         Running in standalone mode — events will not reach the dashboard.",
        config.authorizer_name, config.control_plane_url, MAX_ATTEMPTS,
    );
    None
}

/// Register this authorizer with the control plane.
async fn register(client: &Client, config: &HeartbeatConfig) -> Result<String, HeartbeatError> {
    let url = format!("{}/v1/authorizers/register", config.control_plane_url);

    // Get public key hex from signing key
    let public_key_hex = hex::encode(config.signing_key.public_key().to_bytes());

    let env = &config.environment;
    let meta = if config.environment.metadata.is_empty() {
        None
    } else {
        Some(&config.environment.metadata)
    };
    let request_body = RegisterRequest {
        name: &config.authorizer_name,
        authorizer_type: &config.authorizer_type,
        version: &config.version,
        public_key: Some(public_key_hex),
        agent_id: config.agent_id.as_deref(),
        metadata: meta,
        environment: env.environment.as_deref(),
        k8s_namespace: env.k8s_namespace.as_deref(),
        k8s_pod_name: env.k8s_pod_name.as_deref(),
        k8s_cluster: env.k8s_cluster.as_deref(),
        cloud_provider: env.cloud_provider.as_deref(),
        cloud_region: env.cloud_region.as_deref(),
    };

    let response = client
        .post(&url)
        .header("Authorization", format!("Bearer {}", config.api_key))
        .header("Content-Type", "application/json")
        .json(&request_body)
        .send()
        .await
        .map_err(|e| HeartbeatError::Network(e.to_string()))?;

    if !response.status().is_success() {
        let status = response.status();
        let body = response
            .text()
            .await
            .unwrap_or_else(|_| "<no body>".to_string());
        return Err(HeartbeatError::Api {
            status: status.as_u16(),
            message: body,
        });
    }

    let register_response: RegisterResponse = response
        .json()
        .await
        .map_err(|e| HeartbeatError::Parse(e.to_string()))?;

    Ok(register_response.id)
}

/// Send a heartbeat to the control plane with metrics.
async fn send_heartbeat(
    client: &Client,
    config: &HeartbeatConfig,
    authorizer_id: &str,
) -> Result<HeartbeatResponse, HeartbeatError> {
    let url = format!(
        "{}/v1/authorizers/{}/heartbeat",
        config.control_plane_url, authorizer_id
    );

    // Collect metrics if collector is available
    let (metrics, stats, srl_health) = if let Some(ref collector) = config.metrics {
        let metrics = collector.collect_runtime_metrics().await;
        let stats = collector.collect_and_reset_stats().await;
        let srl_health = collector.collect_srl_health().await;
        (Some(metrics), Some(stats), Some(srl_health))
    } else {
        (None, None, None)
    };

    // Extract srl_version from srl_health for top-level field (Go control plane convenience)
    let srl_version = srl_health.as_ref().and_then(|h| h.current_srl_version);

    let request_body = HeartbeatRequest {
        srl_version,
        metrics,
        stats,
        srl_health,
    };

    let response = client
        .post(&url)
        .header("Authorization", format!("Bearer {}", config.api_key))
        .header("Content-Type", "application/json")
        .json(&request_body)
        .send()
        .await
        .map_err(|e| HeartbeatError::Network(e.to_string()))?;

    if !response.status().is_success() {
        let status = response.status();
        let body = response
            .text()
            .await
            .unwrap_or_else(|_| "<no body>".to_string());
        return Err(HeartbeatError::Api {
            status: status.as_u16(),
            message: body,
        });
    }

    let heartbeat_response: HeartbeatResponse = response
        .json()
        .await
        .map_err(|e| HeartbeatError::Parse(e.to_string()))?;

    Ok(heartbeat_response)
}

/// Fetch the latest SRL from the control plane and apply it to the authorizer.
async fn fetch_and_apply_srl(
    client: &Client,
    config: &HeartbeatConfig,
    authorizer: &Arc<RwLock<Authorizer>>,
    trusted_root: &PublicKey,
) -> Result<u64, HeartbeatError> {
    let url = format!("{}/v1/revocations/srl/signed", config.control_plane_url);

    let response = client
        .get(&url)
        .header("Authorization", format!("Bearer {}", config.api_key))
        .send()
        .await
        .map_err(|e| HeartbeatError::Network(e.to_string()))?;

    if !response.status().is_success() {
        let status = response.status();
        let body = response
            .text()
            .await
            .unwrap_or_else(|_| "<no body>".to_string());
        return Err(HeartbeatError::Api {
            status: status.as_u16(),
            message: body,
        });
    }

    let srl_response: SrlResponse = response
        .json()
        .await
        .map_err(|e| HeartbeatError::Parse(e.to_string()))?;

    // Decode base64 SRL
    let srl_bytes = base64::Engine::decode(
        &base64::engine::general_purpose::STANDARD,
        &srl_response.srl,
    )
    .map_err(|e| HeartbeatError::Parse(format!("Invalid base64 SRL: {}", e)))?;

    // Parse and verify SRL
    let srl = SignedRevocationList::from_bytes(&srl_bytes)
        .map_err(|e| HeartbeatError::Parse(format!("Invalid SRL format: {}", e)))?;

    // Apply to authorizer (this also verifies the signature)
    let mut auth = authorizer.write().await;
    auth.set_revocation_list(srl, trusted_root)
        .map_err(|e| HeartbeatError::Parse(format!("SRL verification failed: {}", e)))?;

    Ok(srl_response.version)
}

/// Errors that can occur during heartbeat operations.
#[derive(Debug)]
pub enum HeartbeatError {
    /// Network error (connection failed, timeout, etc.)
    Network(String),
    /// API returned an error status
    Api { status: u16, message: String },
    /// Failed to parse response
    Parse(String),
}

impl std::fmt::Display for HeartbeatError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            HeartbeatError::Network(msg) => write!(f, "Network error: {}", msg),
            HeartbeatError::Api { status, message } => {
                write!(f, "API error ({}): {}", status, message)
            }
            HeartbeatError::Parse(msg) => write!(f, "Parse error: {}", msg),
        }
    }
}

impl std::error::Error for HeartbeatError {}

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

    fn test_config() -> HeartbeatConfig {
        HeartbeatConfig {
            control_plane_url: "https://api.tenuo.cloud".to_string(),
            api_key: "tc_test".to_string(),
            authorizer_name: "test-auth".to_string(),
            authorizer_type: "sidecar".to_string(),
            version: "0.1.0-beta.7+authz.1".to_string(),
            interval_secs: 30,
            authorizer: None,
            trusted_root: None,
            audit_batch_size: 100,
            audit_flush_interval_secs: 10,
            environment: EnvironmentInfo::default(),
            metrics: None,
            signing_key: SigningKey::generate(),
            id_notify: None,
            agent_id: None,
            connect_token: None,
        }
    }

    #[test]
    fn test_environment_info_default() {
        let env = EnvironmentInfo::default();
        assert!(env.k8s_namespace.is_none());
        assert!(env.cloud_provider.is_none());
        assert!(env.environment.is_none());
    }

    #[test]
    fn test_runtime_metrics_default() {
        let metrics = RuntimeMetrics::default();
        assert_eq!(metrics.uptime_seconds, 0);
        assert_eq!(metrics.requests_total, 0);
    }

    #[test]
    fn test_heartbeat_stats_default() {
        let stats = HeartbeatStats::default();
        assert_eq!(stats.allow_count, 0);
        assert_eq!(stats.deny_count, 0);
        assert!(stats.top_deny_reasons.is_empty());
    }

    #[test]
    fn test_srl_health_default() {
        let health = SrlHealth::default();
        assert!(health.last_fetch_at.is_none());
        assert!(!health.last_fetch_success);
        assert_eq!(health.current_srl_version, None);
    }

    #[tokio::test]
    async fn test_metrics_collector_record_authorization() {
        let collector = MetricsCollector::new();

        // Record an allowed request
        collector
            .record_authorization(true, "read_file", 100, "wid-1", Some("user-1"), None)
            .await;

        // Record a denied request
        collector
            .record_authorization(
                false,
                "write_file",
                200,
                "wid-2",
                Some("user-1"),
                Some("constraint_violation"),
            )
            .await;

        let stats = collector.collect_and_reset_stats().await;
        assert_eq!(stats.allow_count, 1);
        assert_eq!(stats.deny_count, 1);
        assert_eq!(stats.unique_principals, 1);
        assert_eq!(stats.unique_warrants, 2);
    }

    #[tokio::test]
    async fn test_metrics_collector_latency() {
        let collector = MetricsCollector::new();

        // Record some latencies
        for i in 1..=100 {
            collector
                .record_authorization(true, "test", i * 10, &format!("wid-{}", i), None, None)
                .await;
        }

        let metrics = collector.collect_runtime_metrics().await;
        assert_eq!(metrics.requests_total, 100);
        assert!(metrics.avg_latency_us > 0);
        assert!(metrics.p99_latency_us >= metrics.avg_latency_us);
    }

    #[test]
    fn test_authorization_event_allow() {
        let event = AuthorizationEvent::allow(
            "auth-123".to_string(),
            "wid-456".to_string(),
            "read_file".to_string(),
            0,
            Some("root-pk".to_string()),
            Some("base64stack".to_string()),
            1234,
            "req-789".to_string(),
            None,
            None,
        );
        assert_eq!(event.decision, "allow");
        assert!(event.deny_reason.is_none());
        assert_eq!(event.tool, "read_file");
        assert_eq!(event.chain_depth, 0);
        assert_eq!(event.warrant_stack, Some("base64stack".to_string()));
        assert!(event.approvals.is_none());
    }

    #[test]
    fn test_authorization_event_deny() {
        let event = AuthorizationEvent::deny(
            "auth-123".to_string(),
            "wid-456".to_string(),
            "write_file".to_string(),
            "constraint_violation".to_string(),
            Some("path".to_string()),
            1,
            Some("root-pk".to_string()),
            Some("base64stack".to_string()),
            5678,
            "req-999".to_string(),
            None,
            None,
        );
        assert_eq!(event.decision, "deny");
        assert_eq!(event.deny_reason, Some("constraint_violation".to_string()));
        assert_eq!(event.failed_constraint, Some("path".to_string()));
        assert_eq!(event.chain_depth, 1);
        assert_eq!(event.warrant_stack, Some("base64stack".to_string()));
        assert!(event.approvals.is_none());
    }

    #[test]
    fn test_authorization_event_with_approvals() {
        let records = vec![ApprovalRecord {
            approver_key: "abcd1234".to_string(),
            external_id: "arn:aws:iam::123:user/admin".to_string(),
            approved_at: 1700000000,
            expires_at: 1700000300,
            request_hash: "deadbeef".to_string(),
        }];
        let event = AuthorizationEvent::allow(
            "auth-123".to_string(),
            "wid-456".to_string(),
            "execute_payment".to_string(),
            2,
            Some("root-pk".to_string()),
            None,
            500,
            "req-approval".to_string(),
            None,
            Some(records),
        );
        assert_eq!(event.decision, "allow");
        let approvals = event.approvals.as_ref().unwrap();
        assert_eq!(approvals.len(), 1);
        assert_eq!(approvals[0].external_id, "arn:aws:iam::123:user/admin");
        assert_eq!(approvals[0].approved_at, 1700000000);
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"approvals\""));
        assert!(json.contains("arn:aws:iam::123:user/admin"));
    }

    #[test]
    fn test_authorization_event_serialization() {
        let event = AuthorizationEvent::allow(
            "auth-123".to_string(),
            "wid-456".to_string(),
            "read_file".to_string(),
            0,
            None,
            None, // No warrant stack
            1234,
            "req-789".to_string(),
            None,
            None,
        );
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"decision\":\"allow\""));
        assert!(json.contains("\"tool\":\"read_file\""));
        // Optional fields should be omitted when None
        assert!(!json.contains("deny_reason"));
        assert!(!json.contains("warrant_stack"));
        assert!(!json.contains("approvals"));
    }

    #[test]
    fn test_create_audit_channel() {
        let (tx, _rx) = create_audit_channel(100);
        // Should be able to clone the sender
        let _tx2 = tx.clone();
    }

    #[test]
    fn test_heartbeat_config_clone() {
        let config = test_config();

        let cloned = config.clone();
        assert_eq!(cloned.control_plane_url, config.control_plane_url);
        assert_eq!(cloned.api_key, config.api_key);
        assert_eq!(cloned.authorizer_name, config.authorizer_name);
    }

    #[test]
    fn test_heartbeat_error_display() {
        let network_err = HeartbeatError::Network("connection refused".to_string());
        assert!(network_err.to_string().contains("Network error"));

        let api_err = HeartbeatError::Api {
            status: 401,
            message: "Unauthorized".to_string(),
        };
        assert!(api_err.to_string().contains("401"));

        let parse_err = HeartbeatError::Parse("invalid json".to_string());
        assert!(parse_err.to_string().contains("Parse error"));
    }

    #[test]
    fn test_heartbeat_response_deserialization() {
        // Test minimal response (backwards compatible)
        let minimal = r#"{"status": "active"}"#;
        let resp: HeartbeatResponse = serde_json::from_str(minimal).unwrap();
        assert_eq!(resp.status, "active");
        assert_eq!(resp.latest_srl_version, None);
        assert!(!resp.refresh_required);

        // Test full response with SRL info
        let full = r#"{"status": "active", "latest_srl_version": 8, "refresh_required": true}"#;
        let resp: HeartbeatResponse = serde_json::from_str(full).unwrap();
        assert_eq!(resp.status, "active");
        assert_eq!(resp.latest_srl_version, Some(8));
        assert!(resp.refresh_required);
    }

    #[test]
    fn test_srl_response_deserialization() {
        let json = r#"{"srl": "dGVzdA==", "version": 5}"#;
        let resp: SrlResponse = serde_json::from_str(json).unwrap();
        assert_eq!(resp.srl, "dGVzdA==");
        assert_eq!(resp.version, 5);
    }

    #[test]
    fn test_receipt_signing_payload_cbor_encoding() {
        // This test verifies the exact CBOR encoding of the signing payload
        // so we can compare with Go's encoding for cross-language compatibility.
        let payload = ReceiptSigningPayload {
            authorizer_id: "test",
            warrant_chain: &[1, 2, 3],
            action: "hello",
            outcome: "world",
            timestamp: 42,
            root_principal: None,
        };

        let mut buf = Vec::new();
        ciborium::into_writer(&payload, &mut buf).unwrap();

        let hex_str: String = buf.iter().map(|b| format!("{:02x}", b)).collect();
        println!("Rust CBOR hex: {}", hex_str);

        // Verify it's a map with string keys
        // First byte should be 0xa5 (map of 5 items, since root_principal is None)
        assert_eq!(buf[0], 0xa5, "Expected CBOR map of 5 items");

        // Verify the key "1" is a text string (0x61 = text of 1 byte, then 0x31 = '1')
        assert_eq!(buf[1], 0x61, "Key should be 1-byte text string");
        assert_eq!(buf[2], 0x31, "Key should be ASCII '1'");

        // Now test with Go's expected encoding:
        // Go string keys: a56131647465737461324301020361336568656c6c6f613465776f726c646135182a
        let go_hex = "a56131647465737461324301020361336568656c6c6f613465776f726c646135182a";
        assert_eq!(
            hex_str, go_hex,
            "Rust CBOR encoding must match Go encoding for signature verification"
        );
    }

    #[test]
    fn test_receipt_signing_payload_with_root_principal() {
        let payload = ReceiptSigningPayload {
            authorizer_id: "auth-1",
            warrant_chain: &[],
            action: "tool:read",
            outcome: "allow",
            timestamp: 1000,
            root_principal: Some("org:acme"),
        };

        let mut buf = Vec::new();
        ciborium::into_writer(&payload, &mut buf).unwrap();

        // Map of 6 items when root_principal is present
        assert_eq!(buf[0], 0xa6, "Expected CBOR map of 6 items");

        // Verify deterministic: re-encode produces identical bytes
        let mut buf2 = Vec::new();
        ciborium::into_writer(&payload, &mut buf2).unwrap();
        assert_eq!(buf, buf2, "CBOR encoding must be deterministic");
    }

    #[test]
    fn test_sign_event_produces_verifiable_signature() {
        let signing_key = SigningKey::generate();

        let event = AuthorizationEvent::allow(
            "".to_string(),
            "wrt-test-456".to_string(),
            "send_email".to_string(),
            2,
            Some("org:acme".to_string()),
            Some(base64::Engine::encode(
                &base64::engine::general_purpose::STANDARD,
                b"fake-chain",
            )),
            42,
            "req-sign-test".to_string(),
            None,
            None,
        );

        let signed = sign_event(&event, &signing_key, "authz-real-id");

        // Verify the authorizer_id was patched
        assert_eq!(signed.event.authorizer_id, "authz-real-id");

        // Decode signing_payload and verify the signature
        let payload_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signing_payload,
        )
        .expect("signing_payload must be valid base64");

        let sig_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signature,
        )
        .expect("signature must be valid base64");

        let sig = crate::crypto::Signature::from_bytes(
            sig_bytes
                .as_slice()
                .try_into()
                .expect("signature is 64 bytes"),
        )
        .expect("valid Ed25519 signature");

        assert!(
            signing_key
                .public_key()
                .verify(&payload_bytes, &sig)
                .is_ok(),
            "receipt signature must be verifiable with the signing key's public key"
        );
    }

    #[test]
    fn test_sign_event_different_keys_produce_different_signatures() {
        let key1 = SigningKey::generate();
        let key2 = SigningKey::generate();

        let event = AuthorizationEvent::allow(
            "".to_string(),
            "wrt-1".to_string(),
            "read_file".to_string(),
            0,
            None,
            None,
            10,
            "req-1".to_string(),
            None,
            None,
        );

        let signed1 = sign_event(&event, &key1, "auth-1");
        let signed2 = sign_event(&event, &key2, "auth-1");

        assert_ne!(
            signed1.signature, signed2.signature,
            "different keys must produce different signatures"
        );
    }

    #[test]
    fn test_sign_event_tampered_payload_fails_verification() {
        let signing_key = SigningKey::generate();

        let event = AuthorizationEvent::deny(
            "".to_string(),
            "wrt-deny".to_string(),
            "delete_db".to_string(),
            "constraint_violation".to_string(),
            Some("path".to_string()),
            1,
            None,
            None,
            99,
            "req-deny".to_string(),
            None,
            None,
        );

        let signed = sign_event(&event, &signing_key, "auth-id");

        // Decode signing payload and tamper with it
        let mut payload_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signing_payload,
        )
        .unwrap();
        if let Some(byte) = payload_bytes.last_mut() {
            *byte ^= 0xFF;
        }

        let sig_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signature,
        )
        .unwrap();
        let sig = crate::crypto::Signature::from_bytes(sig_bytes.as_slice().try_into().unwrap())
            .expect("valid Ed25519 signature bytes");

        assert!(
            signing_key
                .public_key()
                .verify(&payload_bytes, &sig)
                .is_err(),
            "tampered payload must fail signature verification"
        );
    }

    #[test]
    fn test_sign_event_with_approvals_in_event() {
        let signing_key = SigningKey::generate();

        let records = vec![
            ApprovalRecord {
                approver_key: "aabb".to_string(),
                external_id: "admin@corp.com".to_string(),
                approved_at: 1700000000,
                expires_at: 1700000300,
                request_hash: "ccdd".to_string(),
            },
            ApprovalRecord {
                approver_key: "eeff".to_string(),
                external_id: "security@corp.com".to_string(),
                approved_at: 1700000001,
                expires_at: 1700000301,
                request_hash: "ccdd".to_string(),
            },
        ];

        let event = AuthorizationEvent::allow(
            "".to_string(),
            "wrt-approval".to_string(),
            "execute_payment".to_string(),
            1,
            Some("org:bank".to_string()),
            None,
            50,
            "req-approval".to_string(),
            None,
            Some(records),
        );

        let signed = sign_event(&event, &signing_key, "auth-bank");

        // Approvals are present in the serialized event but NOT in the signing payload.
        // This is by design: individual approvals are self-signed by their approver keys.
        let json = serde_json::to_string(&signed).unwrap();
        assert!(json.contains("admin@corp.com"));
        assert!(json.contains("security@corp.com"));
        assert!(json.contains("approvals"));

        // The signing payload (CBOR of ReceiptSigningPayload) should NOT contain approval data
        let payload_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signing_payload,
        )
        .unwrap();
        let payload_hex: String = payload_bytes.iter().map(|b| format!("{:02x}", b)).collect();
        assert!(
            !payload_hex.contains(&hex::encode(b"admin@corp.com")),
            "approval data must not leak into the signing payload"
        );

        // Signature must still verify
        let sig_bytes = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            &signed.signature,
        )
        .unwrap();
        let sig = crate::crypto::Signature::from_bytes(sig_bytes.as_slice().try_into().unwrap())
            .expect("valid Ed25519 signature bytes");
        assert!(
            signing_key
                .public_key()
                .verify(&payload_bytes, &sig)
                .is_ok(),
            "signature must verify even with approvals in the event"
        );
    }

    #[test]
    fn test_approval_record_serialization_skip_none() {
        let event_with = AuthorizationEvent::allow(
            "a".into(),
            "w".into(),
            "t".into(),
            0,
            None,
            None,
            0,
            "r".into(),
            None,
            Some(vec![ApprovalRecord {
                approver_key: "key".into(),
                external_id: "id".into(),
                approved_at: 1,
                expires_at: 2,
                request_hash: "hash".into(),
            }]),
        );
        let event_without = AuthorizationEvent::allow(
            "a".into(),
            "w".into(),
            "t".into(),
            0,
            None,
            None,
            0,
            "r".into(),
            None,
            None,
        );

        let json_with = serde_json::to_string(&event_with).unwrap();
        let json_without = serde_json::to_string(&event_without).unwrap();

        assert!(json_with.contains("\"approvals\""));
        assert!(
            !json_without.contains("\"approvals\""),
            "None approvals must be omitted from JSON (skip_serializing_if)"
        );
    }
}