car-server-core 0.8.0

//! 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;
        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):
    /// - Approval has `client_id: Some(x)` → caller MUST be `x`.
    /// - 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.
    pub async fn resolve_approval(
        &self,
        caller_client_id: &str,
        req: ResolveHostApprovalRequest,
    ) -> Result<HostApprovalRequest, String> {
        let mut approvals = self.approvals.lock().await;
        let approval = approvals
            .get_mut(&req.approval_id)
            .ok_or_else(|| format!("unknown approval '{}'", req.approval_id))?;
        if let Some(owner) = approval.client_id.as_deref() {
            if owner != caller_client_id {
                return Err(format!(
                    "approval '{}' is owned by another session",
                    req.approval_id
                ));
            }
        }
        approval.status = HostApprovalStatus::Resolved;
        approval.resolution = Some(req.resolution);
        approval.resolved_at = Some(Utc::now());
        let resolved = approval.clone();
        drop(approvals);

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

    /// 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![],
                },
            )
            .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![],
                    },
                    "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![],
                    },
                    "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![],
                },
                "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_only_resolvable_by_creator() {
        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![],
                },
            )
            .await
            .unwrap();

        // Other client cannot squat the approval.
        let err = host
            .resolve_approval(
                "client-B",
                ResolveHostApprovalRequest {
                    approval_id: approval.id.clone(),
                    resolution: "approve".into(),
                },
            )
            .await
            .unwrap_err();
        assert!(err.contains("owned by another session"));

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

    #[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![],
                },
            )
            .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");
    }
}