/
GreyCat
/
openhuman
Обзор
Документация
Войти
/
GreyCat
/
openhuman
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
main
src/core/socketio.rs
1 537 строк
67 KB
Steven Enamakel
feat(agent): track session token costs (#4136)
26 июн 2026, 01:04
Не верифицирован
26 июн 2026, 01:04
dbd1c0c
Код
Авторство
О чём код?
use serde::Deserialize; use serde::Serialize; use serde_json::json; use socketioxide::extract::{Data, SocketRef, TryData}; use socketioxide::SocketIo; /// Marker stored in [`SocketRef::extensions`] once a connection has presented a /// bearer token that matches the active per-process RPC token. /// /// Event handlers consult this before forwarding attacker-controllable input /// into the JSON-RPC dispatcher or the web-chat orchestrator: an unauthenticated /// socket that never picked up the marker is allowed to receive broadcast-style /// events (read-only) but cannot trigger executable work. #[derive(Clone, Copy, Debug)] struct AuthedConnection; /// Connection-time payload the client passes via Socket.IO's `auth` field. /// /// Browsers do not let `EventSource` / `WebSocket` clients attach custom /// headers, so the handshake `auth` map is the only header-equivalent slot /// available for our per-process bearer. The socket-IO Node/JS clients all /// surface `io(url, { auth: { token: "<hex>" } })` for this. #[derive(Debug, Default, Deserialize)] struct HandshakeAuth { #[serde(default)] token: Option<String>, } /// Origins the local core trusts at the Socket.IO handshake. /// /// The document origin of the CEF-served app shell is platform-dependent: /// /// | Platform | Scheme | Host | /// |----------|--------|------| /// | macOS / iOS (native scheme) | `tauri` | `localhost` | /// | Windows (CEF http custom protocol) | `http` | `tauri.localhost` | /// | Linux / older Windows builds | `https` | `tauri.localhost` | /// | Vite dev (`pnpm dev:app`, `pnpm dev`) | `http` | `localhost` / `127.0.0.1` / `[::1]` | /// /// The handshake `Origin` header is stamped by the webview with whichever /// of these shapes loaded the page — it is **not** the destination URL the /// socket is connecting to. We match the parsed host against the allowlist /// so all four shapes pass regardless of scheme, while `starts_with` decoys /// like `http://localhost.attacker.example` are still rejected (parser /// returns a different `host_str`). /// /// A missing `Origin` header is treated as a native (non-browser) client /// and accepted — only the cross-origin browser-page case is the targeted /// bad actor here. pub(crate) fn origin_is_allowed(origin: Option<&str>) -> bool { let Some(origin) = origin else { return true; // native clients (CLI, Tauri shell) — no Origin header }; let origin = origin.trim(); if origin.is_empty() || origin == "null" { return false; } // Parse the URL and compare the host EXACTLY against the loopback + // tauri.localhost allowlist. The earlier scheme-literal short-circuit // (`tauri://localhost` / `https://tauri.localhost`) missed // `http://tauri.localhost`, which is the document origin CEF stamps // on Windows — every flavour of the Tauri webview shell now goes // through the same host check. let Ok(parsed) = url::Url::parse(origin) else { return false; }; // `url::Url::host_str` returns IPv6 hosts with surrounding brackets, // hostnames bare. Accept both shapes. matches!( parsed.host_str(), Some("localhost" | "127.0.0.1" | "::1" | "[::1]" | "tauri.localhost") ) } /// True when `socket` finished the handshake with a valid bearer token. fn socket_is_authed(socket: &SocketRef) -> bool { socket.extensions.get::<AuthedConnection>().is_some() } /// Best-effort disconnect. Called when we discover an unauthenticated socket /// inside an event handler — the connect path already disconnects the bad /// origins / wrong tokens, so this is purely a defense-in-depth path. fn drop_unauthed(socket: &SocketRef, reason: &'static str) { log::warn!( "[socketio] dropping unauthenticated socket id={} reason={}", socket.id, reason ); let _ = socket.clone().disconnect(); } /// Standard event payload for the web channel transport. /// /// This structure defines the data sent to Socket.IO clients for various /// chat-related events, such as message delivery, tool execution, and errors. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct WebChannelEvent { /// The event name (e.g., `chat_message`, `tool_call`). pub event: String, /// Unique identifier for the Socket.IO client. pub client_id: String, /// Identifier for the specific chat thread. pub thread_id: String, /// Unique identifier for the individual request/turn. pub request_id: String, /// The full text of the assistant's response (sent on completion). #[serde(skip_serializing_if = "Option::is_none")] pub full_response: Option<String>, /// A partial message segment or an error description. #[serde(skip_serializing_if = "Option::is_none")] pub message: Option<String>, /// Type of error, if the event represents a failure. #[serde(skip_serializing_if = "Option::is_none")] pub error_type: Option<String>, /// Structured rate-limit / error metadata produced by /// `classify_inference_error` (issue #2606). All four fields are /// additive — older FE clients that only read `message`/`error_type` /// keep working; new clients can read these to render countdown, /// retry-button, and fallback-CTA UI without regexing the message. /// /// Where the limit originated: /// `"provider"` | `"openhuman_budget"` | `"agent_loop"` /// | `"openhuman_billing"` | `"transport"` | `"config"`. #[serde(skip_serializing_if = "Option::is_none")] pub error_source: Option<String>, /// Whether the same prompt can be retried in this same thread. /// `false` for non-retryable business 429s, auth, model_unavailable, /// context_overflow, and billing exhaustion. #[serde(skip_serializing_if = "Option::is_none")] pub error_retryable: Option<bool>, /// Milliseconds to wait before retrying, as supplied by the upstream /// `Retry-After:` / `retry_after:` header. `None` when the upstream /// didn't supply one or the error class has no retry-after concept. #[serde(skip_serializing_if = "Option::is_none")] pub error_retry_after_ms: Option<u64>, /// Provider name extracted from `"<provider> API error (...)"` /// envelopes. `None` for non-provider errors (OpenHuman budget cap, /// agent loop) and for transport failures without a provider prefix. #[serde(skip_serializing_if = "Option::is_none")] pub error_provider: Option<String>, /// `Some(false)` once the reliable-provider chain has exhausted /// every configured `model_fallbacks` entry. `None` means "unknown /// — FE should not promise a fallback". #[serde(skip_serializing_if = "Option::is_none")] pub error_fallback_available: Option<bool>, /// Name of the tool being called. #[serde(skip_serializing_if = "Option::is_none")] pub tool_name: Option<String>, /// ID of the skill owning the tool. #[serde(skip_serializing_if = "Option::is_none")] pub skill_id: Option<String>, /// Arguments passed to the tool. #[serde(skip_serializing_if = "Option::is_none")] pub args: Option<serde_json::Value>, /// The raw output from the tool execution. #[serde(skip_serializing_if = "Option::is_none")] pub output: Option<String>, /// Whether the tool execution or request was successful. #[serde(skip_serializing_if = "Option::is_none")] pub success: Option<bool>, /// The current iteration/round number in a tool-call loop. #[serde(skip_serializing_if = "Option::is_none")] pub round: Option<u32>, /// Emoji reaction the assistant wants to add to the user's message. #[serde(skip_serializing_if = "Option::is_none")] pub reaction_emoji: Option<String>, /// 0-based index when a response is delivered as multiple segments. #[serde(skip_serializing_if = "Option::is_none")] pub segment_index: Option<u32>, /// Total number of segments in a segmented delivery. #[serde(skip_serializing_if = "Option::is_none")] pub segment_total: Option<u32>, /// Fine-grained streaming payload for `text_delta`, `thinking_delta`, /// and `tool_args_delta` events. Concatenating `delta`s in order /// yields the full text/thinking/arguments string. #[serde(skip_serializing_if = "Option::is_none")] pub delta: Option<String>, /// Discriminator for the `delta` payload: `"text"`, `"thinking"`, /// or `"tool_args"`. Only set on streaming delta events. #[serde(skip_serializing_if = "Option::is_none")] pub delta_kind: Option<String>, /// Provider-assigned tool call id that groups `tool_args_delta` /// chunks together and ties them to the eventual `tool_call` / /// `tool_result` events. #[serde(skip_serializing_if = "Option::is_none")] pub tool_call_id: Option<String>, /// Optional citations attached to `chat_done` payloads. #[serde(skip_serializing_if = "Option::is_none")] pub citations: Option<serde_json::Value>, /// Sub-agent specific progress detail. Populated on /// `subagent_spawned`, `subagent_completed`, `subagent_iteration_start`, /// `subagent_tool_call`, and `subagent_tool_result` events so the UI /// can attribute child activity to the parent's live subagent row /// without overloading the flat top-level fields. `None` for any /// non-subagent event. #[serde(skip_serializing_if = "Option::is_none")] pub subagent: Option<SubagentProgressDetail>, /// Per-thread task board snapshot carried by `task_board_updated`. #[serde(skip_serializing_if = "Option::is_none")] pub task_board: Option<serde_json::Value>, /// Server-computed human label for a tool call (on `tool_call` / /// `subagent_tool_call`), e.g. "Reading messages". The frontend renders /// this verbatim for dynamic Composio/MCP/integration tools it can't /// label itself, falling back to its own formatter when absent. #[serde(skip_serializing_if = "Option::is_none")] pub tool_display_label: Option<String>, /// Server-computed contextual detail for a tool call (on `tool_call` / /// `subagent_tool_call`), e.g. "steven@gmail.com" — the bracketed target /// shown after [`Self::tool_display_label`]. #[serde(skip_serializing_if = "Option::is_none")] pub tool_display_detail: Option<String>, /// Holistic token/cost/context usage for a completed turn (parent + /// sub-agents), carried on `chat_done`. Lets the UI footer show session /// tokens, USD cost, and real context-window utilisation, with a /// per-sub-agent hover breakdown. `None` for every non-`chat_done` event and /// for synthetic done events that never ran a real turn. #[serde(skip_serializing_if = "Option::is_none")] pub usage: Option<TurnUsagePayload>, } /// Token/cost/context totals for one completed turn, attached to `chat_done`. /// /// Every numeric is a turn total (parent agent **plus** any sub-agents spawned /// during the turn); the `subagents` list breaks the same spend down per child /// for the UI hover. `context_window` is `0` when the core couldn't resolve the /// model's window (e.g. an unknown cloud model) — the UI falls back to a /// default in that case. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct TurnUsagePayload { pub input_tokens: u64, pub output_tokens: u64, pub cached_input_tokens: u64, pub cost_usd: f64, pub context_window: u64, /// Per-sub-agent spend, omitted from the wire when no sub-agents ran. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub subagents: Vec<SubagentUsagePayload>, } /// One sub-agent's token/cost contribution within a turn (hover breakdown). #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct SubagentUsagePayload { pub task_id: String, pub agent_id: String, pub input_tokens: u64, pub output_tokens: u64, pub cost_usd: f64, } /// Per-event subagent progress detail attached to `WebChannelEvent`. /// /// Carries the fields the parent thread's UI needs to render a live /// subagent block — child iteration counters, mode, child task/agent /// ids when distinct from the flat `tool_name` (which already carries /// the agent id on top-level subagent events but not on nested /// `subagent_tool_*` events where `tool_name` is the *child's* tool), /// and final-run statistics on `subagent_completed`. /// /// Every field is optional and skipped from the JSON payload when /// absent — this keeps the wire format compact for non-subagent events /// (where the whole struct is `None`) and lets new fields land /// non-breakingly behind older clients. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct SubagentProgressDetail { /// Resolved spawn mode — `"typed"` or `"fork"`. #[serde(skip_serializing_if = "Option::is_none")] pub mode: Option<String>, /// Whether the spawn requested a dedicated worker thread. #[serde(skip_serializing_if = "Option::is_none")] pub dedicated_thread: Option<bool>, /// Character length of the delegation prompt (on `subagent_spawned`). #[serde(skip_serializing_if = "Option::is_none")] pub prompt_chars: Option<u64>, /// Sub-agent's child iteration counter (on `subagent_iteration_start`, /// `subagent_tool_call`, `subagent_tool_result`). 1-based. #[serde(skip_serializing_if = "Option::is_none")] pub child_iteration: Option<u32>, /// Sub-agent's configured iteration cap. #[serde(skip_serializing_if = "Option::is_none")] pub child_max_iterations: Option<u32>, /// Child agent id (on nested `subagent_tool_*` events where the flat /// `tool_name` is the child's tool, not the agent). #[serde(skip_serializing_if = "Option::is_none")] pub agent_id: Option<String>, /// Spawn task id (on nested `subagent_tool_*` events). #[serde(skip_serializing_if = "Option::is_none")] pub task_id: Option<String>, /// Elapsed wall-clock for the call/run in milliseconds. #[serde(skip_serializing_if = "Option::is_none")] pub elapsed_ms: Option<u64>, /// Total iterations the sub-agent used (on `subagent_completed`). #[serde(skip_serializing_if = "Option::is_none")] pub iterations: Option<u32>, /// Character length of the sub-agent's final assistant text /// (on `subagent_completed`) or the tool result /// (on `subagent_tool_result`). #[serde(skip_serializing_if = "Option::is_none")] pub output_chars: Option<u64>, /// Persistent worker sub-thread id backing the delegation (on /// `subagent_spawned`). The frontend stores it on the subagent row and /// uses it to reopen the full parent↔subagent conversation from memory. #[serde(skip_serializing_if = "Option::is_none")] pub worker_thread_id: Option<String>, /// Human-readable display name from the agent registry (e.g. /// "Researcher", "Coding Agent"). The frontend uses this for /// consistent agent labels across timeline, sub-mascots, and drawer. #[serde(skip_serializing_if = "Option::is_none")] pub display_name: Option<String>, /// Absolute path to the worker's isolated `git worktree` checkout /// (on `subagent_completed`, when the worker ran with /// `isolation = "worktree"`). Drives the inline worktree row's /// open/diff/remove actions. `None` for non-isolated workers (#3376). #[serde(skip_serializing_if = "Option::is_none")] pub worktree_path: Option<String>, /// Files (relative to the worktree root) the worker changed, snapshot /// after the run (on `subagent_completed`). Absent for non-isolated /// workers and clean worktrees. #[serde(skip_serializing_if = "Option::is_none")] pub changed_files: Option<Vec<String>>, /// Whether the worker's worktree had uncommitted changes after the run /// (on `subagent_completed`). A dirty worktree must not be auto-removed — /// the UI requires an explicit user decision. `None` for non-isolated. #[serde(skip_serializing_if = "Option::is_none")] pub dirty_status: Option<bool>, } #[derive(Debug, Deserialize)] struct SocketRpcRequest { id: serde_json::Value, method: String, #[serde(default)] params: serde_json::Value, } #[derive(Debug, Deserialize)] struct ChatStartPayload { thread_id: String, message: String, #[serde(default)] model: Option<String>, #[serde(default)] model_override: Option<String>, #[serde(default)] temperature: Option<f64>, #[serde(default)] profile_id: Option<String>, #[serde(default)] locale: Option<String>, #[serde(default)] queue_mode: Option<String>, } #[derive(Debug, Deserialize)] struct ChatCancelPayload { thread_id: String, } #[derive(Debug, Deserialize)] struct ThreadSubscribePayload { thread_id: String, } /// Attaches the Socket.IO layer to the Axum router and sets up event handlers. /// /// It configures: /// - Client connection and room joining. /// - `rpc:request`: Invoking JSON-RPC methods over WebSocket. /// - `chat:start`: Initiating a new chat turn. /// - `chat:cancel`: Aborting an active chat turn. pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) { let (layer, io) = SocketIo::new_layer(); log::info!( "[socketio] engine ready (namespace /, path {})", io.config().engine_config.req_path ); io.ns( "/", |socket: SocketRef, TryData(handshake): TryData<HandshakeAuth>| { let client_id = socket.id.to_string(); // Reject cross-origin browser pages before the handshake completes. // Native clients (Tauri shell, CLI) do not set an `Origin` header and // are accepted; only browser pages from origins outside the local // app surface are dropped here. See `origin_is_allowed`. let origin = socket .req_parts() .headers .get(axum::http::header::ORIGIN) .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()); if !origin_is_allowed(origin.as_deref()) { log::warn!( "[socketio] rejecting connect: bad origin {:?} client={}", origin, client_id ); let _ = socket.clone().disconnect(); return; } // Verify the handshake bearer matches the per-process RPC token. // `TryData` lets us treat a missing/malformed `auth` payload as a // soft failure (no panic) and reject the connect cleanly. let supplied = handshake.ok().and_then(|h| h.token).unwrap_or_default(); if !crate::core::auth::verify_bearer_token(&supplied) { log::warn!( "[socketio] rejecting connect: missing or invalid bearer client={}", client_id ); let _ = socket.clone().disconnect(); return; } socket.extensions.insert(AuthedConnection); log::info!("[socketio] client connected id={client_id} (authenticated)"); // Join a room named after the client ID for targeted event delivery. join_room_logged(&socket, &client_id, &client_id); // Also auto-join the "system" room so every connected client // receives broadcast-style events that aren't tied to a // specific chat thread. Today this covers proactive messages // (welcome agent, morning briefing, cron-driven announcements) // which `channels::proactive::ProactiveMessageSubscriber` // emits with `client_id = "system"` — see `emit_web_channel_event`. // If this join fails the welcome message silently disappears, // so we log both success and failure for diagnosability. join_room_logged(&socket, "system", &client_id); let ready_payload = json!({ "sid": client_id }); log::debug!("[socketio] emit event=ready to_client={}", socket.id); let _ = socket.emit("ready", &ready_payload); // Handler for JSON-RPC over WebSocket. socket.on( "rpc:request", |socket: SocketRef, Data(payload): Data<SocketRpcRequest>| async move { if !socket_is_authed(&socket) { drop_unauthed(&socket, "rpc:request from unauthenticated socket"); return; } let client_id = socket.id.to_string(); log::info!( "[socketio] rpc:request method={} id={} client={}", payload.method, payload.id, client_id ); // Invoke the method through the same logic used by the HTTP RPC endpoint. let response = match crate::core::jsonrpc::invoke_method( crate::core::jsonrpc::default_state(), payload.method.as_str(), payload.params, ) .await { Ok(result) => ( "rpc:response", json!({ "id": payload.id, "result": result }), ), Err(message) => ( "rpc:error", json!({ "id": payload.id, "error": { "code": -32000, "message": message } }), ), }; let _ = socket.emit(response.0, &response.1); }, ); // Handler for starting a chat turn. socket.on( "chat:start", |socket: SocketRef, Data(payload): Data<ChatStartPayload>| async move { if !socket_is_authed(&socket) { drop_unauthed(&socket, "chat:start from unauthenticated socket"); return; } let client_id = socket.id.to_string(); let thread_id = payload.thread_id.clone(); let model_override = payload.model_override.or(payload.model); log::debug!( "[socketio] recv event=chat:start client_id={} thread_id={} message_bytes={}", client_id, thread_id, payload.message.len() ); // Trigger the web channel's chat logic. match crate::openhuman::channels::providers::web::start_chat( &client_id, &payload.thread_id, &payload.message, model_override, payload.temperature, payload.profile_id, payload.locale, payload.queue_mode, crate::openhuman::channels::providers::web::ChatRequestMetadata::default(), ) .await { Ok(request_id) => { let accepted_payload = json!({ "event": "chat_accepted", "client_id": client_id, "thread_id": thread_id, "request_id": request_id, }); emit_with_aliases(&socket, "chat_accepted", &accepted_payload); } Err(error) => { let error_payload = json!({ "event": "chat_error", "client_id": client_id, "thread_id": thread_id, "request_id": "", "message": error, "error_type": "inference", }); emit_with_aliases(&socket, "chat_error", &error_payload); } } }, ); // Handler for cancelling an active chat turn. socket.on( "chat:cancel", |socket: SocketRef, Data(payload): Data<ChatCancelPayload>| async move { if !socket_is_authed(&socket) { drop_unauthed(&socket, "chat:cancel from unauthenticated socket"); return; } let client_id = socket.id.to_string(); log::debug!( "[socketio] recv event=chat:cancel client_id={} thread_id={}", client_id, payload.thread_id ); let _ = crate::openhuman::channels::providers::web::cancel_chat( &client_id, &payload.thread_id, ) .await; }, ); // Handler for subscribing this socket to a thread's room. // // Chat-stream events are delivered to BOTH the initiating client's // own room AND a per-thread room (`thread:<id>`). After a socket // reconnects it has a NEW client_id, so it would miss an in-flight // turn's remaining stream (delivered to the OLD client_id room). The // frontend emits this on connect/reconnect for the active thread, so // the new socket re-joins the thread room and keeps receiving the // stream. Membership is dropped automatically on disconnect. socket.on( "thread:subscribe", |socket: SocketRef, Data(payload): Data<ThreadSubscribePayload>| async move { if !socket_is_authed(&socket) { drop_unauthed(&socket, "thread:subscribe from unauthenticated socket"); return; } let thread_id = payload.thread_id.trim(); if thread_id.is_empty() { return; } let room = format!("thread:{thread_id}"); join_room_logged(&socket, &room, &socket.id.to_string()); }, ); }, ); (layer, io) } /// Spawns background bridges to forward various system events to Socket.IO clients. /// /// This function sets up five bridges: /// 1. **Web Channel Bridge**: Forwards chat-related events (messages, tool calls) to specific clients. /// 2. **Dictation Bridge**: Forwards hotkey events to all clients. /// 3. **Overlay Bridge**: Forwards attention bubble events to all clients. /// 4. **Core Notification Bridge**: Forwards core notification events to all clients. /// 5. **Transcription Bridge**: Forwards real-time speech-to-text results to all clients. pub fn spawn_web_channel_bridge(io: SocketIo) { // 1. Web channel events → per-client rooms. let io_web = io.clone(); tokio::spawn(async move { let mut rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} web_channel events due to lag", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; emit_web_channel_event(&io_web, event); } log::debug!("[socketio] web_channel bridge stopped"); }); let io_overlay = io.clone(); let io_notify = io.clone(); let io_transcription = io.clone(); let io_auth = io.clone(); let io_companion = io.clone(); let io_mcp_setup = io.clone(); let io_memory_sync = io.clone(); let io_agent_meetings = io.clone(); let io_tinyplace = io.clone(); let io_channel_status = io.clone(); // 2. Dictation hotkey events → broadcast to all connected clients. tokio::spawn(async move { let mut rx = crate::openhuman::voice::dictation_listener::subscribe_dictation_events(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!("[socketio] dropped {} dictation events due to lag", skipped); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let Ok(payload) = serde_json::to_value(&event) { log::debug!( "[socketio] broadcast dictation:{} to all clients", event.event_type ); // Support both colon and underscore versions for compatibility with different frontends. let _ = io.emit("dictation:toggle", &payload); let _ = io.emit("dictation_toggle", &payload); } } log::debug!("[socketio] dictation bridge stopped"); }); // 3. Overlay attention events → broadcast to all clients. tokio::spawn(async move { let mut rx = crate::openhuman::overlay::subscribe_attention_events(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} overlay attention events due to lag", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let Ok(payload) = serde_json::to_value(&event) { log::debug!( "[socketio] broadcast overlay:attention source={:?}", event.source ); let _ = io_overlay.emit("overlay:attention", &payload); let _ = io_overlay.emit("overlay_attention", &payload); } } log::debug!("[socketio] overlay attention bridge stopped"); }); // 4. Core notification events → broadcast to all connected clients so // the in-app notification center picks them up regardless of which // chat session is active. Pattern mirrors the overlay attention // bridge above — fire-and-forget, no per-client routing. tokio::spawn(async move { let mut rx = crate::openhuman::notifications::subscribe_core_notifications(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} core_notification events due to lag", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let Ok(payload) = serde_json::to_value(&event) { log::debug!( "[socketio] broadcast core_notification id={} category={:?}", event.id, event.category ); let _ = io_notify.emit("core_notification", &payload); let _ = io_notify.emit("core:notification", &payload); } } log::debug!("[socketio] core_notification bridge stopped"); }); // 6. SessionExpired events → broadcast to all clients so the UI can // proactively tear down user-scoped state and route to onboarding // instead of waiting for the next poll to discover the JWT is gone. // Subscribes to the global event bus and filters for // `DomainEvent::SessionExpired`; ignores everything else. tokio::spawn(async move { // Poll until `event_bus::init_global` has run. Socket.IO bridges // spawn from `spawn_web_channel_bridge`, which on some startup // paths runs before `register_domain_subscribers` initialises // the bus. A one-shot check would silently no-op for the rest // of the process; a short polling loop with a hard cap retries // without spinning forever if init genuinely never happens // (e.g. tests that drive the socket layer in isolation). let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — SessionExpired bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (auth bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let crate::core::event_bus::DomainEvent::SessionExpired { source, reason } = event { log::info!( "[socketio] broadcast auth:session_expired source={} reason_len={}", source, reason.len() ); // The UI doesn't need the raw reason (already logged // server-side and we don't want auth-error strings in the // renderer console). Just send the source slug. let payload = serde_json::json!({ "source": source }); let _ = io_auth.emit("auth:session_expired", &payload); let _ = io_auth.emit("auth_session_expired", &payload); } } log::debug!("[socketio] auth session_expired bridge stopped"); }); // 6b. McpSetupSecretRequested → broadcast `mcp_setup:secret_requested` // so the UI can render a native input dialog. Only the opaque // ref + safe display fields are forwarded; raw secret values // are not part of the event payload. tokio::spawn(async move { let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — mcp_setup bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (mcp_setup bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let crate::core::event_bus::DomainEvent::McpSetupSecretRequested { ref_id, key_name, prompt, } = event { log::info!( "[socketio] broadcast mcp_setup:secret_requested ref={} key={}", ref_id, key_name ); let payload = serde_json::json!({ "ref_id": ref_id, "key_name": key_name, "prompt": prompt, }); let _ = io_mcp_setup.emit("mcp_setup:secret_requested", &payload); let _ = io_mcp_setup.emit("mcp_setup_secret_requested", &payload); } } log::debug!("[socketio] mcp_setup secret_requested bridge stopped"); }); // 5. Transcription results → broadcast to all connected clients. tokio::spawn(async move { let mut rx = crate::openhuman::voice::dictation_listener::subscribe_transcription_results(); loop { let text = match rx.recv().await { Ok(text) => text, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} transcription events due to lag", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; log::debug!( "[socketio] broadcast dictation:transcription ({} chars) to all clients", text.len() ); let payload = serde_json::json!({ "text": text }); let _ = io_transcription.emit("dictation:transcription", &payload); } log::debug!("[socketio] transcription bridge stopped"); }); // 7. Companion state change events → broadcast to all clients so the // overlay and settings panel can react to session lifecycle and // state transitions (Idle → Listening → Thinking → Speaking → …). tokio::spawn(async move { let mut rx = crate::openhuman::desktop_companion::bus::subscribe_state_changed(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} companion state_changed events due to lag", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; if let Ok(payload) = serde_json::to_value(&event) { log::debug!( "[socketio] broadcast companion:state_changed session={} {} -> {}", event.session_id, event.previous_state, event.state, ); let _ = io_companion.emit("companion:state_changed", &payload); let _ = io_companion.emit("companion_state_changed", &payload); } } log::debug!("[socketio] companion state bridge stopped"); }); // 8. Memory sync stage + tree-build progress → broadcast to all clients // so the UI can show real-time progress bars and refresh the graph. tokio::spawn(async move { let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — memory_sync bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (memory_sync bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; match event { crate::core::event_bus::DomainEvent::MemorySyncStageChanged { trigger, stage, provider, connection_id, detail, source_id, } => { let payload = serde_json::json!({ "trigger": trigger, "stage": stage, "provider": provider, "connection_id": connection_id, "detail": detail, // source_id is the memory-source row id for frontend per-row // indicator matching (RC#2, issue #3295). connection_id is // preserved unchanged for downstream consumers. "source_id": source_id, }); let _ = io_memory_sync.emit("memory:sync_stage", &payload); } crate::core::event_bus::DomainEvent::TreeSummarizerPropagated { namespace, node_id, level, token_count, } => { let payload = serde_json::json!({ "namespace": namespace, "node_id": node_id, "level": level, "token_count": token_count, }); let _ = io_memory_sync.emit("memory:tree_progress", &payload); } crate::core::event_bus::DomainEvent::TreeSummarizerRebuildCompleted { namespace, total_nodes, } => { let payload = serde_json::json!({ "namespace": namespace, "total_nodes": total_nodes, }); let _ = io_memory_sync.emit("memory:tree_completed", &payload); } crate::core::event_bus::DomainEvent::MemoryTreeBuildProgress { phase, step, tree_scope, level, item_count, detail, } => { let payload = serde_json::json!({ "phase": phase, "step": step, "tree_scope": tree_scope, "level": level, "item_count": item_count, "detail": detail, }); let _ = io_memory_sync.emit("memory:build_progress", &payload); } crate::core::event_bus::DomainEvent::HarnessInitProgress { step_id, state, message, percent, } => { let payload = serde_json::json!({ "step_id": step_id, "state": state, "message": message, "percent": percent, }); let _ = io_memory_sync.emit("init:progress", &payload); } crate::core::event_bus::DomainEvent::HarnessInitCompleted { overall, failed_required, } => { let payload = serde_json::json!({ "overall": overall, "failed_required": failed_required, }); let _ = io_memory_sync.emit("init:completed", &payload); } _ => {} } } log::debug!("[socketio] memory_sync bridge stopped"); }); // 9. Backend Meet bot events → broadcast to all connected frontend sockets. tokio::spawn(async move { let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — agent_meetings bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (agent_meetings bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; match event { crate::core::event_bus::DomainEvent::BackendMeetJoined { meet_url, correlation_id, } => { let payload = serde_json::json!({ "meet_url": meet_url, "correlation_id": correlation_id }); log::debug!("[socketio] broadcast agent_meetings:joined"); let _ = io_agent_meetings.emit("agent_meetings:joined", &payload); } crate::core::event_bus::DomainEvent::BackendMeetLeft { reason, correlation_id, } => { let payload = serde_json::json!({ "reason": reason, "correlation_id": correlation_id }); log::debug!("[socketio] broadcast agent_meetings:left reason={}", reason); let _ = io_agent_meetings.emit("agent_meetings:left", &payload); } crate::core::event_bus::DomainEvent::BackendMeetReply { transcript, reply, emotion, correlation_id, } => { let payload = serde_json::json!({ "transcript": transcript, "reply": reply, "emotion": emotion, "correlation_id": correlation_id, }); log::debug!( "[socketio] broadcast agent_meetings:reply reply_len={}", reply.len() ); let _ = io_agent_meetings.emit("agent_meetings:reply", &payload); } crate::core::event_bus::DomainEvent::BackendMeetHarness { transcript, instruction, emotion, correlation_id, } => { let payload = serde_json::json!({ "transcript": transcript, "instruction": instruction, "emotion": emotion, "correlation_id": correlation_id, }); log::debug!( "[socketio] broadcast agent_meetings:harness instruction_len={}", instruction.len() ); let _ = io_agent_meetings.emit("agent_meetings:harness", &payload); } crate::core::event_bus::DomainEvent::BackendMeetTranscript { turns, duration_ms, correlation_id, } => { let payload = serde_json::json!({ "turns": turns, "duration_ms": duration_ms, "correlation_id": correlation_id, }); log::debug!( "[socketio] broadcast agent_meetings:transcript turns={} duration_ms={}", turns.len(), duration_ms ); let _ = io_agent_meetings.emit("agent_meetings:transcript", &payload); } crate::core::event_bus::DomainEvent::BackendMeetError { error, correlation_id, } => { let payload = serde_json::json!({ "error": error, "correlation_id": correlation_id }); log::debug!("[socketio] broadcast agent_meetings:error"); let _ = io_agent_meetings.emit("agent_meetings:error", &payload); } _ => {} } } log::debug!("[socketio] agent_meetings bridge stopped"); }); // 10. Tinyplace stream events → broadcast to all connected frontend sockets. tokio::spawn(async move { let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — tinyplace bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (tinyplace bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; match event { crate::core::event_bus::DomainEvent::TinyPlaceStreamMessage { stream_id, kind, message, } => { let payload = json!({ "stream_id": stream_id, "kind": kind, "message": message, }); log::debug!( "[socketio] broadcast tinyplace:stream_message stream_id={} kind={}", stream_id, kind ); let _ = io_tinyplace.emit("tinyplace:stream_message", &payload); } crate::core::event_bus::DomainEvent::TinyPlaceStreamStatusChanged { stream_id, status, } => { let payload = json!({ "stream_id": stream_id, "status": status, }); log::debug!( "[socketio] broadcast tinyplace:stream_status stream_id={} status={}", stream_id, status ); let _ = io_tinyplace.emit("tinyplace:stream_status", &payload); } _ => {} } } log::debug!("[socketio] tinyplace stream bridge stopped"); }); // 11. Channel listener health → broadcast `channel:connection-updated` to // all clients so the Messaging tab reflects the *live* connection state // instead of a stale, credential-presence-only "Connected" (issue // #3712). The supervised listener publishes `ChannelConnected` when it // (re)enters its recv loop and `ChannelDisconnected { reason }` when it // errors/exits. Only listener-backed channels (telegram/discord // `bot_token`) fire these, so we map them to the `bot_token` auth mode — // the frontend `normalizeChannelConnectionUpdatePayload` drops any // channel/mode it doesn't recognise. tokio::spawn(async move { let bus = { const RETRY_INTERVAL_MS: u64 = 250; const MAX_WAIT_SECS: u64 = 30; let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; let mut attempts: u64 = 0; loop { if let Some(bus) = crate::core::event_bus::global() { break bus; } attempts += 1; if attempts > max_attempts { log::warn!( "[socketio] event_bus not initialised after {}s — channel_status bridge giving up", MAX_WAIT_SECS ); return; } tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; } }; let mut rx = bus.raw_receiver(); loop { let event = match rx.recv().await { Ok(event) => event, Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { log::warn!( "[socketio] dropped {} event_bus events due to lag (channel_status bridge)", skipped ); continue; } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; let payload = match event { crate::core::event_bus::DomainEvent::ChannelConnected { channel } => { log::debug!( "[socketio] broadcast channel:connection-updated {channel} -> connected" ); Some(channel_connection_update_payload( &channel, "connected", None, )) } crate::core::event_bus::DomainEvent::ChannelDisconnected { channel, reason } => { log::debug!( "[socketio] broadcast channel:connection-updated {channel} -> error reason_len={}", reason.len() ); Some(channel_connection_update_payload( &channel, "error", Some(&reason), )) } _ => None, }; if let Some(payload) = payload { // Emit both colon and underscore variants for FE compatibility. let _ = io_channel_status.emit("channel:connection-updated", &payload); let _ = io_channel_status.emit("channel_connection_updated", &payload); } } log::debug!("[socketio] channel_status bridge stopped"); }); } /// Build the `channel:connection-updated` payload broadcast when a supervised /// channel listener changes state (issue #3712). Listener-backed channels are /// always the `bot_token` auth mode (the only mode that materialises a runtime /// listener for telegram/discord); `last_error` carries the disconnect reason. /// Matches the shape consumed by the frontend /// `normalizeChannelConnectionUpdatePayload`. pub(crate) fn channel_connection_update_payload( channel: &str, status: &str, last_error: Option<&str>, ) -> serde_json::Value { let mut payload = serde_json::json!({ "channel": channel, "auth_mode": "bot_token", "status": status, }); if let Some(reason) = last_error { payload["last_error"] = serde_json::Value::String(reason.to_string()); } payload } /// Join `socket` to `room`, logging the result. /// /// `socket.join()` returns a `Result` that historically was discarded /// with `let _ = …`. Silent failure on the `"system"` room in /// particular makes proactive-message delivery vanish without a trace, /// so both the happy and error paths are logged with enough context /// (room name + client id) to diagnose missing welcome messages from /// logs alone. fn join_room_logged(socket: &SocketRef, room: &str, client_id: &str) { match socket.join(room.to_string()) { Ok(()) => log::debug!("[socketio] joined room '{room}' for client {client_id}"), Err(e) => log::warn!("[socketio] failed to join room '{room}' for client {client_id}: {e}"), } } fn emit_web_channel_event(io: &SocketIo, event: WebChannelEvent) { let name = event.event.clone(); // Deliver to the initiating client's own room AND the per-thread room. The // thread room lets a socket that reconnected with a new client_id (after // re-subscribing via `thread:subscribe`) keep receiving an in-flight turn's // stream. // // ⚠️ socketioxide (0.15.2) does NOT de-duplicate a socket present in // multiple target rooms: `LocalAdapter::apply_opts` flattens each room's // sid-set and collects WITHOUT a dedup pass, so `io.to([a, b]).emit()` // delivers TWICE to a socket in both `a` and `b`. The initiating client is // in both its `client_id` room and the `thread:<id>` room it subscribed to // → every streamed frame doubled ("double thinking"). So we emit to the // `client_id` room, then to the thread room EXCEPT the `client_id` room — // each socket is reached exactly once regardless of room overlap. // "system" broadcasts and events without a thread_id keep single-room delivery. let primary = event.client_id.clone(); let thread_room = (event.client_id != "system" && !event.thread_id.is_empty()) .then(|| format!("thread:{}", event.thread_id)); if let Ok(payload) = serde_json::to_value(event) { log::debug!( "[socketio] send event={} primary={} thread_room={:?} thread_id={} request_id={}", name, primary, thread_room, payload .get("thread_id") .and_then(|v| v.as_str()) .unwrap_or_default(), payload .get("request_id") .and_then(|v| v.as_str()) .unwrap_or_default() ); // Primary: the client_id room. let _ = io.to(primary.clone()).emit(&name, &payload); if let Some(alias) = event_alias(&name) { let _ = io.to(primary.clone()).emit(alias, &payload); } // Thread room minus the client_id room (dedup — see note above). if let Some(tr) = thread_room { let _ = io .to(tr.clone()) .except(primary.clone()) .emit(&name, &payload); if let Some(alias) = event_alias(&name) { let _ = io .to(tr.clone()) .except(primary.clone()) .emit(alias, &payload); } } } } /// Events that stream once per token (their payloads concatenate into the final /// text / thinking / tool-args). Emitting the legacy `:`-delimited alias for /// these doubles every frame on the wire — the "double thinking-token /// streaming" bug — and no client subscribes to the colon variant, so the alias /// is suppressed for exactly these. Enumerated explicitly rather than matched by /// a `*_delta` suffix, so a future *discrete* event whose name happens to end in /// `_delta` still gets its compat alias instead of being silently dropped. const STREAMING_DELTA_EVENTS: &[&str] = &["text_delta", "thinking_delta", "tool_args_delta"]; fn event_alias(name: &str) -> Option<String> { // Match against the canonical underscore form after stripping a `subagent_` // prefix (subagent streaming mirrors the parent's deltas), so `text_delta`, // `text:delta`, and `subagent_text_delta` all resolve to a listed event. // Lower-frequency discrete events keep the compat alias. let normalized = name.replace(':', "_"); let base = normalized.strip_prefix("subagent_").unwrap_or(&normalized); if STREAMING_DELTA_EVENTS.contains(&base) { return None; } if name.contains('_') { return Some(name.replace('_', ":")); } if name.contains(':') { return Some(name.replace(':', "_")); } None } fn emit_with_aliases(socket: &SocketRef, name: &str, payload: &serde_json::Value) { let _ = socket.emit(name, payload); if let Some(alias) = event_alias(name) { let _ = socket.emit(alias, payload); } } #[cfg(test)] mod tests { use super::{channel_connection_update_payload, event_alias, origin_is_allowed}; #[test] fn channel_connection_update_payload_connected_omits_error() { let payload = channel_connection_update_payload("discord", "connected", None); assert_eq!(payload["channel"], "discord"); // Listener-backed channels always map to the bot_token auth mode. assert_eq!(payload["auth_mode"], "bot_token"); assert_eq!(payload["status"], "connected"); assert!( payload.get("last_error").is_none(), "connected payload must not carry a last_error: {payload}" ); } #[test] fn channel_connection_update_payload_error_carries_reason() { let payload = channel_connection_update_payload("discord", "error", Some("gateway closed (4004)")); assert_eq!(payload["channel"], "discord"); assert_eq!(payload["auth_mode"], "bot_token"); assert_eq!(payload["status"], "error"); assert_eq!(payload["last_error"], "gateway closed (4004)"); } #[test] fn event_alias_translates_between_delimiters() { assert_eq!(event_alias("chat_done").as_deref(), Some("chat:done")); assert_eq!(event_alias("chat:error").as_deref(), Some("chat_error")); assert_eq!(event_alias("ready"), None); } #[test] fn event_alias_suppressed_for_streaming_deltas() { // Streaming deltas must NOT be aliased — doubling every token frame is // the "double thinking-token streaming" bug. Discrete events still alias. assert_eq!(event_alias("thinking_delta"), None); assert_eq!(event_alias("text_delta"), None); assert_eq!(event_alias("tool_args_delta"), None); assert_eq!(event_alias("subagent_tool_args_delta"), None); // A *discrete* event that merely ends in `_delta` is NOT a streaming // token event and must keep its compat alias — this is what the explicit // STREAMING_DELTA_EVENTS set guarantees over the old `*_delta` suffix. assert_eq!( event_alias("inventory_delta").as_deref(), Some("inventory:delta") ); // Sanity: a non-delta event in the same family still aliases. assert_eq!(event_alias("tool_call").as_deref(), Some("tool:call")); } #[test] fn origin_allowlist_accepts_native_clients() { assert!(origin_is_allowed(None)); } #[test] fn origin_allowlist_accepts_tauri_localhost_across_schemes() { // The CEF-served app shell stamps a platform-dependent Origin: // - macOS / iOS use the native `tauri://localhost` scheme // - Windows uses the CEF custom HTTP protocol → `http://tauri.localhost` // - Linux / older Windows builds use `https://tauri.localhost` // All three flavours are the same trust tier (the bundled webview), // so each must pass the handshake gate. assert!(origin_is_allowed(Some("tauri://localhost"))); assert!(origin_is_allowed(Some("https://tauri.localhost"))); assert!(origin_is_allowed(Some("http://tauri.localhost"))); } #[test] fn origin_allowlist_accepts_local_dev_server() { assert!(origin_is_allowed(Some("http://localhost:1420"))); assert!(origin_is_allowed(Some("http://127.0.0.1:1420"))); assert!(origin_is_allowed(Some("http://[::1]:1420"))); // Loopback without an explicit port (some CEF builds stamp this // shape when the shell runs on the default port). assert!(origin_is_allowed(Some("http://localhost"))); } #[test] fn origin_allowlist_rejects_cross_origin_browser_pages() { assert!(!origin_is_allowed(Some("https://attacker.example"))); assert!(!origin_is_allowed(Some("http://evil.local"))); assert!(!origin_is_allowed(Some("null"))); assert!(!origin_is_allowed(Some(""))); } #[test] fn origin_allowlist_rejects_host_prefix_decoys() { // Regression: `starts_with("localhost")` accepted these; the exact // host match must not. assert!(!origin_is_allowed(Some( "http://localhost.attacker.example" ))); assert!(!origin_is_allowed(Some( "http://127.0.0.1.attacker.example" ))); assert!(!origin_is_allowed(Some("https://localhost-evil"))); // Same rule applies to the tauri.localhost host — must be exact. assert!(!origin_is_allowed(Some( "http://tauri.localhost.attacker.example" ))); assert!(!origin_is_allowed(Some("https://tauri.localhost.evil"))); } #[test] fn origin_allowlist_rejects_unparseable_origin() { assert!(!origin_is_allowed(Some("not a url"))); assert!(!origin_is_allowed(Some("javascript:alert(1)"))); } }