freenet 0.2.54

use std::sync::Arc;
use std::time::Duration;

use dashmap::DashMap;
use freenet_stdlib::prelude::{ClientResponse, UserInputRequest};
use tokio::sync::{broadcast, oneshot};

/// Timeout for user input prompts. After this duration, the request is auto-denied.
pub(crate) const USER_INPUT_TIMEOUT: Duration = Duration::from_secs(60);

/// Maximum message length for display. Prevents abuse from untrusted delegates.
const MAX_MESSAGE_LEN: usize = 2048;
/// Maximum button label length. 64 chars fits comfortably in a button.
const MAX_LABEL_LEN: usize = 64;
/// Maximum number of response buttons. Keeps the UI usable.
const MAX_LABELS: usize = 10;
/// Maximum stored length of a runtime-attested identity hash.
///
/// Defense-in-depth: today both `delegate_key` and `CallerIdentity::WebApp`
/// hashes come from runtime context that is bounded at the source (BLAKE3
/// hex / base58 contract id, both well under this limit). Capping at the
/// insertion point keeps the prompt store from holding arbitrarily large
/// strings if a future caller passes something larger.
///
/// `permission_prompts.rs` re-uses this same cap at the JSON / HTML
/// rendering boundary, so the two layers cannot disagree about the
/// maximum.
pub(crate) const MAX_IDENTITY_HASH_CHARS: usize = 256;

/// Runtime-attested identity of the entity that triggered a permission prompt.
///
/// This is a display-layer representation: it lives only between the executor
/// (which knows the typed `ContractInstanceId` / future `DelegateKey`) and the
/// permission-prompt UI. Stringification happens at the boundary so the UI
/// doesn't depend on stdlib types and the stored prompt can be serialised to
/// the dashboard JSON without any further conversion.
///
/// Variants reflect what `MessageOrigin` in freenet-stdlib can carry today.
/// A `Delegate(String)` variant is reserved for #3860 (extending
/// `MessageOrigin` to attest delegate-to-delegate callers); the prompt UI is
/// already shaped to render it without further changes.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CallerIdentity {
    /// No structured caller was recorded for this request. Today this covers
    /// both "no web app involved" and "delegate-to-delegate call" — those are
    /// not currently distinguishable at the API level (see #3860).
    None,
    /// The request came from a web app whose contract id was attested by
    /// `MessageOrigin::WebApp(..)`. The string is the `ContractInstanceId`
    /// in its canonical display form.
    WebApp(String),
}

/// Abstracts user prompting for delegate `RequestUserInput` messages.
///
/// Implementations receive the runtime-attested identity of the calling
/// delegate and (when known) of the originating web app. These identities are
/// rendered in the permission prompt UI so the user can see *who* is asking
/// and *which* app triggered the request. The identities MUST come from the
/// runtime context, never from the delegate's own message payload — a rogue
/// delegate must not be able to spoof another delegate's or app's identity.
pub trait UserInputPrompter: Send + Sync {
    fn prompt(
        &self,
        request: &UserInputRequest<'static>,
        delegate_key: &str,
        caller: CallerIdentity,
    ) -> impl std::future::Future<Output = Option<(usize, ClientResponse<'static>)>> + Send;
}

/// A pending permission request awaiting user response via the web dashboard.
pub(crate) struct PendingPrompt {
    pub message: String,
    pub labels: Vec<String>,
    pub delegate_key: String,
    pub caller: CallerIdentity,
    pub response_tx: oneshot::Sender<usize>,
}

/// Shared registry of pending permission prompts, keyed by 128-bit hex nonce.
pub(crate) type PendingPrompts = Arc<DashMap<String, PendingPrompt>>;

/// Global pending prompts registry, shared between the HTTP server (consumer)
/// and the DashboardPrompter (producer).
static PENDING_PROMPTS: std::sync::OnceLock<PendingPrompts> = std::sync::OnceLock::new();

/// Get or initialize the global pending prompts registry.
pub(crate) fn pending_prompts() -> PendingPrompts {
    PENDING_PROMPTS
        .get_or_init(|| Arc::new(DashMap::new()))
        .clone()
}

/// Maximum concurrent pending prompts to prevent memory exhaustion.
const MAX_PENDING_PROMPTS: usize = 32;

/// Snapshot of a prompt's display fields, sufficient for the SSE handler to
/// render an `Added` event without holding the DashMap entry. Cloned out of
/// the registry so the broadcast path doesn't pin the entry's lock.
#[derive(Clone, Debug)]
pub(crate) struct PromptSnapshot {
    pub nonce: String,
    pub message: String,
    pub labels: Vec<String>,
    pub delegate_key: String,
    pub caller: CallerIdentity,
}

/// Lifecycle event for a permission prompt. Consumed by the SSE endpoint to
/// push state changes to every open Freenet tab in real time. The polling
/// endpoint at `/permission/pending` is retained as a fallback and is not
/// driven by this stream.
#[derive(Clone, Debug)]
pub(crate) enum PromptEvent {
    Added(PromptSnapshot),
    Removed { nonce: String },
}

/// Broadcast capacity. Each lifecycle is two events (Added + Removed), and
/// MAX_PENDING_PROMPTS caps concurrent in-flight prompts at 32, so 128 leaves
/// healthy headroom even if a transient SSE subscriber lags briefly. On
/// `RecvError::Lagged`, the SSE handler resyncs from the DashMap snapshot.
const PROMPT_EVENT_CAPACITY: usize = 128;

/// Global lifecycle broadcast for permission prompts.
static PROMPT_EVENTS: std::sync::OnceLock<broadcast::Sender<PromptEvent>> =
    std::sync::OnceLock::new();

/// Get or initialize the global prompt-event broadcaster.
pub(crate) fn prompt_events() -> broadcast::Sender<PromptEvent> {
    PROMPT_EVENTS
        .get_or_init(|| broadcast::channel(PROMPT_EVENT_CAPACITY).0)
        .clone()
}

/// Fire a prompt lifecycle event. `Sender::send` returns `Err` when there
/// are no live subscribers; that's the common case when the shell page
/// isn't open yet, and the polling fallback covers it. Slow subscribers
/// see `Lagged` and resync from the DashMap.
///
/// Caps the nonce length on `Removed` so that even if an unusual code path
/// ever passes a client-supplied nonce here (today the only producers are
/// `DashboardPrompter` itself and the HTTP `respond` handler, which both
/// route through validated DashMap keys), the broadcast can't carry an
/// arbitrarily large string to every connected tab.
pub(crate) fn emit_prompt_event(event: PromptEvent) {
    let event = match event {
        PromptEvent::Removed { nonce } => PromptEvent::Removed {
            nonce: cap_identity_chars(&nonce),
        },
        other => other,
    };
    drop(prompt_events().send(event));
}

/// Opens the user's browser to the permission page on the local dashboard,
/// then waits for the user to click a button. The HTTP POST handler sends
/// the response back via a oneshot channel.
///
/// Works from systemd user services because the node serves HTTP (already
/// running) and the shell page's JS handles browser notification + opening.
pub struct DashboardPrompter {
    pending: PendingPrompts,
}

impl DashboardPrompter {
    pub fn new(pending: PendingPrompts) -> Self {
        Self { pending }
    }
}

impl UserInputPrompter for DashboardPrompter {
    async fn prompt(
        &self,
        request: &UserInputRequest<'static>,
        delegate_key: &str,
        caller: CallerIdentity,
    ) -> Option<(usize, ClientResponse<'static>)> {
        if request.responses.is_empty() {
            tracing::warn!("RequestUserInput has no response options");
            return None;
        }

        if self.pending.len() >= MAX_PENDING_PROMPTS {
            tracing::warn!(
                max = MAX_PENDING_PROMPTS,
                "Too many pending permission prompts, auto-denying"
            );
            return None;
        }

        let message = parse_message(request);
        let labels = parse_button_labels(request);

        // Generate a 128-bit cryptographic nonce for the permission URL
        let nonce = generate_nonce();

        let (tx, rx) = oneshot::channel();

        // Cap stored identity hashes at MAX_IDENTITY_HASH_CHARS at the
        // insertion boundary (not just at the JSON/HTML layer). Defense-in-
        // depth: real-world delegate keys are BLAKE3 hex (~64 chars) and
        // contract ids are base58 (~44 chars), both well under the cap.
        let stored_delegate_key = cap_identity_chars(delegate_key);
        let stored_caller = match caller {
            CallerIdentity::None => CallerIdentity::None,
            CallerIdentity::WebApp(hash) => CallerIdentity::WebApp(cap_identity_chars(&hash)),
        };

        self.pending.insert(
            nonce.clone(),
            PendingPrompt {
                message: message.clone(),
                labels: labels.clone(),
                delegate_key: stored_delegate_key.clone(),
                caller: stored_caller.clone(),
                response_tx: tx,
            },
        );

        // Fire the broadcast Added event AFTER the DashMap insert so any SSE
        // subscriber that wakes up on the event can immediately find the entry
        // if it falls back to a registry lookup.
        emit_prompt_event(PromptEvent::Added(PromptSnapshot {
            nonce: nonce.clone(),
            message,
            labels,
            delegate_key: stored_delegate_key,
            caller: stored_caller,
        }));

        // Log at debug, not info -- nonce is the sole auth token for this prompt
        tracing::debug!(
            request_id = request.request_id,
            "Permission prompt created, waiting for user response via dashboard"
        );

        // Wait for user response with timeout
        let result = tokio::time::timeout(USER_INPUT_TIMEOUT, rx).await;

        // Clean up if still pending. The HTTP `respond` handler removes the
        // entry on a real user click; this remove is a no-op in that case
        // and otherwise covers the timeout / dropped-channel cleanup paths.
        // Always emit Removed so subscribers can hide their overlay
        // regardless of which path retired the prompt; a duplicate Removed
        // (when both the HTTP handler and this cleanup fire) is harmless,
        // because the SSE client's hide is idempotent on nonce.
        let was_present = self.pending.remove(&nonce).is_some();
        if was_present {
            emit_prompt_event(PromptEvent::Removed {
                nonce: nonce.clone(),
            });
        }

        match result {
            Ok(Ok(idx)) if idx < request.responses.len() => {
                let response = request.responses[idx].clone().into_owned();
                Some((idx, response))
            }
            Ok(Ok(_)) => {
                tracing::warn!(nonce = %nonce, "Invalid response index from dashboard");
                None
            }
            Ok(Err(_)) => {
                tracing::debug!(nonce = %nonce, "Permission prompt channel closed");
                None
            }
            Err(_) => {
                tracing::warn!(nonce = %nonce, "Permission prompt timed out after 60s");
                None
            }
        }
    }
}

/// Truncate a runtime-attested identity hash to `MAX_IDENTITY_HASH_CHARS`
/// codepoints. Char-based so multi-byte content doesn't get split mid-grapheme.
fn cap_identity_chars(s: &str) -> String {
    s.chars().take(MAX_IDENTITY_HASH_CHARS).collect()
}

/// Generate a 128-bit cryptographic hex nonce.
fn generate_nonce() -> String {
    use crate::config::GlobalRng;
    let a = GlobalRng::random_u64();
    let b = GlobalRng::random_u64();
    format!("{a:016x}{b:016x}")
}

/// Extract a displayable message from `NotificationMessage` bytes.
pub(crate) fn parse_message(request: &UserInputRequest<'_>) -> String {
    let bytes = request.message.bytes();
    let raw = if let Ok(json_str) = serde_json::from_slice::<String>(bytes) {
        json_str
    } else {
        String::from_utf8(bytes.to_vec())
            .unwrap_or_else(|_| "A delegate is requesting permission.".to_string())
    };
    raw.chars()
        .take(MAX_MESSAGE_LEN)
        .filter(|c| !c.is_control() || *c == '\n')
        .collect()
}

/// Extract button labels from `ClientResponse` bytes as sanitized UTF-8 strings.
pub(crate) fn parse_button_labels(request: &UserInputRequest<'_>) -> Vec<String> {
    request
        .responses
        .iter()
        .take(MAX_LABELS)
        .enumerate()
        .map(|(i, r)| {
            let label =
                String::from_utf8((**r).to_vec()).unwrap_or_else(|_| format!("Option {}", i + 1));
            label
                .chars()
                .take(MAX_LABEL_LEN)
                .filter(|c| !c.is_control())
                .collect()
        })
        .collect()
}

/// Auto-approves by returning the first response. For testing only.
pub struct AutoApprovePrompter;

impl UserInputPrompter for AutoApprovePrompter {
    async fn prompt(
        &self,
        request: &UserInputRequest<'static>,
        _delegate_key: &str,
        _caller: CallerIdentity,
    ) -> Option<(usize, ClientResponse<'static>)> {
        request
            .responses
            .first()
            .map(|r| (0, r.clone().into_owned()))
    }
}

/// Always denies (returns None). For headless environments where no display
/// is available (e.g., gateway servers, CI).
#[allow(dead_code)]
pub struct AutoDenyPrompter;

impl UserInputPrompter for AutoDenyPrompter {
    async fn prompt(
        &self,
        _request: &UserInputRequest<'static>,
        _delegate_key: &str,
        _caller: CallerIdentity,
    ) -> Option<(usize, ClientResponse<'static>)> {
        None
    }
}

#[cfg(test)]
pub(crate) fn make_test_request(message: &str, responses: Vec<&str>) -> UserInputRequest<'static> {
    use freenet_stdlib::prelude::NotificationMessage;

    let msg = NotificationMessage::try_from(&serde_json::Value::String(message.to_string()))
        .expect("valid JSON");
    UserInputRequest {
        request_id: 1,
        message: msg,
        responses: responses
            .into_iter()
            .map(|r| ClientResponse::new(r.as_bytes().to_vec()))
            .collect(),
    }
}

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

    fn webapp(s: &str) -> CallerIdentity {
        CallerIdentity::WebApp(s.to_string())
    }

    #[tokio::test]
    async fn test_auto_approve_returns_first_response() {
        let req = make_test_request("Allow this?", vec!["Allow", "Deny"]);
        let result = AutoApprovePrompter
            .prompt(&req, "dkey", webapp("cid"))
            .await;
        let (idx, response) = result.unwrap();
        assert_eq!(idx, 0);
        assert_eq!(&*response, b"Allow");
    }

    #[tokio::test]
    async fn test_auto_approve_empty_responses() {
        let req = make_test_request("Allow this?", vec![]);
        let result = AutoApprovePrompter
            .prompt(&req, "dkey", webapp("cid"))
            .await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_auto_deny_always_returns_none() {
        let req = make_test_request("Allow this?", vec!["Allow", "Deny"]);
        let result = AutoDenyPrompter.prompt(&req, "dkey", webapp("cid")).await;
        assert!(result.is_none());
    }

    #[test]
    fn test_parse_button_labels() {
        let req = make_test_request("msg", vec!["Allow Once", "Always Allow", "Deny"]);
        let labels = parse_button_labels(&req);
        assert_eq!(labels, vec!["Allow Once", "Always Allow", "Deny"]);
    }

    #[test]
    fn test_parse_message_json_encoded() {
        let req = make_test_request("Hello world", vec![]);
        let msg = parse_message(&req);
        assert_eq!(msg, "Hello world");
    }

    #[test]
    fn test_parse_message_json_with_quotes() {
        use freenet_stdlib::prelude::NotificationMessage;
        let json_val = serde_json::Value::String("Test with \"quotes\"".to_string());
        let msg = NotificationMessage::try_from(&json_val).unwrap();
        let req = UserInputRequest {
            request_id: 1,
            message: msg,
            responses: vec![],
        };
        let parsed = parse_message(&req);
        assert_eq!(parsed, "Test with \"quotes\"");
    }

    #[tokio::test]
    async fn test_dashboard_prompter_max_pending() {
        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());

        for i in 0..MAX_PENDING_PROMPTS {
            let (tx, _rx) = oneshot::channel();
            pending.insert(
                format!("nonce_{i}"),
                PendingPrompt {
                    message: "test".to_string(),
                    labels: vec!["OK".to_string()],
                    delegate_key: String::new(),
                    caller: CallerIdentity::None,
                    response_tx: tx,
                },
            );
        }

        let req = make_test_request("Over limit", vec!["Allow"]);
        let result = prompter.prompt(&req, "dkey", webapp("cid")).await;
        assert!(result.is_none());
    }

    #[test]
    fn test_nonce_is_32_hex_chars() {
        let nonce = generate_nonce();
        assert_eq!(nonce.len(), 32);
        assert!(nonce.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn test_parse_message_strips_control_chars() {
        use freenet_stdlib::prelude::NotificationMessage;
        let json_val = serde_json::Value::String("Hello\x00\x07world".to_string());
        let msg = NotificationMessage::try_from(&json_val).unwrap();
        let req = UserInputRequest {
            request_id: 1,
            message: msg,
            responses: vec![],
        };
        let parsed = parse_message(&req);
        assert_eq!(parsed, "Helloworld");
    }

    #[test]
    fn test_parse_button_labels_invalid_utf8() {
        use freenet_stdlib::prelude::NotificationMessage;
        let req = UserInputRequest {
            request_id: 1,
            message: NotificationMessage::try_from(&serde_json::Value::String("msg".to_string()))
                .unwrap(),
            responses: vec![
                ClientResponse::new(b"Valid".to_vec()),
                ClientResponse::new(vec![0xFF, 0xFE]),
            ],
        };
        let labels = parse_button_labels(&req);
        assert_eq!(labels, vec!["Valid", "Option 2"]);
    }

    #[test]
    fn test_parse_message_raw_utf8() {
        use freenet_stdlib::prelude::NotificationMessage;
        let raw_msg =
            NotificationMessage::try_from(&serde_json::Value::String("Raw message".to_string()))
                .unwrap();
        let req = UserInputRequest {
            request_id: 1,
            message: raw_msg,
            responses: vec![],
        };
        let msg = parse_message(&req);
        assert_eq!(msg, "Raw message");
    }

    #[tokio::test]
    async fn test_auto_approve_with_three_responses() {
        let req = make_test_request("Allow?", vec!["Allow Once", "Always Allow", "Deny"]);
        let result = AutoApprovePrompter
            .prompt(&req, "dkey", webapp("cid"))
            .await;
        let (idx, response) = result.unwrap();
        assert_eq!(idx, 0);
        assert_eq!(&*response, b"Allow Once");
    }

    #[tokio::test]
    async fn test_auto_deny_with_multiple_responses() {
        let req = make_test_request("Allow?", vec!["Allow Once", "Always Allow", "Deny"]);
        let result = AutoDenyPrompter.prompt(&req, "dkey", webapp("cid")).await;
        assert!(result.is_none());
    }

    // Regression test for issue #3857: the dashboard prompter MUST populate
    // PendingPrompt's identity fields from the runtime-attested values passed
    // into prompt(), not hardcode them to "Unknown". Without this, the
    // permission overlay renders "Unknown" in the structured identity slots,
    // forcing the user to trust the delegate's self-authored message for
    // attribution, which a malicious delegate could spoof.
    #[tokio::test]
    async fn test_dashboard_prompter_populates_webapp_caller() {
        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());

        let req = make_test_request("Approve?", vec!["Allow", "Deny"]);
        let pending_clone = pending.clone();
        let handle = tokio::spawn(async move {
            prompter
                .prompt(&req, "DLGKEY123", webapp("CONTRACT456"))
                .await
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let entry = pending_clone
            .iter()
            .next()
            .expect("prompt should be registered");
        assert_eq!(entry.value().delegate_key, "DLGKEY123");
        assert_eq!(
            entry.value().caller,
            CallerIdentity::WebApp("CONTRACT456".to_string())
        );

        let nonce = entry.key().clone();
        drop(entry);
        let (_, prompt) = pending_clone.remove(&nonce).unwrap();
        prompt.response_tx.send(1).unwrap();
        let _ = handle.await.unwrap();
    }

    // When no web-app caller is attested (delegate-to-delegate, local client,
    // or any other non-WebApp invocation path), the prompt should record
    // CallerIdentity::None — distinct from WebApp(""), so the UI can render a
    // dedicated "No app caller" line rather than a blank field.
    #[tokio::test]
    async fn test_dashboard_prompter_records_none_caller() {
        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());

        let req = make_test_request("Approve?", vec!["Allow"]);
        let pending_clone = pending.clone();
        let handle =
            tokio::spawn(
                async move { prompter.prompt(&req, "DLGKEY", CallerIdentity::None).await },
            );

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let entry = pending_clone
            .iter()
            .next()
            .expect("prompt should be registered");
        assert_eq!(entry.value().delegate_key, "DLGKEY");
        assert_eq!(entry.value().caller, CallerIdentity::None);

        let nonce = entry.key().clone();
        drop(entry);
        let (_, prompt) = pending_clone.remove(&nonce).unwrap();
        prompt.response_tx.send(0).unwrap();
        let _ = handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_dashboard_prompter_happy_path() {
        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());

        let req = make_test_request("Allow signing?", vec!["Allow", "Deny"]);

        // Spawn the prompt in a task
        let pending_clone = pending.clone();
        let handle =
            tokio::spawn(async move { prompter.prompt(&req, "dkey", webapp("cid")).await });

        // Wait briefly for the prompt to be inserted
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Find the nonce and respond
        let nonce = pending_clone
            .iter()
            .next()
            .expect("should have a pending prompt")
            .key()
            .clone();

        let (_, prompt) = pending_clone.remove(&nonce).unwrap();
        prompt.response_tx.send(0).unwrap();

        let result = handle.await.unwrap();
        let (idx, response) = result.unwrap();
        assert_eq!(idx, 0);
        assert_eq!(&*response, b"Allow");
    }

    /// `DashboardPrompter::prompt` fires `PromptEvent::Added` after inserting
    /// the entry. The cleanup `Removed` is only fired by the prompter when
    /// the entry was still present at cleanup time. When an external party
    /// (the HTTP `/respond` handler in production) already removed the
    /// entry, the prompter's cleanup is a no-op and the external remover is
    /// responsible for emitting `Removed`. This test exercises that
    /// double-emit-avoidance path; the timeout-driven `Removed` path is
    /// covered by `test_prompt_timeout_emits_removed`.
    #[tokio::test]
    async fn test_prompt_added_emitted_and_external_remove_skips_cleanup_removed() {
        // Subscribe BEFORE spawning the prompter so we don't miss the
        // synchronous Added event.
        let mut rx = prompt_events().subscribe();

        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());
        let req = make_test_request("lifecycle?", vec!["Allow", "Deny"]);

        let pending_clone = pending.clone();
        let handle =
            tokio::spawn(async move { prompter.prompt(&req, "dkey-lc", webapp("cid-lc")).await });

        // Drain events until we see our specific Added (other tests in the
        // same process may fire concurrently against the global broadcast).
        let mut added_nonce: Option<String> = None;
        for _ in 0..50 {
            match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
                Ok(Ok(PromptEvent::Added(snap))) if snap.delegate_key == "dkey-lc" => {
                    added_nonce = Some(snap.nonce);
                    break;
                }
                Ok(_) => continue, // unrelated event from another test, keep draining
                Err(_) => break,   // timeout: no event in window
            }
        }
        let nonce = added_nonce.expect("PromptEvent::Added with our delegate_key");

        // Respond so the prompter exits its timeout wait and runs cleanup.
        let (_, prompt) = pending_clone.remove(&nonce).unwrap();
        prompt.response_tx.send(0).unwrap();
        let _ = handle.await.unwrap();

        // Expect a matching Removed for that nonce. May be interleaved with
        // unrelated events.
        let mut saw_removed = false;
        for _ in 0..50 {
            match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
                Ok(Ok(PromptEvent::Removed { nonce: n })) if n == nonce => {
                    saw_removed = true;
                    break;
                }
                Ok(_) => continue,
                Err(_) => break,
            }
        }
        // Note: in this test the response handler ran inside the
        // pending_clone.remove path (we removed the entry ourselves via
        // pending_clone.remove). DashboardPrompter's cleanup remove
        // returned None (was_present=false) so it did NOT emit Removed.
        // That path is only fired when the prompter's own cleanup
        // actually removes the entry (timeout / channel-dropped paths).
        // The HTTP `/respond` handler fires Removed in the success
        // path; that's covered by the SSE endpoint integration tests.
        assert!(
            !saw_removed,
            "this test exercises the manual-remove path; \
             Removed should fire only from the prompter's own cleanup \
             (timeout) or the HTTP respond handler (covered elsewhere)"
        );
    }

    /// When the prompter's own timeout cleanup runs, it must emit Removed
    /// so SSE subscribers dismiss the overlay.
    #[tokio::test(start_paused = true)]
    async fn test_prompt_timeout_emits_removed() {
        let mut rx = prompt_events().subscribe();

        let pending: PendingPrompts = Arc::new(DashMap::new());
        let prompter = DashboardPrompter::new(pending.clone());
        let req = make_test_request("timeout?", vec!["Allow"]);

        let handle = tokio::spawn(async move {
            prompter
                .prompt(&req, "dkey-timeout", webapp("cid-timeout"))
                .await
        });

        // Capture the nonce of our specific Added event.
        let mut our_nonce: Option<String> = None;
        for _ in 0..50 {
            tokio::task::yield_now().await;
            match rx.try_recv() {
                Ok(PromptEvent::Added(snap)) if snap.delegate_key == "dkey-timeout" => {
                    our_nonce = Some(snap.nonce);
                    break;
                }
                Ok(_) => continue,
                Err(tokio::sync::broadcast::error::TryRecvError::Empty) => {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(_) => break,
            }
        }
        let nonce = our_nonce.expect("Added event for the timeout test");

        // Advance virtual time past the prompt timeout.
        tokio::time::advance(USER_INPUT_TIMEOUT + Duration::from_secs(1)).await;
        let _ = handle.await.unwrap();

        let mut saw_removed = false;
        for _ in 0..50 {
            match rx.try_recv() {
                Ok(PromptEvent::Removed { nonce: n }) if n == nonce => {
                    saw_removed = true;
                    break;
                }
                Ok(_) => continue,
                Err(tokio::sync::broadcast::error::TryRecvError::Empty) => {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(_) => break,
            }
        }
        assert!(
            saw_removed,
            "prompter timeout must emit PromptEvent::Removed"
        );
    }
}