car-server-core 0.51.0

//! Operator attention for a browser that is blocked waiting on a human
//! sign-in — the `host.event` half of the browser drawer.
//!
//! ## Why this is not a `browser.view.event`
//!
//! The drawer's own fanout ([`crate::browser_view`]) is the right place for
//! everything a person watching the drawer needs. It is the wrong place for
//! this one thing, because the whole problem is that **nobody is watching**:
//! `browser_await_signin` blocks the agent for up to 1800s, and with the
//! drawer closed — or another conversation in view — the only trace of it is
//! a strip in a surface that is not on screen. `browser.view.*` requires a
//! live subscription per conversation, so by construction it cannot reach an
//! operator who has not already opened the thing they need to be told to
//! open.
//!
//! `host.event` can. CarHost subscribes to it once, on connect, for the whole
//! daemon (`host.subscribe`), independent of any drawer. That is the entire
//! reason these two kinds live on that channel:
//!
//! - [`BROWSER_SIGNIN_NEEDED`] — an agent's browser is blocked on a human.
//! - [`BROWSER_SIGNIN_RESOLVED`] — that wait ended (signed in, handed back,
//!   timed out/run ended with nobody engaged, host gone, grace expired, or
//!   agent process died).
//!
//! The twin is mandatory, not a nicety: a badge that never clears is worse
//! than no badge, because the operator learns to ignore it.
//!
//! ## The transition rule
//!
//! Both emitters fire on the pending-sign-in STATE TRANSITION, never on an
//! apply or a push:
//!
//! - `None -> Some` emits [`BROWSER_SIGNIN_NEEDED`]
//! - `Some -> None` emits [`BROWSER_SIGNIN_RESOLVED`]
//! - `Some(a) -> Some(b)` emits [`BROWSER_SIGNIN_NEEDED`] when the prompt changed
//! - `Some(a) -> Some(a)` and `None -> None` emit nothing
//!
//! One exception, and it is a route change rather than a state change: a
//! relay producer backs several per-turn views, and when the view REPORTING a
//! wait retires while the browser is still blocked, the daemon emits
//! [`BROWSER_SIGNIN_RESOLVED`] for the retiring key immediately followed by
//! [`BROWSER_SIGNIN_NEEDED`] for a surviving one. The wait moved; it did not
//! end. Ordinary per-key handling on the host gets this right, and the
//! alternative — staying silent — leaves a badge pointing at a conversation
//! the operator can no longer reach.
//!
//! That last line is load-bearing. `browser_producer`'s `presentation_pump`
//! republishes the presentation on every change and its 10-second sweep
//! re-registers known conversations, so a "fire on every apply" emitter would
//! notify the operator every few seconds for one sign-in.
//!
//! ## Two producers, one contract
//!
//! - **In-daemon** (`assistant_start`): [`BrowserTools`] owns the reducer in
//!   this process, so the transition is observed at its single mutator,
//!   `apply_control`, through a [`SignInAttention`] installed at the
//!   construction site that knows the conversation key.
//! - **Relay** (`car do --serve`, CAR Chat): the reducer lives in the agent
//!   process, and the daemon already receives every presentation change over
//!   `browser.producer.presentation` unconditionally. So the daemon DERIVES
//!   the transition once on the process-owned `RelayProducer`, even when that
//!   browser backs several per-turn views. No new agent -> daemon method: the
//!   state is already on the wire.
//!
//! A browser is served by exactly one producer path, which keeps a single
//! sign-in from being announced twice: a relay producer owns the comparison,
//! while only a local `BrowserTools` has a local attention binding.
//!
//! [`BrowserTools`]: crate::assistant::browser_tools::BrowserTools

use std::sync::Arc;

use async_trait::async_trait;
use serde_json::json;

use crate::host::HostState;

pub use car_proto::BrowserSignInSnapshot;

/// `host.event` kind: an agent's browser is blocked waiting for a human to
/// sign in. Payload: `{ conversation_id, standing_session, message }`.
pub const BROWSER_SIGNIN_NEEDED: &str = "browser.signin_needed";

/// `host.event` kind: the sign-in wait announced by [`BROWSER_SIGNIN_NEEDED`]
/// ended. Payload: `{ conversation_id, standing_session }`.
pub const BROWSER_SIGNIN_RESOLVED: &str = "browser.signin_resolved";

/// Where a sign-in wait becomes an operator-facing notification.
///
/// Same shape, and the same reasoning, as
/// [`crate::assistant::browser_tools::HostConnectivity`]: a capability the
/// browser side needs but cannot build for itself, injected by whichever
/// producer constructs it, with `None` meaning "no way to tell anyone" —
/// every test, and any embedder with no daemon behind it.
///
/// `conversation_id` is the view key: `None` is the standing session, which
/// the wire renders as `""` (the host's own `browserViewKey` convention).
#[async_trait]
pub trait SignInAttention: Send + Sync {
    /// A browser just started waiting on a human sign-in. `message` is the
    /// plain-words prompt the drawer's strip shows, e.g. "Sign in at
    /// https://example.com/login".
    async fn signin_needed(&self, conversation_id: Option<&str>, message: &str);

    /// That wait ended — signed in, handed back, timed out/run ended with
    /// nobody engaged, host gone, grace expired, or the producer died.
    async fn signin_resolved(&self, conversation_id: Option<&str>);
}

/// The production [`SignInAttention`]: broadcast on the daemon's always-on
/// `host.event` channel ([`HostState::record_event`]).
///
/// Holds a strong `Arc<HostState>` rather than a `Weak`, unlike
/// `DaemonHostConnectivity`: `HostState` holds no reference back to the view
/// registry or to any `BrowserTools`, so there is no cycle to break here.
pub struct HostSignInAttention {
    host: Arc<HostState>,
}

impl HostSignInAttention {
    pub fn new(host: Arc<HostState>) -> Self {
        Self { host }
    }
}

/// The wire rendering of a view key. `None` (the standing session) is the
/// empty string, matching the host's `browserViewKey(_:)` convention so the
/// two ends agree without a special case on either side.
fn wire_conversation_id(conversation_id: Option<&str>) -> &str {
    conversation_id.unwrap_or("")
}

#[async_trait]
impl SignInAttention for HostSignInAttention {
    async fn signin_needed(&self, conversation_id: Option<&str>, message: &str) {
        // Only the prompt the TOOL composed ("Sign in at <url>") travels. It
        // is model-facing text the agent already holds, so it leaks nothing.
        // Nothing read off the page ever goes in here — no title, no URL from
        // `get_current_url`, no form contents, no cookies — because a sign-in
        // window is exactly when the privacy blackout is up.
        self.host
            .record_event(
                BROWSER_SIGNIN_NEEDED,
                None,
                message.to_string(),
                json!({
                    "conversation_id": wire_conversation_id(conversation_id),
                    "standing_session": conversation_id.is_none(),
                    "message": message,
                }),
            )
            .await;
    }

    async fn signin_resolved(&self, conversation_id: Option<&str>) {
        self.host
            .record_event(
                BROWSER_SIGNIN_RESOLVED,
                None,
                "The browser sign-in wait ended".to_string(),
                json!({
                    "conversation_id": wire_conversation_id(conversation_id),
                    "standing_session": conversation_id.is_none(),
                }),
            )
            .await;
    }
}

/// Emit whichever of the two kinds this pending-sign-in transition calls for,
/// and nothing at all when it is not a transition.
///
/// The ONE place the rule lives, shared by both producers so the local and
/// relay paths cannot drift apart. `before`/`after` are the pending sign-in's
/// plain-words message on each side of the change.
pub async fn notify_signin_transition(
    attention: &Arc<dyn SignInAttention>,
    conversation_id: Option<&str>,
    before: Option<&str>,
    after: Option<&str>,
) {
    match (before, after) {
        (None, Some(message)) => attention.signin_needed(conversation_id, message).await,
        (Some(_), None) => attention.signin_resolved(conversation_id).await,
        (Some(before), Some(after)) if before != after => {
            attention.signin_needed(conversation_id, after).await
        }
        // An unchanged pending prompt is the republish/resync case;
        // `None -> None` is every other event. Neither is news.
        _ => {}
    }
}

/// One recorded call: `(kind, conversation_id, message)`.
#[cfg(test)]
pub(crate) type RecordedAttention = (String, Option<String>, Option<String>);

/// A [`SignInAttention`] that records what it was told.
///
/// Lives outside `mod tests` because BOTH producer suites assert against it —
/// `assistant::browser_tools` for the local path and `browser_view` for the
/// relay twin — and the point of this type is that the two paths are held to
/// the identical contract.
#[cfg(test)]
#[derive(Default)]
pub(crate) struct RecordingAttention {
    calls: std::sync::Mutex<Vec<RecordedAttention>>,
}

#[cfg(test)]
impl RecordingAttention {
    pub(crate) fn calls(&self) -> Vec<RecordedAttention> {
        self.calls.lock().unwrap().clone()
    }

    pub(crate) fn kinds(&self) -> Vec<String> {
        self.calls().into_iter().map(|(kind, _, _)| kind).collect()
    }
}

#[cfg(test)]
#[async_trait]
impl SignInAttention for RecordingAttention {
    async fn signin_needed(&self, conversation_id: Option<&str>, message: &str) {
        self.calls.lock().unwrap().push((
            BROWSER_SIGNIN_NEEDED.to_string(),
            conversation_id.map(str::to_string),
            Some(message.to_string()),
        ));
    }

    async fn signin_resolved(&self, conversation_id: Option<&str>) {
        self.calls.lock().unwrap().push((
            BROWSER_SIGNIN_RESOLVED.to_string(),
            conversation_id.map(str::to_string),
            None,
        ));
    }
}

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

    #[tokio::test]
    async fn only_the_two_real_transitions_notify() {
        let recorder = Arc::new(RecordingAttention::default());
        let attention: Arc<dyn SignInAttention> = recorder.clone();

        notify_signin_transition(&attention, Some("conv-1"), None, Some("Sign in at x")).await;
        // A changed request is news: it can point at a different sign-in URL.
        notify_signin_transition(
            &attention,
            Some("conv-1"),
            Some("Sign in at x"),
            Some("other"),
        )
        .await;
        // An unchanged request is only a republish/resync.
        notify_signin_transition(&attention, Some("conv-1"), Some("other"), Some("other")).await;
        notify_signin_transition(&attention, Some("conv-1"), None, None).await;
        notify_signin_transition(&attention, Some("conv-1"), Some("Sign in at x"), None).await;

        assert_eq!(
            recorder.kinds(),
            vec![
                BROWSER_SIGNIN_NEEDED,
                BROWSER_SIGNIN_NEEDED,
                BROWSER_SIGNIN_RESOLVED
            ],
            "a start, a changed prompt, and a resolution are news"
        );
    }

    #[tokio::test]
    async fn the_host_emitter_puts_both_kinds_on_the_host_event_channel() {
        let host = Arc::new(HostState::new());
        let attention: Arc<dyn SignInAttention> =
            Arc::new(HostSignInAttention::new(Arc::clone(&host)));

        attention
            .signin_needed(Some("conv-1"), "Sign in at https://example.com/login")
            .await;
        attention.signin_resolved(Some("conv-1")).await;
        // The standing session's key is the empty string on the wire.
        attention.signin_needed(None, "Sign in to continue").await;

        let events = host.events(10).await;
        let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).rev().collect();
        assert_eq!(
            kinds,
            vec![
                BROWSER_SIGNIN_NEEDED,
                BROWSER_SIGNIN_RESOLVED,
                BROWSER_SIGNIN_NEEDED
            ]
        );

        let needed = events
            .iter()
            .rev()
            .find(|e| e.kind == BROWSER_SIGNIN_NEEDED)
            .expect("the needed event is recorded");
        assert_eq!(needed.payload["conversation_id"], "conv-1");
        assert_eq!(needed.payload["standing_session"], false);
        assert_eq!(
            needed.payload["message"],
            "Sign in at https://example.com/login"
        );
        assert_eq!(needed.message, "Sign in at https://example.com/login");

        let standing = events
            .iter()
            .find(|e| e.kind == BROWSER_SIGNIN_NEEDED)
            .expect("the standing-session event is recorded");
        assert_eq!(standing.payload["conversation_id"], "");
        assert_eq!(standing.payload["standing_session"], true);
    }
}