//! Host-facing state for OS integrations.
//!
//! This module is intentionally UI-agnostic. Native shells such as a macOS menu
//! bar app, Windows tray app, or Linux status notifier can subscribe to these
//! events and render the same agent control surface.

use crate::session::WsChannel;
use car_proto::{
    CreateHostApprovalRequest, HostAgent, HostAgentStatus, HostApprovalRequest, HostApprovalStatus,
    HostEvent, RegisterHostAgentRequest, ResolveHostApprovalRequest, SetHostAgentStatusRequest,
};
use chrono::Utc;
use futures::SinkExt;
use serde_json::Value;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, Notify};
use tokio_tungstenite::tungstenite::Message;

const MAX_EVENTS: usize = 500;

/// Approvals accumulate within a daemon's uptime: a gate approval that
/// times out is left as a forensic `Pending` row, and no session/agent
/// disconnect reaps it (gate approvals are system-level, `client_id:
/// None`). With no age/count bound (events have `MAX_EVENTS`), the queue
/// grew unbounded over a long uptime. Drop approvals older than this that
/// nobody is actively awaiting. 24h is far longer than any gate `timeout`,
/// so an actively-parked waiter is never reaped.
const APPROVAL_TTL_SECS: i64 = 24 * 60 * 60;

#[derive(Default)]
pub struct HostState {
    agents: Mutex<HashMap<String, HostAgent>>,
    approvals: Mutex<HashMap<String, HostApprovalRequest>>,
    events: Mutex<VecDeque<HostEvent>>,
    subscribers: Mutex<HashMap<String, Arc<WsChannel>>>,
    /// Per-approval `Notify` so [`HostState::wait_for_resolution`]
    /// can park efficiently instead of polling the approvals map.
    /// Inserted when the gate creates an approval, removed when the
    /// resolution comes in (or when the wait drops it on timeout).
    /// Kept off the public surface because it's an implementation
    /// detail of the gate path — direct callers of `create_approval`
    /// don't need it.
    notifies: Mutex<HashMap<String, Arc<Notify>>>,
}

/// Outcome of a gated high-risk call.
///
/// Returned by [`HostState::request_and_wait_approval`]. Callers map
/// each variant to the JSON-RPC error / success they want to surface.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalOutcome {
    /// The user picked the explicit "approve" option (or whatever the
    /// caller declared as the approve label).
    Approved,
    /// The user picked any other option, or the resolution string
    /// didn't match the approve label.
    Denied,
    /// No resolution arrived inside the supplied timeout. Treated as
    /// deny by callers; the approval row is left in `Pending` so the
    /// UI can still display it for forensics.
    TimedOut,
}

impl HostState {
    pub fn new() -> Self {
        Self::default()
    }

    pub async fn subscribe(&self, client_id: &str, channel: Arc<WsChannel>) {
        self.subscribers
            .lock()
            .await
            .insert(client_id.to_string(), channel);
    }

    pub async fn unsubscribe(&self, client_id: &str) {
        self.subscribers.lock().await.remove(client_id);
    }

    /// True if `client_id` is currently registered for `host.event`
    /// notifications — i.e. it is a CarHost host-client (the management
    /// surface), not a bare agent connection. Used by the run-trace
    /// authorization gate (R16/KTD10): a host-client is entitled to
    /// subscribe to any managed agent's run trace, while a non-host
    /// connection may only read runs of the `agent_id` it owns.
    pub async fn is_subscribed(&self, client_id: &str) -> bool {
        self.subscribers.lock().await.contains_key(client_id)
    }

    pub async fn register_agent(
        &self,
        client_id: &str,
        req: RegisterHostAgentRequest,
    ) -> Result<HostAgent, String> {
        let id = req.id.unwrap_or_else(|| format!("agent-{}", short_id()));
        // ACL (audit 2026-05): if an agent with this id already
        // exists and was registered by a *different* session, refuse
        // — pre-flip a second client could overwrite the first
        // client's agent record.
        {
            let agents = self.agents.lock().await;
            if let Some(existing) = agents.get(&id) {
                match existing.session_id.as_deref() {
                    Some(owner) if owner != client_id => {
                        return Err(format!(
                            "agent '{id}' is owned by another session; \
                             unregister it from that session first"
                        ));
                    }
                    _ => {}
                }
            }
        }

        let agent = HostAgent {
            id: id.clone(),
            name: req.name,
            kind: req.kind,
            capabilities: req.capabilities,
            project: req.project,
            session_id: Some(client_id.to_string()),
            status: HostAgentStatus::Idle,
            current_task: None,
            pid: req.pid,
            display: req.display,
            updated_at: Utc::now(),
            metadata: req.metadata,
        };

        self.agents.lock().await.insert(id.clone(), agent.clone());
        self.record_event(
            "agent.registered",
            Some(id),
            format!("{} registered", agent.name),
            serde_json::to_value(&agent).map_err(|e| e.to_string())?,
        )
        .await;

        Ok(agent)
    }

    pub async fn unregister_agent(
        &self,
        caller_client_id: &str,
        agent_id: &str,
    ) -> Result<(), String> {
        // ACL (audit 2026-05): only the registering session can
        // unregister. Agents with no session_id (legacy / admin-
        // installed records) accept any caller.
        {
            let agents = self.agents.lock().await;
            if let Some(existing) = agents.get(agent_id) {
                if let Some(owner) = existing.session_id.as_deref() {
                    if owner != caller_client_id {
                        return Err(format!("agent '{agent_id}' is owned by another session"));
                    }
                }
            }
        }

        let removed = self.agents.lock().await.remove(agent_id);
        if removed.is_none() {
            return Err(format!("unknown agent '{}'", agent_id));
        }
        self.record_event(
            "agent.unregistered",
            Some(agent_id.to_string()),
            format!("{} unregistered", agent_id),
            Value::Null,
        )
        .await;
        // The agent that owned these is gone — its pending approvals
        // are no longer decidable (no one will read the resolution).
        // Auto-cancel so the queue stays in sync. car-releases#48.
        self.reap_agent_approvals(caller_client_id, agent_id).await;
        Ok(())
    }

    pub async fn set_status(
        &self,
        caller_client_id: &str,
        req: SetHostAgentStatusRequest,
    ) -> Result<HostAgent, String> {
        let mut agents = self.agents.lock().await;
        let agent = agents
            .get_mut(&req.agent_id)
            .ok_or_else(|| format!("unknown agent '{}'", req.agent_id))?;

        // ACL (audit 2026-05): only the agent's owning session can
        // change its status. Pre-flip a second client could mark
        // someone else's agent as Errored / WaitingForApproval /
        // anything else from outside the workflow.
        if let Some(owner) = agent.session_id.as_deref() {
            if owner != caller_client_id {
                return Err(format!(
                    "agent '{}' is owned by another session",
                    req.agent_id
                ));
            }
        }

        agent.status = req.status.clone();
        agent.current_task = req.current_task.clone();
        agent.updated_at = Utc::now();
        let updated = agent.clone();
        drop(agents);

        let message = req
            .message
            .unwrap_or_else(|| format!("{} is {:?}", updated.name, updated.status));
        self.record_event(
            "agent.status_changed",
            Some(updated.id.clone()),
            message,
            if req.payload.is_null() {
                serde_json::to_value(&updated).map_err(|e| e.to_string())?
            } else {
                req.payload
            },
        )
        .await;

        Ok(updated)
    }

    /// Drop approvals older than [`APPROVAL_TTL_SECS`] that have no live
    /// `request_and_wait_approval` waiter parked on them. Bounds the queue
    /// over a long daemon uptime (gate timeouts leave forensic `Pending`
    /// rows that nothing else reaps). A parked gate waiter — id present in
    /// `notifies` — is always retained; its timeout is minutes, far inside
    /// the TTL, so this only ever reaps abandoned/forensic rows, never a
    /// live one. Called on the growth path (`create_approval`).
    ///
    /// Unlike [`reap_approvals`], this drops rows *silently* — no
    /// `approval.resolved`/`expired` event. That's intentional: these are
    /// abandoned rows >24h old, and the Approvals surface re-fetches via
    /// `host.approvals` (poll), so a removal event isn't needed to keep it
    /// in sync.
    async fn prune_stale_approvals(&self) {
        let cutoff = Utc::now() - chrono::Duration::seconds(APPROVAL_TTL_SECS);
        let awaited: std::collections::HashSet<String> =
            self.notifies.lock().await.keys().cloned().collect();
        self.approvals
            .lock()
            .await
            .retain(|id, a| a.created_at >= cutoff || awaited.contains(id));
    }

    /// Create an approval owned by `caller_client_id`. Pass `None`
    /// to mark the approval *system-level* — the high-risk-method
    /// approval gate uses this so the local UI session (a different
    /// session than the one whose dispatch is parking) can resolve
    /// it. Audit 2026-05: prior to this caller arg, `create_approval`
    /// had no notion of ownership and `resolve_approval` was open
    /// to any caller, allowing cross-session approval squatting.
    pub async fn create_approval(
        &self,
        caller_client_id: Option<&str>,
        req: CreateHostApprovalRequest,
    ) -> Result<HostApprovalRequest, String> {
        // Bound the queue on the growth path before adding another row.
        self.prune_stale_approvals().await;
        let approval = HostApprovalRequest {
            id: format!("approval-{}", short_id()),
            agent_id: req.agent_id,
            client_id: caller_client_id.map(|s| s.to_string()),
            action: req.action,
            details: req.details,
            options: if req.options.is_empty() {
                vec!["approve".to_string(), "deny".to_string()]
            } else {
                req.options
            },
            status: HostApprovalStatus::Pending,
            created_at: Utc::now(),
            resolved_at: None,
            resolution: None,
        };

        self.approvals
            .lock()
            .await
            .insert(approval.id.clone(), approval.clone());
        self.record_event(
            "approval.requested",
            approval.agent_id.clone(),
            format!("Approval requested: {}", approval.action),
            serde_json::to_value(&approval).map_err(|e| e.to_string())?,
        )
        .await;
        Ok(approval)
    }

    /// Resolve an approval. ACL rules (audit 2026-05, fan-out added 2026-05-15):
    /// - Approval has `client_id: None` (system-level, e.g. raised
    ///   by the high-risk-method gate) → any authed caller may
    ///   resolve, since the gate's whole point is that *the user*
    ///   acks it via whichever session their UI happens to use.
    /// - Approval has `client_id: Some(x)` AND caller IS `x` →
    ///   resolve directly, fire `approval.resolved`.
    /// - Approval has `client_id: Some(x)` AND caller is a DIFFERENT
    ///   session AND owner is still subscribed → fan-out: record an
    ///   `approval.resolve_requested` event that the owning agent
    ///   hooks to call `resolve_approval` on its own session. The
    ///   approval row stays Pending until the owner completes it.
    ///   Caller gets back the pending approval (status: Pending).
    ///   This unblocks UIs like CarHost that surface every approval
    ///   in `host.approvals` — including ones agents pushed via
    ///   `host.request_approval` — without breaking the squat-
    ///   prevention property: only the OWNER ever mutates the row,
    ///   the non-owning caller just signals intent.
    /// - Cross-session AND owner is NOT subscribed → return error
    ///   identifying the disconnected owner. Caller's UI knows the
    ///   resolution can't land right now and surfaces accordingly.
    pub async fn resolve_approval(
        &self,
        caller_client_id: &str,
        req: ResolveHostApprovalRequest,
    ) -> Result<HostApprovalRequest, String> {
        // First pass: snapshot the approval and check ownership.
        // We don't hold the approvals lock across the subscribers
        // lock — both are mutexes and arbitrary lock-order would
        // invite a deadlock.
        let (owner_opt, pending_snapshot) = {
            let mut approvals = self.approvals.lock().await;
            let approval = approvals
                .get_mut(&req.approval_id)
                .ok_or_else(|| format!("unknown approval '{}'", req.approval_id))?;
            (approval.client_id.clone(), approval.clone())
        };

        if let Some(owner) = owner_opt {
            if owner != caller_client_id {
                // Cross-session resolve. If the owning session is
                // currently subscribed, fan out and let them act on
                // their own session. If not, fail explicitly so the
                // caller's UI can tell the user.
                let owner_subscribed =
                    self.subscribers.lock().await.contains_key(&owner);
                if !owner_subscribed {
                    return Err(format!(
                        "approval '{}' is owned by session '{}' which is not currently connected",
                        req.approval_id, owner,
                    ));
                }
                self.record_event(
                    "approval.resolve_requested",
                    pending_snapshot.agent_id.clone(),
                    format!(
                        "Resolution requested for {}: {}",
                        pending_snapshot.action, req.resolution
                    ),
                    serde_json::json!({
                        "approval_id": req.approval_id,
                        "resolution": req.resolution,
                        "requesting_client_id": caller_client_id,
                        "owner_client_id": owner,
                    }),
                )
                .await;
                // Return the still-Pending row. The owner's resolve
                // call will fire `approval.resolved`, which the
                // caller's subscription picks up to update its UI.
                return Ok(pending_snapshot);
            }
        }

        // Same-session OR system-level. Resolve directly.
        let resolved = {
            let mut approvals = self.approvals.lock().await;
            let approval = approvals
                .get_mut(&req.approval_id)
                .ok_or_else(|| format!("approval '{}' vanished mid-resolve", req.approval_id))?;
            approval.status = HostApprovalStatus::Resolved;
            approval.resolution = Some(req.resolution);
            approval.resolved_at = Some(Utc::now());
            approval.clone()
        };

        // Wake any gate task parked on this approval. Take the Notify
        // out of the map (it's one-shot) before notifying so the wait
        // task can drop its Arc cleanly. `notify_one` is safe even if
        // no one is waiting yet — the Notify holds the permit.
        if let Some(notify) = self.notifies.lock().await.remove(&resolved.id) {
            notify.notify_one();
        }

        self.record_event(
            "approval.resolved",
            resolved.agent_id.clone(),
            format!("Approval resolved: {}", resolved.action),
            serde_json::to_value(&resolved).map_err(|e| e.to_string())?,
        )
        .await;
        Ok(resolved)
    }

    /// Auto-resolve every still-`Pending` approval matched by
    /// `should_reap`, with `resolution = "agent_gone"`, and fan out
    /// `approval.resolved` so subscribed shells (CarHost) drop them
    /// from their queue. Returns how many were reaped.
    ///
    /// Lock order matches [`resolve_approval`]: mutate under the
    /// `approvals` lock, collect snapshots, release it, *then* wake
    /// gate waiters and fire events (`record_event` takes the events
    /// + subscribers locks).
    async fn reap_approvals<F>(&self, should_reap: F) -> usize
    where
        F: Fn(&HostApprovalRequest) -> bool,
    {
        let reaped: Vec<HostApprovalRequest> = {
            let mut approvals = self.approvals.lock().await;
            let mut out = Vec::new();
            for approval in approvals.values_mut() {
                if approval.status == HostApprovalStatus::Pending && should_reap(approval) {
                    approval.status = HostApprovalStatus::Resolved;
                    approval.resolution = Some("agent_gone".to_string());
                    approval.resolved_at = Some(Utc::now());
                    out.push(approval.clone());
                }
            }
            out
        };

        for approval in &reaped {
            // Defensive: gate-raised approvals are system-level
            // (client_id None) so they won't match the session/agent
            // predicates, but if one ever did, don't leave a parked
            // gate task hung.
            if let Some(notify) = self.notifies.lock().await.remove(&approval.id) {
                notify.notify_one();
            }
            self.record_event(
                "approval.resolved",
                approval.agent_id.clone(),
                format!("Approval auto-cancelled (agent gone): {}", approval.action),
                serde_json::to_value(approval).unwrap_or(Value::Null),
            )
            .await;
        }
        reaped.len()
    }

    /// Reap a disconnected session's pending approvals. Called on WS
    /// close — covers graceful unregister+close, hard crash (TCP
    /// reset), and ping timeout in one place. Only approvals *owned*
    /// by this session (`client_id == Some(client_id)`) are touched;
    /// system-level gate approvals (`client_id: None`) are left for
    /// the user to act on. car-releases#48.
    pub async fn reap_session_approvals(&self, client_id: &str) -> usize {
        self.reap_approvals(|a| a.client_id.as_deref() == Some(client_id))
            .await
    }

    /// Reap a specific agent's pending approvals when it unregisters
    /// while its session stays open (agent restarts under a new id on
    /// the same WS). Scoped to approvals this caller's session owns so
    /// it can't cancel another session's — or a system gate's — work.
    pub async fn reap_agent_approvals(&self, caller_client_id: &str, agent_id: &str) -> usize {
        self.reap_approvals(|a| {
            a.agent_id.as_deref() == Some(agent_id)
                && a.client_id.as_deref() == Some(caller_client_id)
        })
        .await
    }

    /// Create an approval and block until the user resolves it (or
    /// `timeout` elapses).
    ///
    /// Used by the high-risk-method gate in the WS dispatcher to make
    /// the human a load-bearing participant in actions like
    /// `automation.run_applescript`, `messages.send`, etc. The
    /// outcome maps as follows:
    ///
    /// - resolution string equals `approve_label` → [`ApprovalOutcome::Approved`]
    /// - any other resolution string → [`ApprovalOutcome::Denied`]
    /// - timeout fires before resolution → [`ApprovalOutcome::TimedOut`]
    ///
    /// Subscribers receive the standard `approval.requested` event
    /// the moment the approval is created; the local HTML UI and
    /// any other host shell can render approve/deny buttons that
    /// call `host.resolve_approval`.
    ///
    /// On timeout, the approval row is left in `Pending` on purpose
    /// — the UI still shows it (with a "expired" hint the renderer
    /// can derive from `created_at`) and the gate path returns
    /// `TimedOut` so the caller surfaces a clear error.
    pub async fn request_and_wait_approval(
        &self,
        req: CreateHostApprovalRequest,
        approve_label: &str,
        timeout: Duration,
    ) -> Result<ApprovalOutcome, String> {
        // Gate-raised approvals are system-level (caller_client_id =
        // None) so the local UI session — which is a *different*
        // session from the one whose dispatch is parking — is
        // permitted to resolve them. Per-session ACL would deadlock
        // the UX otherwise.
        let approval = self.create_approval(None, req).await?;
        let approval_id = approval.id.clone();

        // Register the wakeup channel BEFORE we sleep. resolve_approval
        // pulls this notify out of the map and signals it; if it's
        // missing (because resolve raced ahead), we re-check the
        // approvals map below before waiting.
        let notify = Arc::new(Notify::new());
        {
            let mut map = self.notifies.lock().await;
            map.insert(approval_id.clone(), notify.clone());
        }

        // Defensive re-check: if resolve_approval landed between
        // create_approval and the notify insert, the wakeup is gone
        // but the approvals map already has the resolution. Pull it
        // out and short-circuit.
        if let Some(resolved) = self.approvals.lock().await.get(&approval_id).cloned() {
            if matches!(resolved.status, HostApprovalStatus::Resolved) {
                self.notifies.lock().await.remove(&approval_id);
                return Ok(classify_resolution(&resolved, approve_label));
            }
        }

        // Park until either the notify fires or the timeout elapses.
        // We hold the Arc<Notify>; drop ours on the way out so the
        // map slot is free either way (resolve removes it; timeout
        // also removes it below).
        let woken = tokio::time::timeout(timeout, notify.notified()).await;
        if woken.is_err() {
            self.notifies.lock().await.remove(&approval_id);
            return Ok(ApprovalOutcome::TimedOut);
        }

        let resolved = self
            .approvals
            .lock()
            .await
            .get(&approval_id)
            .cloned()
            .ok_or_else(|| format!("approval '{}' vanished after notify", approval_id))?;
        Ok(classify_resolution(&resolved, approve_label))
    }

    pub async fn agents(&self) -> Vec<HostAgent> {
        let mut agents: Vec<_> = self.agents.lock().await.values().cloned().collect();
        agents.sort_by(|a, b| a.name.cmp(&b.name).then(a.id.cmp(&b.id)));
        agents
    }

    pub async fn approvals(&self) -> Vec<HostApprovalRequest> {
        let mut approvals: Vec<_> = self.approvals.lock().await.values().cloned().collect();
        approvals.sort_by(|a, b| b.created_at.cmp(&a.created_at));
        approvals
    }

    pub async fn events(&self, limit: usize) -> Vec<HostEvent> {
        self.events
            .lock()
            .await
            .iter()
            .rev()
            .take(limit)
            .cloned()
            .collect()
    }

    pub async fn record_event(
        &self,
        kind: impl Into<String>,
        agent_id: Option<String>,
        message: impl Into<String>,
        payload: Value,
    ) -> HostEvent {
        let event = HostEvent {
            id: format!("event-{}", short_id()),
            timestamp: Utc::now(),
            kind: kind.into(),
            agent_id,
            message: message.into(),
            payload,
        };

        {
            let mut events = self.events.lock().await;
            events.push_back(event.clone());
            while events.len() > MAX_EVENTS {
                events.pop_front();
            }
        }

        self.broadcast_event(&event).await;
        event
    }

    async fn broadcast_event(&self, event: &HostEvent) {
        let subscribers: Vec<Arc<WsChannel>> =
            self.subscribers.lock().await.values().cloned().collect();

        let Ok(json) = serde_json::to_string(&serde_json::json!({
            "jsonrpc": "2.0",
            "method": "host.event",
            "params": event,
        })) else {
            return;
        };

        for channel in subscribers {
            let _ = channel
                .write
                .lock()
                .await
                .send(Message::Text(json.clone().into()))
                .await;
        }
    }
}

/// Capacity of each run-trace subscriber's bounded channel (agent run
/// tracing, U4). One slot per buffered `runs.trace.event` between the
/// producer (the recorder / lifecycle path) and the per-subscriber drain
/// task. Sized generously: a healthy CarHost drains far faster than turns
/// are produced, so the buffer only fills if a socket genuinely wedges —
/// at which point `try_send` drops the event rather than blocking the
/// producer (invariant #2). 256 absorbs a burst of fast turns without
/// dropping under normal load.
pub const RUN_TRACE_CHANNEL_CAP: usize = 256;

/// One live `runs.trace.event` subscriber — the producer side of the
/// bounded channel whose drain task writes frames to the subscriber's
/// WebSocket (agent run tracing, U4).
///
/// Keyed in [`crate::session::ServerState::run_subscribers`] by
/// `(run_id, host_client_id)` so two CarHost windows on the same run are
/// independent streams (the explicit fanout the single-subscriber-
/// per-method notification registry can't provide), and so disconnect
/// cleanup can drop exactly this connection's subscriptions.
///
/// The producer holds ONLY the `tx` and calls [`RunTraceSubscriber::push`]
/// — a non-blocking `try_send`. It never touches the WS socket, so a slow
/// CarHost can never stall the recorder, the `runs` lock, or any other
/// in-flight RPC (invariant #2). The dedicated drain task owns the socket
/// write.
pub struct RunTraceSubscriber {
    /// The connection that subscribed — its WS `client_id`.
    pub host_client_id: String,
    /// Non-blocking producer handle into the drain task's channel.
    tx: tokio::sync::mpsc::Sender<car_proto::RunTraceEvent>,
}

impl RunTraceSubscriber {
    /// Spawn a drain task bound to `channel` and return the producer-side
    /// subscriber handle. The drain task serializes each
    /// `runs.trace.event` to a JSON-RPC notification frame and writes it
    /// to the subscriber's WS; it exits when the `tx` is dropped (the
    /// subscriber is removed on unsubscribe / disconnect) — at which point
    /// the channel closes and `recv()` returns `None`.
    pub fn spawn(host_client_id: String, channel: Arc<WsChannel>) -> Self {
        let (tx, mut rx) =
            tokio::sync::mpsc::channel::<car_proto::RunTraceEvent>(RUN_TRACE_CHANNEL_CAP);
        tokio::spawn(async move {
            while let Some(event) = rx.recv().await {
                let Ok(json) = serde_json::to_string(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "method": "runs.trace.event",
                    "params": event,
                })) else {
                    continue;
                };
                // A wedged socket should not hang the drain task forever —
                // bound the write so a dead connection's task exits and
                // frees the channel rather than parking on a full TCP
                // buffer. On any write error the connection is gone; stop
                // draining (the subscriber is reaped on disconnect).
                let mut guard = channel.write.lock().await;
                let send = tokio::time::timeout(
                    Duration::from_secs(10),
                    guard.send(Message::Text(json.into())),
                )
                .await;
                drop(guard);
                match send {
                    Ok(Ok(())) => {}
                    // Timed out or errored — the socket is unusable. Exit
                    // the drain loop; the producer's `try_send` will start
                    // dropping (bounded) and the subscriber is cleaned up
                    // on disconnect.
                    _ => break,
                }
            }
        });
        Self { host_client_id, tx }
    }

    /// Non-blocking push of one event onto the drain channel. Returns
    /// `false` when the channel is full (a wedged subscriber) — the event
    /// is dropped rather than blocking the producer (invariant #2). The
    /// producer treats the drop as best-effort: the client detects the
    /// resulting cursor gap and re-subscribes to backfill (R8).
    pub fn push(&self, event: car_proto::RunTraceEvent) -> bool {
        self.tx.try_send(event).is_ok()
    }

    #[cfg(test)]
    pub fn is_closed(&self) -> bool {
        self.tx.is_closed()
    }
}

fn short_id() -> String {
    uuid::Uuid::new_v4().simple().to_string()[..12].to_string()
}

fn classify_resolution(approval: &HostApprovalRequest, approve_label: &str) -> ApprovalOutcome {
    match approval.resolution.as_deref() {
        Some(r) if r == approve_label => ApprovalOutcome::Approved,
        _ => ApprovalOutcome::Denied,
    }
}

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

    /// Build a `WsChannel` whose writes land in a `futures::mpsc`
    /// receiver the test can drain to observe the drain task's frames.
    /// `futures::channel::mpsc::UnboundedSender` already implements
    /// `Sink<Message>`; we map its `SendError` into the tungstenite error
    /// the `WsSink` alias expects.
    fn channel_with_capture() -> (
        Arc<WsChannel>,
        futures::channel::mpsc::UnboundedReceiver<Message>,
    ) {
        use futures::sink::SinkExt;
        let (tx, rx) = futures::channel::mpsc::unbounded::<Message>();
        let sink: crate::session::WsSink = Box::pin(
            tx.sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed),
        );
        let channel = Arc::new(WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(0),
        });
        (channel, rx)
    }

    fn approval_row(id: &str, age: chrono::Duration) -> HostApprovalRequest {
        HostApprovalRequest {
            id: id.to_string(),
            agent_id: None,
            client_id: None,
            action: "test".into(),
            details: Value::Null,
            options: vec!["approve".into(), "deny".into()],
            status: HostApprovalStatus::Pending,
            created_at: Utc::now() - age,
            resolved_at: None,
            resolution: None,
        }
    }

    #[tokio::test]
    async fn create_approval_prunes_stale_rows_but_keeps_recent_and_awaited() {
        let host = HostState::new();
        let day = chrono::Duration::days(1);

        // Seed three pre-existing rows directly: a stale abandoned one, a
        // stale one that a gate waiter is still parked on, and a recent one.
        {
            let mut map = host.approvals.lock().await;
            map.insert("stale".into(), approval_row("stale", day * 2));
            map.insert("stale-awaited".into(), approval_row("stale-awaited", day * 2));
            map.insert("recent".into(), approval_row("recent", chrono::Duration::minutes(5)));
        }
        // A live waiter protects "stale-awaited" regardless of its age.
        host.notifies
            .lock()
            .await
            .insert("stale-awaited".into(), Arc::new(Notify::new()));

        // Adding any approval runs the prune on the growth path.
        let fresh = host
            .create_approval(None, CreateHostApprovalRequest {
                agent_id: None,
                action: "new".into(),
                details: Value::Null,
                options: vec![],
                system_level: true,
            })
            .await
            .unwrap();

        let ids: std::collections::HashSet<String> =
            host.approvals().await.into_iter().map(|a| a.id).collect();
        assert!(!ids.contains("stale"), "stale, unawaited row reaped");
        assert!(ids.contains("stale-awaited"), "live waiter protects its row");
        assert!(ids.contains("recent"), "recent row kept");
        assert!(ids.contains(&fresh.id), "the new approval is present");
    }

    #[tokio::test]
    async fn run_trace_subscriber_drains_event_to_socket() {
        let (channel, mut rx) = channel_with_capture();
        let sub = RunTraceSubscriber::spawn("host-1".to_string(), channel);
        let event = car_proto::RunTraceEvent {
            run_id: "run-1".to_string(),
            agent_id: "agent-a".to_string(),
            record: car_proto::RunRecord::Started(car_proto::RunStarted {
                run_id: "run-1".to_string(),
                agent_id: "agent-a".to_string(),
                intent: "go".to_string(),
                outcome_description: None,
                started_at: chrono::Utc::now(),
            }),
            cursor: 0,
            status: car_proto::RunLiveStatus::InProgress,
        };
        assert!(sub.push(event), "push onto a fresh channel succeeds");

        // The drain task serializes the event to a runs.trace.event frame.
        use futures::StreamExt;
        let frame = tokio::time::timeout(Duration::from_secs(2), rx.next())
            .await
            .expect("drain task wrote within the deadline")
            .expect("a frame");
        let text = match frame {
            Message::Text(t) => t.to_string(),
            other => panic!("expected a text frame, got {other:?}"),
        };
        let json: serde_json::Value = serde_json::from_str(&text).unwrap();
        assert_eq!(json["method"], "runs.trace.event");
        assert_eq!(json["params"]["run_id"], "run-1");
        assert_eq!(json["params"]["status"], "in_progress");
    }

    #[tokio::test]
    async fn dropping_subscriber_ends_its_drain_task() {
        let (channel, _rx) = channel_with_capture();
        let sub = RunTraceSubscriber::spawn("host-1".to_string(), channel);
        assert!(!sub.is_closed());
        // Dropping the subscriber drops its sender; the drain task's
        // recv() returns None and the task exits. We can only assert the
        // sender side here (the task is detached), but is_closed flips
        // once the receiver is gone — which happens when the task ends.
        drop(sub);
        // No panic / hang is the assertion; the spawned task tears down.
    }

    #[tokio::test]
    async fn host_is_subscribed_reflects_membership() {
        let host = HostState::new();
        let ch = Arc::new(WsChannel::test_stub());
        assert!(!host.is_subscribed("host-1").await);
        host.subscribe("host-1", ch).await;
        assert!(host.is_subscribed("host-1").await);
        host.unsubscribe("host-1").await;
        assert!(!host.is_subscribed("host-1").await);
    }

    #[tokio::test]
    async fn host_tracks_agents_events_and_approvals() {
        let host = HostState::new();

        let agent = host
            .register_agent(
                "client-1",
                RegisterHostAgentRequest {
                    id: Some("agent-1".to_string()),
                    name: "Researcher".to_string(),
                    kind: "builtin".to_string(),
                    capabilities: vec!["search".to_string()],
                    project: Some("/tmp/project".to_string()),
                    pid: None,
                    display: car_proto::HostAgentDisplay {
                        label: Some("Research Lead".to_string()),
                        icon: Some("magnifying-glass".to_string()),
                        accent: Some("#0a84ff".to_string()),
                    },
                    metadata: Value::Null,
                },
            )
            .await
            .expect("register agent");

        assert_eq!(agent.status, HostAgentStatus::Idle);
        assert_eq!(agent.display.label.as_deref(), Some("Research Lead"));
        assert_eq!(agent.display.icon.as_deref(), Some("magnifying-glass"));
        assert_eq!(agent.display.accent.as_deref(), Some("#0a84ff"));
        assert_eq!(host.agents().await.len(), 1);

        let updated = host
            .set_status(
                "client-1",
                SetHostAgentStatusRequest {
                    agent_id: "agent-1".to_string(),
                    status: HostAgentStatus::Running,
                    current_task: Some("Collect facts".to_string()),
                    message: None,
                    payload: Value::Null,
                },
            )
            .await
            .expect("set status");

        assert_eq!(updated.status, HostAgentStatus::Running);
        assert_eq!(updated.current_task.as_deref(), Some("Collect facts"));

        let approval = host
            .create_approval(
                Some("client-1"),
                CreateHostApprovalRequest {
                    agent_id: Some("agent-1".to_string()),
                    action: "Run tests".to_string(),
                    details: serde_json::json!({ "command": "cargo test" }),
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .expect("create approval");

        assert_eq!(approval.options, vec!["approve", "deny"]);
        assert_eq!(approval.status, HostApprovalStatus::Pending);

        let resolved = host
            .resolve_approval(
                "client-1",
                ResolveHostApprovalRequest {
                    approval_id: approval.id,
                    resolution: "approve".to_string(),
                },
            )
            .await
            .expect("resolve approval");

        assert_eq!(resolved.status, HostApprovalStatus::Resolved);
        assert_eq!(resolved.resolution.as_deref(), Some("approve"));
        assert!(host.events(10).await.len() >= 4);
    }

    #[tokio::test]
    async fn request_and_wait_returns_approved_when_user_approves() {
        let host = Arc::new(HostState::new());
        let host2 = host.clone();

        // Launch the gate in the background; resolve after a small
        // delay to mirror the realistic UI round-trip.
        let waiter = tokio::spawn(async move {
            host2
                .request_and_wait_approval(
                    CreateHostApprovalRequest {
                        agent_id: None,
                        action: "automation.run_applescript".into(),
                        details: serde_json::json!({}),
                        options: vec![],
                    system_level: false,
                    },
                    "approve",
                    Duration::from_secs(2),
                )
                .await
                .expect("gate ran")
        });

        // Find the pending approval and resolve it. The gate raises
        // approvals as system-level (client_id None), so any caller
        // can resolve — using "ui-session" here mirrors how the
        // local UI session would.
        tokio::time::sleep(Duration::from_millis(20)).await;
        let pending = host.approvals().await;
        assert_eq!(pending.len(), 1, "exactly one pending approval");
        assert!(
            pending[0].client_id.is_none(),
            "gate-raised approvals must be system-level"
        );
        host.resolve_approval(
            "ui-session",
            ResolveHostApprovalRequest {
                approval_id: pending[0].id.clone(),
                resolution: "approve".into(),
            },
        )
        .await
        .unwrap();

        let outcome = waiter.await.unwrap();
        assert_eq!(outcome, ApprovalOutcome::Approved);
    }

    #[tokio::test]
    async fn request_and_wait_returns_denied_on_other_resolution() {
        let host = Arc::new(HostState::new());
        let host2 = host.clone();
        let waiter = tokio::spawn(async move {
            host2
                .request_and_wait_approval(
                    CreateHostApprovalRequest {
                        agent_id: None,
                        action: "messages.send".into(),
                        details: serde_json::json!({}),
                        options: vec![],
                    system_level: false,
                    },
                    "approve",
                    Duration::from_secs(2),
                )
                .await
                .unwrap()
        });
        tokio::time::sleep(Duration::from_millis(20)).await;
        let pending = host.approvals().await;
        host.resolve_approval(
            "ui-session",
            ResolveHostApprovalRequest {
                approval_id: pending[0].id.clone(),
                resolution: "deny".into(),
            },
        )
        .await
        .unwrap();
        assert_eq!(waiter.await.unwrap(), ApprovalOutcome::Denied);
    }

    #[tokio::test]
    async fn request_and_wait_times_out_when_no_resolution() {
        let host = HostState::new();
        let outcome = host
            .request_and_wait_approval(
                CreateHostApprovalRequest {
                    agent_id: None,
                    action: "vision.ocr".into(),
                    details: serde_json::json!({}),
                    options: vec![],
                    system_level: false,
                },
                "approve",
                Duration::from_millis(50),
            )
            .await
            .unwrap();
        assert_eq!(outcome, ApprovalOutcome::TimedOut);
        // Approval row stays in Pending so the UI keeps a record.
        let pending = host.approvals().await;
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].status, HostApprovalStatus::Pending);
    }

    // -- ACL regression tests (audit 2026-05) --

    fn make_register_request(name: &str) -> RegisterHostAgentRequest {
        RegisterHostAgentRequest {
            id: Some(name.into()),
            name: name.into(),
            kind: "test".into(),
            capabilities: vec![],
            project: None,
            pid: None,
            display: car_proto::HostAgentDisplay {
                label: None,
                icon: None,
                accent: None,
            },
            metadata: Value::Null,
        }
    }

    #[tokio::test]
    async fn set_status_rejects_non_owning_session() {
        let host = HostState::new();
        host.register_agent("client-A", make_register_request("worker"))
            .await
            .unwrap();

        // Client B tries to mutate Client A's agent.
        let err = host
            .set_status(
                "client-B",
                SetHostAgentStatusRequest {
                    agent_id: "worker".into(),
                    status: HostAgentStatus::Errored,
                    current_task: None,
                    message: None,
                    payload: Value::Null,
                },
            )
            .await
            .unwrap_err();
        assert!(
            err.contains("owned by another session"),
            "unexpected rejection message: {err}"
        );

        // Owner still works.
        host.set_status(
            "client-A",
            SetHostAgentStatusRequest {
                agent_id: "worker".into(),
                status: HostAgentStatus::Running,
                current_task: None,
                message: None,
                payload: Value::Null,
            },
        )
        .await
        .expect("owner can mutate");
    }

    #[tokio::test]
    async fn unregister_rejects_non_owning_session() {
        let host = HostState::new();
        host.register_agent("client-A", make_register_request("worker"))
            .await
            .unwrap();

        let err = host
            .unregister_agent("client-B", "worker")
            .await
            .unwrap_err();
        assert!(err.contains("owned by another session"));
        assert_eq!(host.agents().await.len(), 1, "agent must survive");

        host.unregister_agent("client-A", "worker")
            .await
            .expect("owner can unregister");
        assert_eq!(host.agents().await.len(), 0);
    }

    #[tokio::test]
    async fn register_refuses_to_overwrite_other_sessions_agent() {
        let host = HostState::new();
        host.register_agent("client-A", make_register_request("worker"))
            .await
            .unwrap();

        let err = host
            .register_agent("client-B", make_register_request("worker"))
            .await
            .unwrap_err();
        assert!(
            err.contains("owned by another session"),
            "unexpected message: {err}"
        );
    }

    #[tokio::test]
    async fn manual_approval_cross_session_with_disconnected_owner_errors() {
        // client-A creates an approval but is NOT subscribed. client-B
        // tries to resolve it; the fan-out path needs the owner
        // connected, so this returns a clear "not connected" error
        // (rather than silently no-op'ing or pretending success).
        let host = HostState::new();
        let approval = host
            .create_approval(
                Some("client-A"),
                CreateHostApprovalRequest {
                    agent_id: None,
                    action: "manual.action".into(),
                    details: serde_json::json!({}),
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .unwrap();

        let err = host
            .resolve_approval(
                "client-B",
                ResolveHostApprovalRequest {
                    approval_id: approval.id.clone(),
                    resolution: "approve".into(),
                },
            )
            .await
            .unwrap_err();
        assert!(
            err.contains("not currently connected"),
            "expected disconnected-owner message, got: {err}",
        );
        // Approval stays Pending — non-owner failed call must not
        // mutate state.
        let still = host
            .approvals()
            .await
            .into_iter()
            .find(|a| a.id == approval.id)
            .expect("approval survives failed resolve");
        assert_eq!(still.status, HostApprovalStatus::Pending);

        // Owner can still resolve directly (same-session fast path).
        host.resolve_approval(
            "client-A",
            ResolveHostApprovalRequest {
                approval_id: approval.id,
                resolution: "deny".into(),
            },
        )
        .await
        .expect("owner can resolve");
    }

    #[tokio::test]
    async fn manual_approval_cross_session_with_subscribed_owner_fans_out() {
        // client-A creates an approval AND subscribes. client-B tries
        // to resolve. The fan-out path records an
        // `approval.resolve_requested` event the owner is expected to
        // hook to perform the resolve on its own session. The approval
        // row itself stays Pending until the owner acts — non-owner
        // never mutates state (audit-2026-05 squat-prevention holds).
        let host = HostState::new();

        // Subscribe client-A. We only exercise the subscribers-map
        // membership check, so a WsChannel::test_stub() (drain sink)
        // is enough — nothing is actually written to the channel in
        // this test.
        host.subscribe("client-A", Arc::new(WsChannel::test_stub()))
            .await;

        let approval = host
            .create_approval(
                Some("client-A"),
                CreateHostApprovalRequest {
                    agent_id: Some("worker".into()),
                    action: "Send email to bob@example.com".into(),
                    details: serde_json::json!({"kind": "email_reply"}),
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .unwrap();

        // client-B asks to resolve — should fan-out, not mutate.
        let returned = host
            .resolve_approval(
                "client-B",
                ResolveHostApprovalRequest {
                    approval_id: approval.id.clone(),
                    resolution: "deny".into(),
                },
            )
            .await
            .expect("cross-session resolve fans out without error");
        assert_eq!(
            returned.status,
            HostApprovalStatus::Pending,
            "returned row must stay Pending — only the owner mutates",
        );

        // An `approval.resolve_requested` event landed in the log.
        let events = host.events(10).await;
        let resolve_req = events
            .iter()
            .find(|e| e.kind == "approval.resolve_requested")
            .expect("resolve_requested event recorded");
        assert_eq!(
            resolve_req.payload.get("approval_id").and_then(|v| v.as_str()),
            Some(approval.id.as_str()),
        );
        assert_eq!(
            resolve_req.payload.get("resolution").and_then(|v| v.as_str()),
            Some("deny"),
        );
        assert_eq!(
            resolve_req.payload.get("owner_client_id").and_then(|v| v.as_str()),
            Some("client-A"),
        );
        assert_eq!(
            resolve_req
                .payload
                .get("requesting_client_id")
                .and_then(|v| v.as_str()),
            Some("client-B"),
        );

        // Approval row in the map still Pending — non-owner did NOT
        // squat the state.
        let still = host
            .approvals()
            .await
            .into_iter()
            .find(|a| a.id == approval.id)
            .unwrap();
        assert_eq!(still.status, HostApprovalStatus::Pending);

        // Owner picks up the event and resolves on its own session —
        // this is the real fan-out completion path. After this fires,
        // the approval flips to Resolved and `approval.resolved`
        // lands in the event log for the original caller's UI.
        let resolved = host
            .resolve_approval(
                "client-A",
                ResolveHostApprovalRequest {
                    approval_id: approval.id.clone(),
                    resolution: "deny".into(),
                },
            )
            .await
            .expect("owner resolves on own session");
        assert_eq!(resolved.status, HostApprovalStatus::Resolved);
        assert_eq!(resolved.resolution.as_deref(), Some("deny"));
    }


    #[tokio::test]
    async fn system_approval_resolvable_by_any_session() {
        // The high-risk-method gate raises approvals with caller =
        // None so the local UI session — a different WS connection
        // from the one whose dispatch is parking — can resolve them.
        let host = HostState::new();
        let approval = host
            .create_approval(
                None,
                CreateHostApprovalRequest {
                    agent_id: None,
                    action: "ws.method:automation.run_applescript".into(),
                    details: serde_json::json!({}),
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .unwrap();
        assert!(
            approval.client_id.is_none(),
            "system approval must carry no owner"
        );

        host.resolve_approval(
            "ui-session",
            ResolveHostApprovalRequest {
                approval_id: approval.id,
                resolution: "approve".into(),
            },
        )
        .await
        .expect("any session can resolve a system approval");
    }

    // -- car-releases#48: stale-approval reaping --

    #[tokio::test]
    async fn unregister_agent_reaps_its_pending_approvals() {
        let host = HostState::new();
        host.register_agent("client-1", make_register_request("agent-1"))
            .await
            .unwrap();
        let approval = host
            .create_approval(
                Some("client-1"),
                CreateHostApprovalRequest {
                    agent_id: Some("agent-1".to_string()),
                    action: "Wire $1M".to_string(),
                    details: Value::Null,
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .unwrap();
        assert_eq!(approval.status, HostApprovalStatus::Pending);

        host.unregister_agent("client-1", "agent-1").await.unwrap();

        let pending: Vec<_> = host
            .approvals()
            .await
            .into_iter()
            .filter(|a| a.status == HostApprovalStatus::Pending)
            .collect();
        assert!(pending.is_empty(), "agent's approval must be auto-cancelled");
        let all = host.approvals().await;
        let reaped = all.iter().find(|a| a.id == approval.id).unwrap();
        assert_eq!(reaped.status, HostApprovalStatus::Resolved);
        assert_eq!(reaped.resolution.as_deref(), Some("agent_gone"));
    }

    #[tokio::test]
    async fn session_disconnect_reaps_owned_but_not_system_approvals() {
        let host = HostState::new();
        // Owned by the disconnecting session.
        let owned = host
            .create_approval(
                Some("client-9"),
                CreateHostApprovalRequest {
                    agent_id: Some("agent-x".to_string()),
                    action: "owned".to_string(),
                    details: Value::Null,
                    options: vec![],
                    system_level: false,
                },
            )
            .await
            .unwrap();
        // System-level gate approval (client_id None) — must survive.
        let system = host
            .create_approval(
                None,
                CreateHostApprovalRequest {
                    agent_id: None,
                    action: "system gate".to_string(),
                    details: Value::Null,
                    options: vec![],
                    system_level: true,
                },
            )
            .await
            .unwrap();

        let n = host.reap_session_approvals("client-9").await;
        assert_eq!(n, 1, "only the session-owned approval is reaped");

        let all = host.approvals().await;
        let owned_now = all.iter().find(|a| a.id == owned.id).unwrap();
        let system_now = all.iter().find(|a| a.id == system.id).unwrap();
        assert_eq!(owned_now.status, HostApprovalStatus::Resolved);
        assert_eq!(owned_now.resolution.as_deref(), Some("agent_gone"));
        assert_eq!(
            system_now.status,
            HostApprovalStatus::Pending,
            "system-level gate approvals must outlive a client disconnect"
        );
    }
}
