admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cef): in-process CDP DevTools transport (foundation) (#3498)

Browse Source
This commit is contained in:
oxoxDev
2026-06-08 23:39:21 -07:00
committed by GitHub
parent f68069e502
commit 4c886765cc
8 changed files with 886 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src-tauri/src/cdp/conn.rs
+205 -94
View File
@@ -1,26 +1,40 @@
//! CDP WebSocket client. Supports both short-lived request/response ticks
//! (whatsapp / slack / telegram periodic scans) and long-lived streaming
//! sessions with a pending-id table (discord MITM, and the new per-account
//! session opener).
//! [`CdpConn`] — per-attach handle on top of the in-process CDP transport.
//!
//! Not re-entrant: `call` is sequential during the setup phase, and once
//! `pump_events` takes over the read stream callers issue follow-up calls
//! via the pending-table machinery (TODO — V1.5, not needed yet).
//! Wraps an [`Arc<WebviewCdpTransport>`](super::in_process::WebviewCdpTransport)
//! with the same `call` / `pump_events` surface the scanners and the
//! per-account session opener were already using. The previous
//! WebSocket-backed implementation (one socket per attach) is gone for
//! new code paths; all attaches for a given webview now share the same
//! in-process channel, and a [`CdpConn`] is just a cheap session-scoped
//! view.
//!
//! For backward compatibility with per-scanner duplicated implementations
//! that still attach via the TCP loopback DevTools port (whatsapp,
//! slack, telegram, wechat, meet_video — see issue follow-up), the
//! [`CdpConn::open_ws`] legacy constructor keeps the old `tungstenite`
//! WebSocket path alive. Callers should migrate to
//! [`super::conn_for_account`] over time.
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::oneshot;
use tokio_tungstenite::{connect_async, tungstenite::Message};
/// Timeout applied to a single request/response round-trip during the setup
/// phase. Long enough to cover a cold-attach on a sluggish machine;
/// `pump_events` uses no timeout since CDP events can arrive hours apart.
const CALL_TIMEOUT: Duration = Duration::from_secs(35);
use super::in_process::{EventFrame, WebviewCdpTransport};
pub struct CdpConn {
/// Timeout applied to a single request/response round-trip in the
/// legacy WebSocket transport. Long enough to cover a cold-attach on a
/// sluggish machine; the in-process transport uses
/// [`crate::cdp::CALL_TIMEOUT`] (also 35s) for symmetry.
const LEGACY_CALL_TIMEOUT: Duration = Duration::from_secs(35);
/// Internal: legacy WebSocket dispatch state.
struct LegacyWs {
sink: futures_util::stream::SplitSink<
tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
@@ -36,113 +50,210 @@ pub struct CdpConn {
pending: HashMap<i64, oneshot::Sender<Result<Value, String>>>,
}
enum Backend {
InProcess(Arc<WebviewCdpTransport>),
LegacyWs(LegacyWs),
}
/// Per-attach CDP handle. Internally either wraps an
/// `Arc<WebviewCdpTransport>` (in-process channel) or a tungstenite
/// `WebSocketStream` (legacy TCP loopback). The session_id filter is
/// per-handle so concurrent attachers don't see each other's events.
pub struct CdpConn {
backend: Backend,
label: String,
}
impl CdpConn {
pub async fn open(ws_url: &str) -> Result<Self, String> {
/// Wrap an already-installed in-process transport. Callers obtain
/// the transport from the per-app [`super::CdpRegistry`]
/// (`app.state()`) — typically via
/// [`super::conn_for_account`].
pub fn new(transport: Arc<WebviewCdpTransport>) -> Self {
let label = transport.label().to_string();
Self {
backend: Backend::InProcess(transport),
label,
}
}
/// Legacy: open a CDP connection over the loopback TCP WebSocket
/// exposed by `--remote-debugging-port`. Kept for the per-scanner
/// duplicated implementations (whatsapp, slack, telegram, wechat,
/// meet_video) that have not yet migrated to the in-process
/// channel. New code paths should use [`super::conn_for_account`].
pub async fn open_ws(ws_url: &str) -> Result<Self, String> {
let (ws, _resp) = connect_async(ws_url)
.await
.map_err(|e| format!("ws connect: {e}"))?;
let (sink, stream) = ws.split();
Ok(Self {
sink,
stream,
next_id: 1,
pending: HashMap::new(),
backend: Backend::LegacyWs(LegacyWs {
sink,
stream,
next_id: 1,
pending: HashMap::new(),
}),
label: format!("ws:{ws_url}"),
})
}
/// Setup-phase request/response: sends a JSON-RPC call and drains inbound
/// messages until the matching response arrives. Unrelated events and
/// responses for other ids are dropped on the floor — only safe before
/// `pump_events` takes over the read side.
/// Setup-phase request/response: sends a JSON-RPC call and awaits
/// the matching response. `session_id`, when supplied, is inlined
/// into the envelope so the call routes to a previously-attached
/// child target (via `Target.attachToTarget`).
pub async fn call(
&mut self,
method: &str,
params: Value,
session_id: Option<&str>,
) -> Result<Value, String> {
let id = self.next_id;
self.next_id += 1;
let mut req = json!({ "id": id, "method": method, "params": params });
if let Some(s) = session_id {
req["sessionId"] = json!(s);
}
let body = serde_json::to_string(&req).map_err(|e| format!("encode: {e}"))?;
self.sink
.send(Message::Text(body))
.await
.map_err(|e| format!("ws send: {e}"))?;
loop {
let msg = tokio::time::timeout(CALL_TIMEOUT, self.stream.next())
.await
.map_err(|_| format!("ws read timeout (method={method})"))?
.ok_or_else(|| format!("ws closed (method={method})"))?
.map_err(|e| format!("ws recv: {e}"))?;
let text = match msg {
Message::Text(t) => t,
Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
continue
}
Message::Close(_) => return Err("ws closed".into()),
};
let v: Value = serde_json::from_str(&text).map_err(|e| format!("decode: {e}"))?;
if v.get("id").and_then(|x| x.as_i64()) != Some(id) {
continue;
}
if let Some(err) = v.get("error") {
return Err(format!("cdp error: {err}"));
}
return Ok(v.get("result").cloned().unwrap_or(Value::Null));
match &mut self.backend {
Backend::InProcess(transport) => transport.call(method, params, session_id).await,
Backend::LegacyWs(ws) => legacy_ws_call(ws, method, params, session_id).await,
}
}
/// Take over the read stream and dispatch every inbound event via the
/// supplied callback until the WebSocket closes. Responses to outstanding
/// `call` requests (none in V1) route through `pending`.
/// Subscribe to the transport's event stream and dispatch every
/// inbound CDP event via the supplied callback until the channel
/// signals it cannot keep up.
///
/// `session_id` filters incoming events: CDP multiplexes all sessions
/// through one ws once `flatten: true` is set, so we drop events
/// belonging to other sessions.
/// `session_id` filters incoming events — CDP multiplexes all
/// sessions through the same transport when `flatten: true` is set,
/// so we drop events belonging to other sessions.
///
/// Returns when the channel closes (the transport has been
/// forgotten / ws shut down) or on an unrecoverable error.
/// `Lagged` is treated as a continuation signal — the caller's idle
/// watchdog will eventually time out the session and the outer
/// reconnect loop re-attaches.
pub async fn pump_events<F>(&mut self, session_id: &str, mut on_event: F) -> Result<(), String>
where
F: FnMut(&str, &Value),
{
loop {
let msg = self
.stream
.next()
.await
.ok_or_else(|| "ws closed".to_string())?
.map_err(|e| format!("ws recv: {e}"))?;
let text = match msg {
Message::Text(t) => t,
Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
continue
match &mut self.backend {
Backend::InProcess(transport) => {
let mut rx = transport.subscribe_events();
loop {
match rx.recv().await {
Ok(EventFrame {
method,
params,
session_id: evt_session,
}) => {
if !evt_session.is_empty() && evt_session != session_id {
continue;
}
on_event(&method, &params);
}
Err(RecvError::Lagged(skipped)) => {
log::warn!(
"[cdp][{}] event channel lagged skipped={} session_id={}",
self.label,
skipped,
session_id
);
continue;
}
Err(RecvError::Closed) => return Ok(()),
}
}
Message::Close(_) => return Ok(()),
};
let v: Value = match serde_json::from_str(&text) {
Ok(v) => v,
Err(_) => continue,
};
if let Some(id) = v.get("id").and_then(|x| x.as_i64()) {
if let Some(tx) = self.pending.remove(&id) {
let res = if let Some(err) = v.get("error") {
Err(format!("cdp error: {err}"))
} else {
Ok(v.get("result").cloned().unwrap_or(Value::Null))
};
let _ = tx.send(res);
}
continue;
}
let method = v.get("method").and_then(|x| x.as_str()).unwrap_or("");
let evt_session = v.get("sessionId").and_then(|x| x.as_str()).unwrap_or("");
if !evt_session.is_empty() && evt_session != session_id {
continue;
}
let params = v.get("params").cloned().unwrap_or(Value::Null);
on_event(method, &params);
Backend::LegacyWs(ws) => legacy_ws_pump_events(ws, session_id, on_event).await,
}
}
/// Diagnostic helper — webview label (in-process) or
/// `"ws:<url>"` (legacy WS) this connection is bound to.
pub fn label(&self) -> &str {
&self.label
}
}
async fn legacy_ws_call(
ws: &mut LegacyWs,
method: &str,
params: Value,
session_id: Option<&str>,
) -> Result<Value, String> {
let id = ws.next_id;
ws.next_id += 1;
let mut req = json!({ "id": id, "method": method, "params": params });
if let Some(s) = session_id {
req["sessionId"] = json!(s);
}
let body = serde_json::to_string(&req).map_err(|e| format!("encode: {e}"))?;
ws.sink
.send(Message::Text(body))
.await
.map_err(|e| format!("ws send: {e}"))?;
loop {
let msg = tokio::time::timeout(LEGACY_CALL_TIMEOUT, ws.stream.next())
.await
.map_err(|_| format!("ws read timeout (method={method})"))?
.ok_or_else(|| format!("ws closed (method={method})"))?
.map_err(|e| format!("ws recv: {e}"))?;
let text = match msg {
Message::Text(t) => t,
Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
continue
}
Message::Close(_) => return Err("ws closed".into()),
};
let v: Value = serde_json::from_str(&text).map_err(|e| format!("decode: {e}"))?;
if v.get("id").and_then(|x| x.as_i64()) != Some(id) {
continue;
}
if let Some(err) = v.get("error") {
return Err(format!("cdp error: {err}"));
}
return Ok(v.get("result").cloned().unwrap_or(Value::Null));
}
}
async fn legacy_ws_pump_events<F>(
ws: &mut LegacyWs,
session_id: &str,
mut on_event: F,
) -> Result<(), String>
where
F: FnMut(&str, &Value),
{
loop {
let msg = ws
.stream
.next()
.await
.ok_or_else(|| "ws closed".to_string())?
.map_err(|e| format!("ws recv: {e}"))?;
let text = match msg {
Message::Text(t) => t,
Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {
continue
}
Message::Close(_) => return Ok(()),
};
let v: Value = match serde_json::from_str(&text) {
Ok(v) => v,
Err(_) => continue,
};
if let Some(id) = v.get("id").and_then(|x| x.as_i64()) {
if let Some(tx) = ws.pending.remove(&id) {
let res = if let Some(err) = v.get("error") {
Err(format!("cdp error: {err}"))
} else {
Ok(v.get("result").cloned().unwrap_or(Value::Null))
};
let _ = tx.send(res);
}
continue;
}
let method = v.get("method").and_then(|x| x.as_str()).unwrap_or("");
let evt_session = v.get("sessionId").and_then(|x| x.as_str()).unwrap_or("");
if !evt_session.is_empty() && evt_session != session_id {
continue;
}
let params = v.get("params").cloned().unwrap_or(Value::Null);
on_event(method, &params);
}
}
app/src-tauri/src/cdp/in_process.rs
+497
View File
@@ -0,0 +1,497 @@
//! In-process Chrome DevTools Protocol transport built on
//! [`tauri::Webview::send_dev_tools_message`] +
//! [`tauri::Webview::on_dev_tools_protocol`].
//!
//! Replaces the legacy WebSocket-to-loopback transport that required
//! Chromium to be spawned with `--remote-debugging-port=19222`. The old
//! transport opened an unauthenticated TCP listener that any same-UID
//! local process could attach to. The in-process path stays entirely
//! within our process boundary — there is no listener for an external
//! attacker to reach.
//!
//! # Architecture
//!
//! - One [`WebviewCdpTransport`] per CEF webview. Installed at the
//! moment a webview is created by `webview_accounts` (and similar
//! creation sites). The webview-creator calls
//! [`install_for_webview`] with its concrete `Webview<tauri::Cef>`
//! handle.
//! - The bootstrapped transport is registered in a process-global
//! [`CdpRegistry`] (`app.state::<CdpRegistry>()`) keyed by account id
//! so scanners that are generic over `Runtime: tauri::Runtime` never
//! need to obtain a typed `Webview<Cef>` themselves.
//! - The transport registers a single
//! [`tauri::Webview::on_dev_tools_protocol`] callback at install
//! time. The CEF runtime keeps the callback alive for the webview's
//! whole lifetime; there is no un-register path, so the install must
//! be a one-shot keyed by webview label.
//! - Every outbound request gets a unique numeric `id`. A pending-map
//! ([`PendingMap`]) routes responses back to the awaiting
//! `tokio::sync::oneshot`.
//! - Events (CDP frames with no `id`) are fanned out through a
//! `tokio::sync::broadcast` channel. Each subscriber sees its own
//! queue.
//!
//! # Wire format
//!
//! Tauri-CEF delivers each inbound DevTools message in three variants
//! (`CefDevToolsProtocol::Message` raw JSON, `MethodResult` pre-parsed
//! response, `Event` pre-parsed event). We listen on the raw `Message`
//! variant only — it carries the full envelope including `sessionId`,
//! which the other two strip.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use serde_json::{json, Value};
use tokio::sync::{broadcast, oneshot};
use tauri::{AppHandle, Manager, Webview};
/// Timeout for a single request/response round-trip. Long enough for cold
/// attach on slow machines but short enough to fail fast on a stuck channel.
pub const CALL_TIMEOUT: Duration = Duration::from_secs(35);
/// Capacity of the per-transport event broadcast channel. CDP can produce
/// bursts (e.g. on `Page.enable` the initial frame-history dump). 256 keeps
/// memory bounded while absorbing typical burst sizes.
const EVENT_CHANNEL_CAP: usize = 256;
type PendingMap = Arc<Mutex<HashMap<i64, oneshot::Sender<Result<Value, String>>>>>;
/// One CDP frame delivered as an event (no `id` in the envelope).
#[derive(Clone, Debug)]
pub struct EventFrame {
pub method: String,
pub params: Value,
pub session_id: String,
}
/// Per-webview CDP transport. Holds the pending-id table and event
/// broadcaster. Constructed by [`install_for_webview`] and (typically)
/// inserted into the per-app [`CdpRegistry`] for later lookup.
pub struct WebviewCdpTransport {
label: String,
webview: Webview<tauri::Cef>,
next_id: Mutex<i64>,
pending: PendingMap,
events_tx: broadcast::Sender<EventFrame>,
}
impl WebviewCdpTransport {
/// Submit a CDP request and await its response.
///
/// The request `id` is auto-assigned. `session_id`, when supplied, is
/// inlined into the envelope so the call routes to a previously-attached
/// child target.
pub async fn call(
self: &Arc<Self>,
method: &str,
params: Value,
session_id: Option<&str>,
) -> Result<Value, String> {
let id = {
let mut n = self.next_id.lock().expect("next_id mutex poisoned");
let id = *n;
*n += 1;
id
};
let (tx, rx) = oneshot::channel();
{
let mut p = self.pending.lock().expect("pending mutex poisoned");
p.insert(id, tx);
}
let mut req = json!({ "id": id, "method": method, "params": params });
if let Some(s) = session_id {
req["sessionId"] = json!(s);
}
let body = serde_json::to_vec(&req).map_err(|e| format!("encode: {e}"))?;
// `send_dev_tools_message` blocks on a `std::sync::mpsc::Receiver`
// while the message is dispatched onto the CEF main thread. Off-load
// to a blocking pool so we don't park a tokio worker thread.
log::trace!(
"[cdp][{}] >> id={} method={} session_id={:?}",
self.label,
id,
method,
session_id
);
let webview = self.webview.clone();
let send_res = tauri::async_runtime::spawn_blocking(move || {
webview
.send_dev_tools_message(&body)
.map_err(|e| format!("send_dev_tools_message: {e}"))
})
.await
.map_err(|e| format!("spawn_blocking join: {e}"))?;
if let Err(e) = send_res {
// Clean up the pending entry; otherwise it leaks until next call.
let mut p = self.pending.lock().expect("pending mutex poisoned");
p.remove(&id);
return Err(e);
}
match tokio::time::timeout(CALL_TIMEOUT, rx).await {
Ok(Ok(res)) => res,
Ok(Err(_)) => {
// oneshot dropped — transport torn down between dispatch and
// receive. Surface as a transport error rather than panic.
Err(format!("cdp response channel dropped (method={method})"))
}
Err(_) => {
// Timeout. Make a best-effort to evict the pending entry so
// a delayed response doesn't pollute the next caller.
let mut p = self.pending.lock().expect("pending mutex poisoned");
p.remove(&id);
Err(format!("cdp call timeout (method={method})"))
}
}
}
/// Subscribe to the event stream for this transport. Each subscriber
/// gets its own queue; lagged subscribers see `RecvError::Lagged` and
/// must re-sync.
pub fn subscribe_events(self: &Arc<Self>) -> broadcast::Receiver<EventFrame> {
self.events_tx.subscribe()
}
/// Webview label associated with this transport — for diagnostics only.
pub fn label(&self) -> &str {
&self.label
}
}
/// Process-global registry of installed CDP transports keyed by webview
/// label. Stored via `app.manage(CdpRegistry::default())` at boot so
/// scanners and the per-account session opener can look up the correct
/// transport without holding a typed `Webview<Cef>` themselves.
#[derive(Default)]
pub struct CdpRegistry {
transports: Mutex<HashMap<String, Arc<WebviewCdpTransport>>>,
}
impl CdpRegistry {
/// Look up an installed transport by webview label. Returns `None`
/// when the webview has not been created yet (cold boot before the
/// account is opened, or after it was forgotten).
pub fn by_label(&self, label: &str) -> Option<Arc<WebviewCdpTransport>> {
self.transports
.lock()
.expect("CdpRegistry mutex poisoned")
.get(label)
.cloned()
}
/// Look up the transport for an `acct_*`-labelled webview. The
/// account id is the unsuffixed value passed to
/// [`webview_accounts::label_for`] (typically already sanitized).
pub fn by_account(&self, account_id: &str) -> Option<Arc<WebviewCdpTransport>> {
self.by_label(&format!("acct_{account_id}"))
}
/// Atomic "get or create" for a transport keyed by webview label.
///
/// Holds the registry mutex across both the existence check AND the
/// caller-supplied creator, so two concurrent
/// [`install_for_webview`] callers can never both register a
/// `on_dev_tools_protocol` observer on the same webview. The CEF
/// runtime offers no un-register hook, so a double-registration is
/// permanent — every inbound frame would fan out to both observer
/// closures, splitting `id`-keyed responses between two
/// disconnected `next_id` counters and breaking response routing.
///
/// The creator closure is responsible for constructing the transport
/// (which includes registering the CEF observer). If the closure
/// returns `Err`, the registry is unchanged.
fn get_or_create<F>(&self, label: &str, create: F) -> Result<Arc<WebviewCdpTransport>, String>
where
F: FnOnce() -> Result<Arc<WebviewCdpTransport>, String>,
{
let mut t = self.transports.lock().expect("CdpRegistry mutex poisoned");
if let Some(existing) = t.get(label) {
return Ok(Arc::clone(existing));
}
let transport = create()?;
t.insert(label.to_string(), Arc::clone(&transport));
Ok(transport)
}
/// Remove a transport from the registry by label. Called when a
/// webview is closed / forgotten so subsequent
/// [`Self::by_label`] / [`Self::by_account`] lookups return `None`
/// instead of a stale entry.
pub fn forget_label(&self, label: &str) {
self.transports
.lock()
.expect("CdpRegistry mutex poisoned")
.remove(label);
}
/// Convenience wrapper around [`Self::forget_label`] for the
/// account-suffixed label scheme.
pub fn forget_account(&self, account_id: &str) {
self.forget_label(&format!("acct_{account_id}"));
}
}
/// Process-global typed CEF [`AppHandle`]. Populated exactly once from
/// the `Builder::<tauri::Cef>::setup` callback in `lib.rs`. Used by
/// [`install_for_account`] to look up a webview by label with the
/// concrete `Webview<Cef>` typed handle that
/// [`Webview::send_dev_tools_message`] requires — avoids a
/// runtime-generic transmute at the install site.
///
/// `OnceLock` is appropriate because the cell is written exactly once,
/// during `Builder::setup`, before any webview is created.
static CEF_APP_HANDLE: OnceLock<AppHandle<tauri::Cef>> = OnceLock::new();
/// Record the typed CEF [`AppHandle`] in the process-global cell. Called
/// once from the `Builder::setup` callback in `lib.rs`. Subsequent calls
/// are silently ignored (`OnceLock::set` returns `Err` on the second
/// write) so re-entry during hot-reload doesn't panic.
pub fn set_cef_app_handle(app: AppHandle<tauri::Cef>) {
let _ = CEF_APP_HANDLE.set(app);
}
/// Install (or look up) the in-process CDP transport for an
/// account-keyed webview. The account id is the same value the
/// `webview_accounts::label_for` helper composes a label from
/// (`acct_{id}`). Idempotent — repeated calls for the same account
/// return the cached transport.
///
/// Returns `Err` when the webview hasn't been created yet (the typical
/// cold-boot race against a scanner that started before
/// `webview_accounts::open` finished). Callers back off and retry.
pub fn install_for_account(account_id: &str) -> Result<Arc<WebviewCdpTransport>, String> {
let app = CEF_APP_HANDLE
.get()
.ok_or_else(|| "cdp::set_cef_app_handle has not been called yet".to_string())?;
let label = format!("acct_{account_id}");
let registry_state = app
.try_state::<CdpRegistry>()
.ok_or_else(|| "CdpRegistry not managed by app".to_string())?;
let registry = registry_state.inner();
if let Some(existing) = registry.by_label(&label) {
return Ok(existing);
}
let webview = app
.get_webview(&label)
.ok_or_else(|| format!("no webview for label={label}"))?;
install_for_webview(registry, webview)
}
/// Install a CDP transport on `webview`, register the observer, and
/// insert the resulting [`WebviewCdpTransport`] into `registry`.
///
/// Idempotent and concurrency-safe: the existence check, observer
/// registration, and registry insert all happen while the registry
/// mutex is held, so two concurrent callers for the same webview
/// label can never both attach an observer (CEF gives no un-register
/// hook — a double-registration would permanently split responses
/// between two `next_id` counters).
pub fn install_for_webview(
registry: &CdpRegistry,
webview: Webview<tauri::Cef>,
) -> Result<Arc<WebviewCdpTransport>, String> {
let label = webview.label().to_string();
registry.get_or_create(&label, || {
let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
let (events_tx, _rx0) = broadcast::channel::<EventFrame>(EVENT_CHANNEL_CAP);
let transport = Arc::new(WebviewCdpTransport {
label: label.clone(),
webview: webview.clone(),
next_id: Mutex::new(1),
pending: Arc::clone(&pending),
events_tx: events_tx.clone(),
});
let pending_for_observer = Arc::clone(&pending);
let events_for_observer = events_tx.clone();
let label_for_observer = label.clone();
webview
.on_dev_tools_protocol(move |protocol| {
use tauri::CefDevToolsProtocol as P;
// Listen only on the raw `Message` variant — it carries
// the full envelope (including `sessionId`) which
// `MethodResult` and `Event` strip. Tauri-CEF still
// fires all three for the same wire message, so
// consuming both would double-dispatch.
if let P::Message(bytes) = protocol {
let v: Value = match serde_json::from_slice(&bytes) {
Ok(v) => v,
Err(e) => {
log::warn!(
"[cdp][{}] inbound: parse error: {} (bytes_len={})",
label_for_observer,
e,
bytes.len()
);
return;
}
};
handle_inbound(
&label_for_observer,
&v,
&pending_for_observer,
&events_for_observer,
);
}
})
.map_err(|e| format!("on_dev_tools_protocol: {e}"))?;
log::info!(
"[cdp][{}] in-process transport installed (observer registered)",
label
);
Ok(transport)
})
}
fn handle_inbound(
label: &str,
v: &Value,
pending: &PendingMap,
events_tx: &broadcast::Sender<EventFrame>,
) {
if let Some(id) = v.get("id").and_then(|x| x.as_i64()) {
let waiter = {
let mut p = pending.lock().expect("pending mutex poisoned");
p.remove(&id)
};
match waiter {
Some(tx) => {
let res = if let Some(err) = v.get("error") {
Err(format!("cdp error: {err}"))
} else {
Ok(v.get("result").cloned().unwrap_or(Value::Null))
};
let _ = tx.send(res);
}
None => {
log::trace!(
"[cdp][{}] inbound: orphan response id={} (caller already gone)",
label,
id
);
}
}
return;
}
let Some(method) = v.get("method").and_then(|x| x.as_str()) else {
return;
};
let params = v.get("params").cloned().unwrap_or(Value::Null);
let session_id = v
.get("sessionId")
.and_then(|x| x.as_str())
.unwrap_or("")
.to_string();
let frame = EventFrame {
method: method.to_string(),
params,
session_id,
};
// `send` only errors when there are zero subscribers — fine to drop
// events when nobody is listening.
let _ = events_tx.send(frame);
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// Inbound response carrying a known `id` resolves the matching pending
/// oneshot with the `result` payload.
#[tokio::test]
async fn inbound_response_resolves_pending() {
let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
let (events_tx, _) = broadcast::channel::<EventFrame>(8);
let (tx, rx) = oneshot::channel();
pending.lock().unwrap().insert(42, tx);
handle_inbound(
"test",
&json!({ "id": 42, "result": { "ok": true } }),
&pending,
&events_tx,
);
let got = rx.await.expect("oneshot resolved").expect("ok response");
assert_eq!(got, json!({ "ok": true }));
assert!(
pending.lock().unwrap().is_empty(),
"pending entry must be evicted on resolve"
);
}
/// Inbound error frame routes to the pending sender as `Err(_)`.
#[tokio::test]
async fn inbound_error_resolves_pending_as_err() {
let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
let (events_tx, _) = broadcast::channel::<EventFrame>(8);
let (tx, rx) = oneshot::channel();
pending.lock().unwrap().insert(7, tx);
handle_inbound(
"test",
&json!({ "id": 7, "error": { "code": -32000, "message": "boom" } }),
&pending,
&events_tx,
);
let got = rx.await.expect("oneshot resolved");
assert!(got.is_err(), "error frame must surface as Err");
}
/// Inbound event frame with no `id` fans out to subscribers.
#[tokio::test]
async fn inbound_event_broadcasts_to_subscribers() {
let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
let (events_tx, mut events_rx) = broadcast::channel::<EventFrame>(8);
handle_inbound(
"test",
&json!({
"method": "Page.loadEventFired",
"sessionId": "abc",
"params": { "timestamp": 1.23 },
}),
&pending,
&events_tx,
);
let frame = events_rx.recv().await.expect("event received");
assert_eq!(frame.method, "Page.loadEventFired");
assert_eq!(frame.session_id, "abc");
assert_eq!(frame.params, json!({ "timestamp": 1.23 }));
}
/// Orphan responses (no matching pending entry) drop silently.
#[tokio::test]
async fn inbound_orphan_response_drops_silently() {
let pending: PendingMap = Arc::new(Mutex::new(HashMap::new()));
let (events_tx, _) = broadcast::channel::<EventFrame>(8);
handle_inbound(
"test",
&json!({ "id": 999, "result": {} }),
&pending,
&events_tx,
);
assert!(pending.lock().unwrap().is_empty());
}
// `registry.forget_account` / `by_account` are exercised in the
// integration test `cdp_in_process_e2e.rs` against a real Tauri-CEF
// runtime — unit tests cannot synthesize a `Webview<Cef>` without a
// running CEF instance, and the registry map itself is a trivial
// `HashMap` so the wiring tests above already cover all branches.
}
app/src-tauri/src/cdp/mod.rs
+34 -7
View File
@@ -3,28 +3,55 @@
//! Consolidates the CdpConn / target-discovery / notification-shim plumbing
//! that used to be copy-pasted across `discord_scanner`, `whatsapp_scanner`,
//! `slack_scanner`, and `telegram_scanner`. Scanners now call helpers here
//! instead of maintaining their own WebSocket dispatch.
//! instead of maintaining their own dispatch.
//!
//! # Transport
//!
//! Two transports coexist while migration is in progress:
//!
//! - **In-process** — see [`in_process`]. CDP messages travel directly
//! between the Tauri shell and the embedded CEF browser via
//! `Webview::send_dev_tools_message` /
//! `Webview::on_dev_tools_protocol`. No listener, no network surface;
//! any same-UID process is shut out by construction. The per-account
//! session opener (`session.rs`) already uses this path.
//! - **Legacy TCP WebSocket** — driven by [`CDP_HOST`] / [`CDP_PORT`]
//! (`127.0.0.1:19222`). Kept alive for the per-scanner `CdpConn`
//! duplicates in `discord_scanner`, `whatsapp_scanner`,
//! `slack_scanner`, `telegram_scanner`, `wechat_scanner`, and
//! `meet_video` that have not yet migrated. While this path exists,
//! `app/src-tauri/src/lib.rs` still passes
//! `--remote-debugging-port=19222`, which is the same unauthenticated
//! loopback listener it has always been. The flag will be dropped
//! once all scanners cut over to the in-process channel.
pub mod conn;
pub mod in_process;
pub mod input;
pub mod session;
pub mod snapshot;
pub mod target;
pub use conn::CdpConn;
pub use in_process::{
install_for_account, install_for_webview, set_cef_app_handle, CdpRegistry, EventFrame,
WebviewCdpTransport, CALL_TIMEOUT,
};
pub use session::{
placeholder_marker, placeholder_url, spawn_session, target_url_fragment, SpawnedSession,
};
#[allow(unused_imports)] // `Rect` re-export consumed once turn 2 lands; keep stable.
pub use snapshot::{Rect, Snapshot};
pub use target::{
browser_ws_url, connect_and_attach_matching, detach_session, find_page_target_where,
browser_ws_url, conn_for_account, connect_and_attach_matching, detach_session,
find_page_target_where,
};
/// Remote debugging host — matches `--remote-debugging-port=19222` in
/// `lib.rs`. Kept as constants so scanners and the session opener
/// agree. Port was 9222 originally but collided with ollama's
/// `127.0.0.1:9222` listener (silent CDP-attach failure → blank
/// child webviews). If you change either constant, update both.
/// Remote debugging host — historical constant for scanner modules that
/// still use the TCP WebSocket path. The in-process transport in
/// [`in_process`] does not use these constants, but the per-scanner
/// `CdpConn` duplicates in `discord_scanner` / `whatsapp_scanner` /
/// `slack_scanner` / `telegram_scanner` still need them until they are
/// migrated to the shared in-process channel.
pub const CDP_HOST: &str = "127.0.0.1";
pub const CDP_PORT: u16 = 19222;
app/src-tauri/src/cdp/session.rs
+9 -6
View File
@@ -24,7 +24,8 @@ use tokio::task::JoinHandle;
// elapsed check honours `tokio::time::pause()` / `advance()` in unit tests.
use tokio::time::{sleep, Instant};
use super::{browser_ws_url, find_page_target_where, CdpConn};
use super::target::conn_for_account;
use super::{find_page_target_where, CdpConn};
use crate::webview_accounts::{emit_load_finished, redact_url_for_log, RevealTrigger};
/// Backoff between failed attach attempts / reconnects. Intentionally
@@ -413,12 +414,14 @@ async fn run_session_cycle<R: Runtime>(
real_url: &str,
progress_slot: &ProgressSlot,
) -> Result<(), String> {
let browser_ws = browser_ws_url().await?;
let mut cdp = CdpConn::open(&browser_ws).await?;
let mut cdp = conn_for_account(app, account_id)?;
// Account-unique match. The placeholder URL and the real provider URL
// both carry account-specific fragments, so we can use ends_with and
// avoid substring collisions like `…account-abc` vs `…account-abcdef`.
// Account-unique match. Each webview is itself scoped to one
// account, but a webview can host popups (OAuth, attachment
// previews, …) that also surface as `kind=page` targets. The
// placeholder URL and the real provider URL both carry
// account-specific fragments, so we filter explicitly to pick the
// primary frame and ignore popups.
let fragment = target_url_fragment(account_id);
let target =
find_page_target_where(&mut cdp, |t| target_matches_account_url(&t.url, account_id))
app/src-tauri/src/cdp/target.rs
+94 -19
View File
@@ -1,11 +1,19 @@
//! CDP target discovery. Replaces the four hand-rolled copies in the
//! per-provider scanners.
//! CDP target discovery + per-attach helpers.
//!
//! Each CEF webview is its own browser instance with its own DevTools
//! channel (see [`super::in_process`]), so the multi-target multiplexer
//! that used to live in this module has been simplified — there is no
//! more `browser_ws_url()` HTTP discovery and no remote attach. The
//! remaining helpers (`Target.getTargets` walk, `Target.attachToTarget`
//! flatten-attach, detach) still apply because the page itself may
//! contain iframes / workers that the scanners care about.
use std::time::Duration;
use serde_json::{json, Value};
use tauri::{AppHandle, Manager, Runtime};
use super::{CdpConn, CDP_HOST, CDP_PORT};
use super::{in_process::CdpRegistry, CdpConn, CDP_HOST, CDP_PORT};
#[derive(Debug, Clone)]
pub struct CdpTarget {
@@ -15,9 +23,15 @@ pub struct CdpTarget {
pub title: String,
}
/// Discover the browser-level WebSocket endpoint via `/json/version`. All
/// CDP sessions in the app tunnel through this one ws once `flatten: true`
/// is set on attach.
/// Legacy TCP WebSocket discovery — kept for the per-scanner `CdpConn`
/// duplicates that have not yet migrated to the in-process transport.
/// New code paths use [`conn_for_account`] which goes through the
/// in-process channel installed by `webview_accounts::open`.
///
/// Returns the browser-level WebSocket URL by hitting
/// `http://{CDP_HOST}:{CDP_PORT}/json/version`. Requires Chromium to
/// have been spawned with `--remote-debugging-port=<CDP_PORT>` — see
/// `app/src-tauri/src/lib.rs`.
pub async fn browser_ws_url() -> Result<String, String> {
let client = reqwest::Client::builder()
.user_agent("openhuman-cdp/1.0")
@@ -36,8 +50,6 @@ pub async fn browser_ws_url() -> Result<String, String> {
last_err = Some(format!("no webSocketDebuggerUrl in {url}"));
}
Err(e) => {
// Don't bail out — fall through so the next host in the
// candidate list still gets a chance to resolve the ws url.
last_err = Some(format!("parse {url}: {e}"));
}
},
@@ -49,6 +61,9 @@ pub async fn browser_ws_url() -> Result<String, String> {
Err(last_err.unwrap_or_else(|| "failed to resolve CDP websocket URL".to_string()))
}
/// Parse the response of a `Target.getTargets` CDP call into a list of
/// targets. Public so scanners using the lower-level [`CdpConn::call`]
/// can interpret target lists.
pub fn parse_targets(v: &Value) -> Vec<CdpTarget> {
v.get("targetInfos")
.and_then(|x| x.as_array())
@@ -75,19 +90,79 @@ pub fn parse_targets(v: &Value) -> Vec<CdpTarget> {
.unwrap_or_default()
}
/// Full short-lived attach sequence: connect to the browser, find the
/// matching page target, attach with `flatten: true`. Caller gets a ready
/// CdpConn + session id for issuing commands. Caller MUST `detach_session`
/// (or drop the CdpConn entirely) when done so we don't leak sessions.
/// Get a [`CdpConn`] for an account-keyed webview, looking up the
/// pre-installed in-process transport from the [`CdpRegistry`] managed
/// on `app`.
///
/// The predicate must match on per-account fragment + URL prefix so
/// multi-account webviews on the same origin resolve uniquely.
/// On a cache miss, falls back to
/// [`super::in_process::install_for_account`] so a transient install
/// failure during `webview_accounts::open` (logged as a warning by the
/// account-open path, not fatal) doesn't permanently lock the account
/// out of CDP. The install call is idempotent and cheap on the cached
/// path. Still returns `Err` when the webview itself has not yet been
/// created — caller backs off and retries.
pub fn conn_for_account<R: Runtime>(
app: &AppHandle<R>,
account_id: &str,
) -> Result<CdpConn, String> {
let registry = app
.try_state::<CdpRegistry>()
.ok_or_else(|| "CdpRegistry not managed by app".to_string())?;
if let Some(transport) = registry.by_account(account_id) {
return Ok(CdpConn::new(transport));
}
// Retry — the install path is idempotent. The most common cause of
// a cache miss here is an earlier non-fatal `install_for_account`
// failure in `webview_accounts::open` (warn-logged) that left the
// webview alive without a transport.
let transport = super::in_process::install_for_account(account_id)
.map_err(|e| format!("no cdp transport for account {account_id} (install retry: {e})"))?;
Ok(CdpConn::new(transport))
}
/// Full short-lived attach sequence on the account's webview via the
/// in-process channel: look up the [`CdpRegistry`] transport for the
/// given account, find the matching page target via
/// `Target.getTargets`, attach with `flatten: true`. Caller gets a
/// ready `CdpConn` + session id. Caller MUST `detach_session` (or drop
/// the `CdpConn`) when done so the session id doesn't linger inside
/// CEF.
pub async fn connect_and_attach_matching_in_process<R, F>(
app: &AppHandle<R>,
account_id: &str,
pred: F,
) -> Result<(CdpConn, String), String>
where
R: Runtime,
F: Fn(&CdpTarget) -> bool,
{
let mut cdp = conn_for_account(app, account_id)?;
let target = find_page_target_where(&mut cdp, pred).await?;
let attach = cdp
.call(
"Target.attachToTarget",
json!({ "targetId": target.id, "flatten": true }),
None,
)
.await?;
let session = attach
.get("sessionId")
.and_then(|x| x.as_str())
.ok_or_else(|| "attach missing sessionId".to_string())?
.to_string();
Ok((cdp, session))
}
/// Legacy TCP-WS attach helper, kept for the per-scanner `CdpConn`
/// duplicates that still discover targets via the global
/// `Target.getTargets` walk. New code paths use
/// [`connect_and_attach_matching_in_process`].
pub async fn connect_and_attach_matching<F>(pred: F) -> Result<(CdpConn, String), String>
where
F: Fn(&CdpTarget) -> bool,
{
let ws = browser_ws_url().await?;
let mut cdp = CdpConn::open(&ws).await?;
let mut cdp = CdpConn::open_ws(&ws).await?;
let target = find_page_target_where(&mut cdp, pred).await?;
let attach = cdp
.call(
@@ -114,10 +189,10 @@ pub async fn detach_session(cdp: &mut CdpConn, session_id: &str) {
.await;
}
/// Generalised variant — caller supplies the predicate (url-hash marker,
/// title marker, etc). Used by the per-account session opener, which matches
/// on `#openhuman-account-{id}` so multiple webviews on the same origin
/// don't collide.
/// Generalised target search — caller supplies the predicate
/// (url-hash marker, title marker, etc). Used by the per-account
/// session opener, which matches on `#openhuman-account-{id}` so
/// multiple webviews on the same origin don't collide.
pub async fn find_page_target_where<F>(cdp: &mut CdpConn, pred: F) -> Result<CdpTarget, String>
where
F: Fn(&CdpTarget) -> bool,
app/src-tauri/src/lib.rs
+24 -23
View File
@@ -2474,21 +2474,16 @@ pub fn run() {
// mock; `password-store=basic` is the equivalent for the password
// manager. Both are no-ops on Windows/Linux, so safe to always set.
//
// In debug builds we additionally expose the Chrome DevTools
// Protocol on localhost:19222 so every CEF webview can be
// inspected from a regular browser (right-click "Inspect" does
// not propagate to CEF child webviews on macOS). Release builds
// intentionally do NOT open the CDP port — it would let any
// process on the machine drive the embedded WhatsApp/Slack/etc.
// webviews.
//
// The port was 9222 (Chromium's default) but ollama's
// OpenAI-compatible server squats on 127.0.0.1:9222 in some
// installs, which silently broke CDP attach (our client hit
// ollama, the WS handshake failed, child webviews stayed at
// about:blank → black screen). Picked 19222 to dodge that
// collision; if you change it here also update
// `cdp::CDP_PORT` and `whatsapp_scanner::CDP_PORT`.
// CDP attach is migrating to the in-process channel — see
// `app/src-tauri/src/cdp/in_process.rs` and the per-account
// session opener (`cdp/session.rs`). The legacy TCP DevTools
// port is still passed below (search for
// `--remote-debugging-port`) because the per-scanner `CdpConn`
// duplicates in `discord_scanner`, `whatsapp_scanner`,
// `slack_scanner`, `telegram_scanner`, `wechat_scanner`, and
// `meet_video` have not migrated yet. Once they do, the flag
// can be dropped and the unauthenticated same-UID loopback
// listener with it.
//
// NOTE: flags must be prefixed with `--`. The runtime's
// `on_before_command_line_processing` dispatch (in
@@ -2593,14 +2588,13 @@ pub fn run() {
args.push(("--use-fake-ui-for-media-stream", None));
args.push(("--use-file-for-fake-video-capture", Some(path)));
}
// Always expose the CDP port, not just in debug. The webview-accounts
// CDP session opener navigates each embedded provider webview from its
// `about:blank#openhuman-acct-...` placeholder to the real provider URL
// via `Page.navigate`. Without this port available in release builds,
// the CDP client can't attach (`browser_ws_url()` 404s on /json/version),
// the navigation never fires, and the embedded webview stays on
// `about:blank` (blank panel for Telegram / WhatsApp / Slack / Discord).
// Same port the `cdp::CDP_HOST`/`cdp::CDP_PORT` constants expect.
// CDP attach is migrating to in-process. The per-account
// session opener (`cdp/session.rs`) uses the in-process channel
// installed by `webview_accounts::open`. The per-scanner
// duplicates (whatsapp, slack, telegram, wechat, discord,
// meet_video) still reach the embedded browser over the TCP
// loopback DevTools port — once they migrate this flag can be
// dropped and the unauthenticated listener closed for good.
args.push(("--remote-debugging-port", Some("19222")));
let force_gpu_env = std::env::var("OPENHUMAN_FORCE_GPU").ok();
append_platform_cef_gpu_workarounds(
@@ -2705,6 +2699,7 @@ pub fn run() {
std::sync::Mutex::new(Vec::new()),
))
.manage(webview_accounts::WebviewAccountsState::default())
.manage(cdp::CdpRegistry::default())
.manage(notification_settings::NotificationSettingsState::new())
.manage(PendingAppUpdateState::default());
let builder = builder.manage(std::sync::Arc::new(imessage_scanner::ScannerRegistry::new()));
@@ -2722,6 +2717,12 @@ pub fn run() {
let builder = builder.manage(meet_video::frame_bus::MeetVideoFrameBusState::new());
builder
.setup(move |app| {
// Stash the typed CEF `AppHandle` for the in-process CDP
// transport. Lets `cdp::install_for_account` reach the
// concrete `Webview<Cef>` (which `send_dev_tools_message`
// requires) from generic `<R: Runtime>` call sites.
cdp::set_cef_app_handle(app.handle().clone());
#[cfg(windows)]
{
// `register_all` writes HKCU\Software\Classes\openhuman so the
app/src-tauri/src/slack_scanner/mod.rs
+1 -1
View File
@@ -812,7 +812,7 @@ async fn dom_scan_once(
use crate::cdp::CdpConn as CanonicalCdpConn;
let browser_ws = crate::cdp::browser_ws_url().await?;
let mut probe = CanonicalCdpConn::open(&browser_ws).await?;
let mut probe = CanonicalCdpConn::open_ws(&browser_ws).await?;
let targets_v = probe
.call("Target.getTargets", serde_json::json!({}), None)
.await?;
app/src-tauri/src/webview_accounts/mod.rs
+22
View File
@@ -1009,6 +1009,12 @@ fn teardown_account_scanners<R: Runtime>(app: &AppHandle<R>, account_id: &str) {
{
registry.inner().forget(account_id);
}
// Drop the in-process CDP transport for this account so a reopen
// installs a fresh observer instead of re-using the dead one tied
// to the closed webview.
if let Some(registry) = app.try_state::<crate::cdp::CdpRegistry>() {
registry.inner().forget_account(account_id);
}
}
#[derive(Debug, Clone)]
@@ -2420,6 +2426,22 @@ pub async fn webview_account_open<R: Runtime>(
.add_child(builder, initial_position, initial_size)
.map_err(|e| format!("add_child failed: {e}"))?;
// Install the in-process CDP transport so the per-account session
// opener and the provider scanners can attach without the
// `--remote-debugging-port=19222` TCP listener. Failure here is
// logged but not fatal — the scanners retry through
// `cdp::conn_for_account` once the registry is populated, so a
// transient install error just delays first attach by one backoff
// tick.
if let Err(err) = crate::cdp::install_for_account(&args.account_id) {
log::warn!(
"[webview-accounts] cdp install_for_account({}) failed: {} \
(scanners will retry)",
args.account_id,
err
);
}
// Capture the cold-spawn timestamp so the reveal-time log can compute
// spawn -> frontend reveal latency for the Slack first-load investigation.
state