car_server_core/

host.rs

//! 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;

#[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);
    }

    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)
    }

    /// 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> {
        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;
        }
    }
}

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::*;

    #[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"
        );
    }
}