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

fix(devices): use socket ack for pairing register (#4355)

Browse Source 
Co-authored-by: M3gA-Mind <megamind@mahadao.com>
This commit is contained in:
YOMXXX
2026-06-30 22:09:51 +05:30
committed by GitHub
co-authored by M3gA-Mind
parent b9e25dd425
commit 06e9bd10f5
12 changed files with 302 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitbooks/developing/architecture.md
+4 -4
View File
@@ -367,13 +367,13 @@ iOS App (React + Tauri iOS shell)
|-- CloudHttpTransport fallback via cloud backend API
```
Transport is selected by `ConnectionProfile` stored in secure storage. On pairing, the iOS app stores `{channelId, sessionToken, corePubkey, devicePrivkey}`.
Transport is selected by `ConnectionProfile` stored in secure storage. On pairing, the iOS app stores `{channelId, sessionToken, corePubkey, devicePrivkey}` after the client-side `tunnel:connect` succeeds.
### Pairing flow
1. Desktop: `devices_create_pairing` RPC -> backend issues `{channelId, pairingToken, sessionToken}`.
1. Desktop: `devices_create_pairing` RPC -> backend ACKs `tunnel:register` with `{channelId, pairingToken, pairingExpiresAt}`.
2. Desktop shows QR: `openhuman://pair?cid=<>&pt=<>&cpk=<>&rpc=<>&exp=<>`.
3. iOS scans QR, generates X25519 keypair, connects to backend (`tunnel:connect`, `role:client`).
3. iOS scans QR, generates X25519 keypair, connects to backend (`tunnel:connect`, `role:client`, `pairingToken`).
4. Backend consumes `pairingToken` (single-use) and returns iOS `sessionToken`.
5. X25519 key agreement over `tunnel:frame` -> XChaCha20-Poly1305 symmetric key.
6. Desktop emits `DomainEvent::DevicePaired`; device appears in the Devices panel.
@@ -394,7 +394,7 @@ Transport is selected by `ConnectionProfile` stored in secure storage. On pairin
- Tunnel backend is a blind forwarder -- never sees plaintext payloads.
- `pairingToken` is single-use, TTL'd, hashed at rest on backend.
- `sessionToken` is per-peer and revocable from the desktop Devices panel.
- `sessionToken` is per-client peer and revocable from the desktop Devices panel; the desktop core does not receive a session token during register.
- Speech recognition runs on-device (Apple Speech framework); audio never leaves the device.
- **TODO:** migrate iOS symmetric session key to Keychain for persistence across restarts.
src/core/event_bus/events.rs
+1 -10
View File
@@ -812,13 +812,6 @@ pub enum DomainEvent {
channel_id: String,
payload_b64: String,
},
/// The backend acknowledged `tunnel:register` with channel credentials.
DeviceTunnelRegistered {
channel_id: String,
pairing_token: String,
session_token: String,
},
// ── Memory tree ─────────────────────────────────────────────────────
/// A document (chat batch, email thread, or standalone document) was
/// fully canonicalised and its chunks written to the memory tree.
@@ -1282,8 +1275,7 @@ impl DomainEvent {
| Self::DeviceRevoked { .. }
| Self::DevicePeerOnline { .. }
| Self::DevicePeerOffline { .. }
| Self::DeviceTunnelFrame { .. }
| Self::DeviceTunnelRegistered { .. } => "device",
| Self::DeviceTunnelFrame { .. } => "device",
Self::CompanionSessionStarted { .. }
| Self::CompanionStateChanged { .. }
@@ -1429,7 +1421,6 @@ impl DomainEvent {
Self::DevicePeerOnline { .. } => "DevicePeerOnline",
Self::DevicePeerOffline { .. } => "DevicePeerOffline",
Self::DeviceTunnelFrame { .. } => "DeviceTunnelFrame",
Self::DeviceTunnelRegistered { .. } => "DeviceTunnelRegistered",
Self::CompanionSessionStarted { .. } => "CompanionSessionStarted",
Self::CompanionStateChanged { .. } => "CompanionStateChanged",
Self::CompanionSessionEnded { .. } => "CompanionSessionEnded",
src/openhuman/devices/README.md
+7 -8
View File
@@ -23,8 +23,8 @@ Mobile-device pairing domain. Brokers a secure, end-to-end-encrypted tunnel betw
| `src/openhuman/devices/schemas.rs` | Controller schemas, `all_controller_schemas`/`all_registered_controllers`, and `handle_*` bridges delegating to `rpc.rs`. Mirrors `cron/schemas.rs`. |
| `src/openhuman/devices/store.rs` | SQLite persistence (`paired_devices` table) via the per-call `with_connection` pattern. |
| `src/openhuman/devices/crypto.rs` | `DeviceKeypair` (X25519 keygen, DH, byte round-trip), `TunnelCipher` (XChaCha20-Poly1305 seal/open with a `WINDOW_SIZE`=128 replay window), and base64url helpers. |
| `src/openhuman/devices/tunnel_client.rs` | Emits/parses `tunnel:*` events over the shared `SocketManager`; wire types; the one-shot ack registry (`PENDING_REGISTER`) that resolves `tunnel:register`. Frame cap 64 KB. |
| `src/openhuman/devices/bus.rs` | `DeviceTunnelSubscriber` event handler — drives handshake completion, persistence, peer-status updates, and register-ack resolution. |
| `src/openhuman/devices/tunnel_client.rs` | Emits/parses `tunnel:*` events over the shared `SocketManager`; wire types; `tunnel:register` uses Socket.IO ACK via `SocketManager::emit_with_ack`. Frame cap 64 KB. |
| `src/openhuman/devices/bus.rs` | `DeviceTunnelSubscriber` event handler — drives handshake completion, persistence, and peer-status updates. |
## Public surface
@@ -33,7 +33,7 @@ Re-exported from `mod.rs`:
- `all_devices_controller_schemas` / `all_devices_registered_controllers` (alias for `schemas::all_controller_schemas` / `all_registered_controllers`).
- Types: `CreatePairingResponse`, `ListDevicesResponse`, `PairedDevice`, `PairingSession`, `RevokeDeviceResponse`.
Other notable public items (used across the crate but not re-exported at the domain root): `bus::register_device_tunnel_subscriber`, `crypto::{DeviceKeypair, TunnelCipher, base64url_encode, base64url_decode}`, `tunnel_client::{emit_register, emit_connect, emit_frame, resolve_register_ack, TunnelPeerStatus, TunnelFrame, TunnelRegisterResponse}`.
Other notable public items (used across the crate but not re-exported at the domain root): `bus::register_device_tunnel_subscriber`, `crypto::{DeviceKeypair, TunnelCipher, base64url_encode, base64url_decode}`, `tunnel_client::{emit_register, emit_connect, emit_frame, TunnelPeerStatus, TunnelFrame, TunnelRegisterResponse}`.
## RPC / controllers
@@ -41,7 +41,7 @@ Namespace `devices` (invoked as `openhuman.devices_<function>`):
| Method | Inputs | Output | Behavior |
| --- | --- | --- | --- |
| `devices_create_pairing` | `label?: string` | `CreatePairingResponse` | Registers a channel, generates+persists keypair, emits `tunnel:connect`, returns QR fields. Pairing token expires in 10 min (backend enforces real TTL). |
| `devices_create_pairing` | `label?: string` | `CreatePairingResponse` | Registers a channel via Socket.IO ACK, generates+persists keypair, emits tokenless core `tunnel:connect`, returns QR fields using the backend-provided pairing expiry. |
| `devices_list` | — | `ListDevicesResponse` | Lists non-revoked devices, overlaying live `peer_online` from `PEER_STATUS`. |
| `devices_revoke` | `channel_id: string` | `RevokeDeviceResponse` | Soft-deletes the device, clears all in-memory state for the channel, publishes `DeviceRevoked`. |
@@ -57,14 +57,13 @@ Subscriber registered at startup from `src/core/jsonrpc.rs` via `register_device
- `DevicePeerOnline` / `DevicePeerOffline` → update `PEER_STATUS`.
- `DeviceTunnelFrame` → complete handshake + persist `PairedDevice`.
- `DeviceTunnelRegistered` → resolve the pending `tunnel:register` ack in `tunnel_client`.
**Publishes**:
- `DevicePaired` (after successful handshake + persistence).
- `DeviceRevoked` (from `devices_revoke`).
Note: the `DevicePeerOnline/Offline`, `DeviceTunnelFrame`, and `DeviceTunnelRegistered` events are *originated* by `src/openhuman/socket/event_handlers.rs` (which parses the raw `tunnel:peer-status` / `tunnel:frame` / `tunnel:registered` / `tunnel:evicted` Socket.IO events and re-publishes them as `DomainEvent`s). This domain consumes them; it does not re-publish peer-status itself.
Note: the `DevicePeerOnline/Offline` and `DeviceTunnelFrame` events are *originated* by `src/openhuman/socket/event_handlers.rs` (which parses the raw `tunnel:peer-status` / `tunnel:frame` / `tunnel:evicted` Socket.IO events and re-publishes them as `DomainEvent`s). This domain consumes them; it does not re-publish peer-status itself.
## Persistence
@@ -75,7 +74,7 @@ SQLite DB at `{workspace_dir}/devices/devices.db`, table `paired_devices`:
| `channel_id` | PK; 128-bit base32 channel id. |
| `label` | Human-readable label. |
| `device_pubkey` | Base64url X25519 device public key. |
| `core_session_token_hash` | SHA-256 of the core session token. |
| `core_session_token_hash` | Legacy column name; currently stores a SHA-256 hash of the pairing credential because the backend no longer mints a core session token. |
| `shared_secret_encrypted` | BLOB, currently always written `NULL`. |
| `created_at` / `last_seen_at` | ISO 8601; `last_seen_at` set by `touch_device`. |
| `revoked` | Soft-delete flag; `list_devices` filters `revoked = 0`. |
@@ -107,6 +106,6 @@ Separately, encrypted X25519 private keys are persisted as `enc2:` strings (via
- The `label` persisted on pairing currently falls back to the `channel_id` (the pending session stores no real label field; `PairingSession.channel_id` is used as the label source).
- `devices_revoke` only tears down local + in-memory state. There is **no backend revoke endpoint yet** (TODO referencing PR #709 follow-up); the backend channel is left to expire via the pairing-token TTL.
- `rpc_url` LAN detection uses the UDP "connect to 8.8.8.8" trick to read the local IPv4; port comes from `OPENHUMAN_CORE_RPC_PORT` env (default `7788`). Non-fatal if it fails.
- `tunnel:register` has no native Socket.IO ack support, so `tunnel_client` uses a global one-shot (`PENDING_REGISTER`) resolved by the bus on `tunnel:registered`, with a 10-second timeout.
- `tunnel:register` uses `SocketManager::emit_with_ack` and expects backend ACK shape `{channelId, pairingToken, pairingExpiresAt}` with a 10-second timeout.
- `PairingSession` and the keypair maps are in-memory only (TTL/cleanup deferred to backend semantics); they are cleared on revoke.
- Outbound `tunnel:frame` payloads are capped at 64 KB; callers are expected to stay ≤ 100 frames/s.
src/openhuman/devices/bus.rs
+22 -40
View File
@@ -6,7 +6,6 @@
//! - Completing the X25519 handshake when the device sends its pubkey.
//! - Persisting the `PairedDevice` record after a successful handshake.
//! - Publishing `DomainEvent::DevicePaired / DevicePeerOnline / DevicePeerOffline`.
//! - Resolving `tunnel:registered` acks for `tunnel_client`.
use std::sync::{Arc, OnceLock};
@@ -18,9 +17,7 @@ use crate::openhuman::devices::rpc::{
ACTIVE_CIPHERS, PEER_STATUS, PENDING_KEYPAIRS, PENDING_SESSIONS,
};
use crate::openhuman::devices::store;
use crate::openhuman::devices::tunnel_client::{
emit_frame, resolve_register_ack, TunnelRegisterResponse,
};
use crate::openhuman::devices::tunnel_client::emit_frame;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
@@ -86,13 +83,6 @@ impl EventHandler for DeviceTunnelSubscriber {
} => {
handle_tunnel_frame(channel_id, payload_b64).await;
}
DomainEvent::DeviceTunnelRegistered {
channel_id,
pairing_token,
session_token,
} => {
handle_registered(channel_id, pairing_token, session_token);
}
_ => {}
}
}
@@ -333,22 +323,28 @@ async fn handle_tunnel_frame(channel_id: &str, payload_b64: &str) {
}
}
// Persist the paired device.
let label = PENDING_SESSIONS
.lock()
.unwrap()
.get(channel_id)
.map(|s| s.channel_id.clone()) // use channel_id as fallback label
.unwrap_or_else(|| channel_id.to_string());
// Persist the paired device. The pending session holds both the label
// source and the pairing credential. Fail closed when it is absent rather
// than persisting a hash of an empty token into the legacy column: a frame
// with no pending session has no valid pairing context (CodeRabbit #4355).
let (label, pairing_token) = {
let sessions = PENDING_SESSIONS.lock().unwrap();
match sessions.get(channel_id) {
Some(session) => (session.channel_id.clone(), session.pairing_token.clone()),
None => {
log::warn!(
"[devices/bus] no pending session for channel_id={} — skipping persist (fail closed)",
channel_id
);
return;
}
}
};
let session_token_hash = hash_session_token(
&PENDING_SESSIONS
.lock()
.unwrap()
.get(channel_id)
.map(|s| s.core_session_token.clone())
.unwrap_or_default(),
);
// Legacy DB column name is `core_session_token_hash`; the backend no
// longer mints a core session token, so persist a hash of the pairing
// credential for this channel.
let session_token_hash = hash_session_token(&pairing_token);
// Load config from global env (best-effort; pairing persists even if config
// loading is slow — the UI will see the device on next list call).
@@ -612,20 +608,6 @@ async fn emit_tunnel_response(
}
}
/// Resolve the pending `tunnel:register` ack in `tunnel_client`.
fn handle_registered(channel_id: &str, pairing_token: &str, session_token: &str) {
log::debug!(
"[devices/bus] tunnel:registered channel_id={} token_len={}",
channel_id,
pairing_token.len()
);
resolve_register_ack(TunnelRegisterResponse {
channel_id: channel_id.to_string(),
pairing_token: pairing_token.to_string(),
session_token: session_token.to_string(),
});
}
fn hash_session_token(token: &str) -> String {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
src/openhuman/devices/rpc.rs
+25 -25
View File
@@ -13,8 +13,6 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use chrono::Utc;
use crate::openhuman::config::Config;
use crate::openhuman::devices::crypto::{
base64url_decode, base64url_encode, DeviceKeypair, TunnelCipher,
@@ -63,7 +61,7 @@ pub(crate) static ACTIVE_CIPHERS: once_cell::sync::Lazy<
/// `openhuman.devices_create_pairing`
///
/// 1. Calls `tunnel:register` on the shared socket — backend returns
/// `{channelId, pairingToken, sessionToken}`.
/// `{channelId, pairingToken, pairingExpiresAt}` via Socket.IO ACK.
/// 2. Generates an X25519 keypair and persists the private half in-memory.
/// 3. Emits `tunnel:connect` with `role:"core"` so the core starts listening.
/// 4. Detects the local LAN IP for the optional direct fast-path `rpc_url`.
@@ -121,8 +119,31 @@ pub async fn devices_create_pairing(
.unwrap()
.insert(reg.channel_id.clone(), Arc::new(keypair));
// Best-effort LAN URL detection (non-fatal if it fails).
let rpc_url = detect_lan_rpc_url();
if let Some(ref url) = rpc_url {
log::debug!("[devices/rpc] LAN rpc_url detected: {}", url);
}
let expires_at = reg.pairing_expires_at.clone();
// Insert the pending session BEFORE opening the tunnel: `emit_connect`
// starts `tunnel:frame` handling, and `bus.rs` now derives the persisted
// pairing credential from this map, so a fast inbound frame must never race
// ahead of the entry (CodeRabbit #4355).
PENDING_SESSIONS.lock().unwrap().insert(
reg.channel_id.clone(),
PairingSession {
channel_id: reg.channel_id.clone(),
pairing_token: reg.pairing_token.clone(),
core_pubkey: core_pubkey.clone(),
rpc_url: rpc_url.clone(),
expires_at: expires_at.clone(),
},
);
// Connect as "core" role to start listening on this channel.
tunnel_client::emit_connect(&reg.channel_id, &reg.session_token)
tunnel_client::emit_connect(&reg.channel_id)
.await
.map_err(|e| {
log::error!("[devices/rpc] tunnel:connect failed: {e}");
@@ -134,27 +155,6 @@ pub async fn devices_create_pairing(
reg.channel_id
);
// Best-effort LAN URL detection (non-fatal if it fails).
let rpc_url = detect_lan_rpc_url();
if let Some(ref url) = rpc_url {
log::debug!("[devices/rpc] LAN rpc_url detected: {}", url);
}
// Pairing token expires in 10 minutes (backend enforces the real TTL).
let expires_at = (Utc::now() + chrono::Duration::minutes(10)).to_rfc3339();
PENDING_SESSIONS.lock().unwrap().insert(
reg.channel_id.clone(),
PairingSession {
channel_id: reg.channel_id.clone(),
pairing_token: reg.pairing_token.clone(),
core_session_token: reg.session_token.clone(),
core_pubkey: core_pubkey.clone(),
rpc_url: rpc_url.clone(),
expires_at: expires_at.clone(),
},
);
log::info!(
"[devices/rpc] devices_create_pairing done channel_id={}",
reg.channel_id
src/openhuman/devices/tunnel_client.rs
+44 -76
View File
@@ -23,15 +23,15 @@ pub struct TunnelRegisterPayload {
pub role: String, // always "core"
}
/// Response from `tunnel:register` emitted back by the backend.
/// Response from the `tunnel:register` ACK callback.
#[derive(Debug, Clone, Deserialize)]
pub struct TunnelRegisterResponse {
#[serde(rename = "channelId")]
pub channel_id: String,
#[serde(rename = "pairingToken")]
pub pairing_token: String,
#[serde(rename = "sessionToken")]
pub session_token: String,
#[serde(rename = "pairingExpiresAt")]
pub pairing_expires_at: String,
}
/// Payload emitted as `tunnel:connect` to join a channel.
@@ -40,10 +40,6 @@ pub struct TunnelConnectPayload {
#[serde(rename = "channelId")]
pub channel_id: String,
pub role: String, // "core" or "client"
#[serde(rename = "sessionToken", skip_serializing_if = "Option::is_none")]
pub session_token: Option<String>,
#[serde(rename = "pairingToken", skip_serializing_if = "Option::is_none")]
pub pairing_token: Option<String>,
}
/// Inbound `tunnel:peer-status` event payload.
@@ -75,60 +71,46 @@ struct TunnelFrameEmit<'a> {
// Tunnel operations
// ---------------------------------------------------------------------------
/// Emit `tunnel:register` on the shared socket and parse the response.
///
/// The backend returns `{channelId, pairingToken, sessionToken}` via the
/// same socket in a `tunnel:registered` ack. Since the existing `SocketManager`
/// does not support request/response acks over the raw WebSocket, we use
/// a one-shot `tokio::sync::oneshot` channel registered in a global pending-ack
/// map and resolved by `devices::bus` when the `tunnel:registered` event arrives.
///
/// For v1 this is simplified: we emit the registration event and expect the
/// caller (rpc.rs) to await the response via the in-process ack mechanism.
/// Emit `tunnel:register` on the shared socket and parse the ACK response.
pub async fn emit_register() -> Result<TunnelRegisterResponse, String> {
log::debug!("[devices/tunnel] emit_register: sending tunnel:register");
let mgr = global_socket_manager()
.ok_or_else(|| "[devices/tunnel] SocketManager not initialized".to_string())?;
let payload = json!({ "role": "core" });
// Register a pending ack before emitting to avoid a race.
let rx = PENDING_REGISTER.register_pending();
mgr.emit("tunnel:register", payload)
let ack = mgr
.emit_with_ack(
"tunnel:register",
payload,
std::time::Duration::from_secs(10),
)
.await
.map_err(|e| format!("[devices/tunnel] emit tunnel:register failed: {e}"))?;
log::debug!("[devices/tunnel] tunnel:register emitted, awaiting response");
// Wait up to 10 s for the backend ack.
tokio::time::timeout(std::time::Duration::from_secs(10), rx)
.await
.map_err(|_| "[devices/tunnel] timeout waiting for tunnel:registered".to_string())?
.map_err(|_| "[devices/tunnel] ack channel dropped".to_string())
serde_json::from_value::<TunnelRegisterResponse>(ack)
.map_err(|e| format!("[devices/tunnel] parse tunnel:register ack failed: {e}"))
}
/// Emit `tunnel:connect` to start listening on a channel as `role:"core"`.
pub async fn emit_connect(channel_id: &str, session_token: &str) -> Result<(), String> {
log::debug!(
"[devices/tunnel] emit_connect channel_id={} token_len={}",
channel_id,
session_token.len()
);
pub async fn emit_connect(channel_id: &str) -> Result<(), String> {
log::debug!("[devices/tunnel] emit_connect channel_id={channel_id}");
let mgr = global_socket_manager()
.ok_or_else(|| "[devices/tunnel] SocketManager not initialized".to_string())?;
let payload = json!({
"channelId": channel_id,
"role": "core",
"sessionToken": session_token,
});
let payload = build_core_connect_payload(channel_id);
mgr.emit("tunnel:connect", payload)
.await
.map_err(|e| format!("[devices/tunnel] emit tunnel:connect failed: {e}"))
}
fn build_core_connect_payload(channel_id: &str) -> serde_json::Value {
json!({
"channelId": channel_id,
"role": "core",
})
}
/// Emit a `tunnel:frame` carrying an encrypted payload for the peer.
///
/// `payload_b64` is the base64url-encoded sealed frame from `TunnelCipher::seal`.
@@ -152,46 +134,32 @@ pub async fn emit_frame(channel_id: &str, payload_b64: &str) -> Result<(), Strin
.map_err(|e| format!("[devices/tunnel] emit tunnel:frame failed: {e}"))
}
/// Resolve a pending `tunnel:register` ack when the backend responds.
///
/// Called by `socket::event_handlers` when it receives `tunnel:registered`.
pub fn resolve_register_ack(response: TunnelRegisterResponse) {
log::debug!(
"[devices/tunnel] resolving tunnel:registered ack channel_id={}",
response.channel_id
);
PENDING_REGISTER.resolve(response);
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
// ---------------------------------------------------------------------------
// One-shot ack registry for tunnel:register
// ---------------------------------------------------------------------------
#[test]
fn tunnel_register_response_accepts_backend_ack_shape_without_session_token() {
let response: TunnelRegisterResponse = serde_json::from_value(json!({
"channelId": "ch_123",
"pairingToken": "pt_123",
"pairingExpiresAt": "2026-06-30T15:00:00Z"
}))
.expect("backend register ack shape should parse");
use std::sync::Mutex;
use tokio::sync::oneshot;
struct PendingRegisterAck {
tx: Mutex<Option<oneshot::Sender<TunnelRegisterResponse>>>,
}
impl PendingRegisterAck {
const fn new() -> Self {
Self {
tx: Mutex::new(None),
}
assert_eq!(response.channel_id, "ch_123");
assert_eq!(response.pairing_token, "pt_123");
assert_eq!(response.pairing_expires_at, "2026-06-30T15:00:00Z");
}
fn register_pending(&self) -> oneshot::Receiver<TunnelRegisterResponse> {
let (tx, rx) = oneshot::channel();
*self.tx.lock().unwrap() = Some(tx);
rx
}
#[test]
fn build_core_connect_payload_omits_session_token_for_core_role() {
let payload = build_core_connect_payload("ch_123");
fn resolve(&self, response: TunnelRegisterResponse) {
if let Some(tx) = self.tx.lock().unwrap().take() {
let _ = tx.send(response);
}
assert_eq!(payload["channelId"], "ch_123");
assert_eq!(payload["role"], "core");
assert!(payload.get("sessionToken").is_none());
assert!(payload.get("pairingToken").is_none());
}
}
static PENDING_REGISTER: PendingRegisterAck = PendingRegisterAck::new();
src/openhuman/devices/types.rs
-2
View File
@@ -33,8 +33,6 @@ pub struct PairingSession {
pub channel_id: String,
/// Base64url pairing token (single-use, TTL'd, hashed at rest on backend).
pub pairing_token: String,
/// Core's reconnect credential for this channel (hashed at rest in SQLite).
pub core_session_token: String,
/// Base64url-encoded X25519 public key generated for this pairing.
pub core_pubkey: String,
/// Optional LAN URL for the direct HTTP fast path.
src/openhuman/socket/README.md
+4 -5
View File
@@ -18,8 +18,8 @@ Persistent, Rust-native Socket.IO client to the OpenHuman backend. The `socket`
| File | Role |
| --- | --- |
| `src/openhuman/socket/mod.rs` | Module docstring + exports only. Re-exports `SocketManager`, `global_socket_manager`, `set_global_socket_manager`, and the `all_socket_controller_schemas` / `all_socket_registered_controllers` pair. |
| `src/openhuman/socket/manager.rs` | `SocketManager` handle + `SharedState`; global `OnceLock` accessor; `connect` / `connect_with_provider` / `disconnect` / `emit` / `get_state`; spawns the background `ws_loop`. Holds emit/shutdown channels and the loop join handle. |
| `src/openhuman/socket/ws_loop.rs` | The background reconnection loop and a single connection attempt: Engine.IO/Socket.IO handshake, redirect-following connect, ping-timeout deadline, backoff, invalid-token decision logic, failure-escalation logging. |
| `src/openhuman/socket/manager.rs` | `SocketManager` handle + `SharedState`; global `OnceLock` accessor; `connect` / `connect_with_provider` / `disconnect` / `emit` / `emit_with_ack` / `get_state`; spawns the background `ws_loop`. Holds emit/shutdown channels, ACK waiters, and the loop join handle. |
| `src/openhuman/socket/ws_loop.rs` | The background reconnection loop and a single connection attempt: Engine.IO/Socket.IO handshake, Socket.IO ACK packet dispatch, redirect-following connect, ping-timeout deadline, backoff, invalid-token decision logic, failure-escalation logging. |
| `src/openhuman/socket/event_handlers.rs` | Inbound SIO event dispatch (`handle_sio_event`), SIO frame parsing (`parse_sio_event`), outbound frame helper (`emit_via_channel`). Maps event names → `DomainEvent` publishes. Redacts payload content from logs. |
| `src/openhuman/socket/token_provider.rs` | `TokenProvider` type alias + `static_token_provider`, `token_provider_from_config`, and `is_invalid_token_error` (strict double-anchor matcher). |
| `src/openhuman/socket/schemas.rs` | Controller schemas + RPC handlers for the `socket` namespace. |
@@ -28,7 +28,7 @@ Persistent, Rust-native Socket.IO client to the OpenHuman backend. The `socket`
## Public surface
- `SocketManager` — the connection handle. Key methods: `new`, `connect(url, token)`, `connect_with_provider(url, provider)`, `disconnect`, `emit(event, data)`, `get_state() -> SocketState`, `is_connected`, `set_webhook_router` / `webhook_router`.
- `SocketManager` — the connection handle. Key methods: `new`, `connect(url, token)`, `connect_with_provider(url, provider)`, `disconnect`, `emit(event, data)`, `emit_with_ack(event, data, timeout)`, `get_state() -> SocketState`, `is_connected`, `set_webhook_router` / `webhook_router`.
- `global_socket_manager() -> Option<&'static Arc<SocketManager>>` and `set_global_socket_manager(Arc<SocketManager>)` — the process-global singleton (set once at bootstrap).
- `all_socket_controller_schemas` / `all_socket_registered_controllers` — controller-registry exports wired into `src/core/all.rs`.
@@ -58,7 +58,6 @@ All handlers go through `require_manager()` and error with `"SocketManager not i
| `composio:trigger` | `ComposioTriggerReceived { toolkit, trigger, metadata_id, metadata_uuid, payload }` | composio |
| `tunnel:peer-status` | `DevicePeerOnline` / `DevicePeerOffline` | devices |
| `tunnel:frame` | `DeviceTunnelFrame` | devices |
| `tunnel:registered` | `DeviceTunnelRegistered` | devices |
| `tunnel:evicted` | `DevicePeerOffline` | devices |
| `*:message` (suffix match) | `ChannelInboundMessage { event_name, channel, message, sender, reply_target, thread_ts, raw_data }` | channels |
@@ -66,7 +65,7 @@ This module is a **publisher only** — it owns no `bus.rs` / `EventHandler` imp
## Persistence
None of its own. State (`status`, `socket_id`, `error`, attached `WebhookRouter`) lives in-memory in `SharedState` behind `parking_lot::RwLock`. The session token is read on demand from the profile store via `crate::api::jwt::get_session_token` (live-refresh path); there is no `store.rs`.
None of its own. State (`status`, `socket_id`, `error`, attached `WebhookRouter`, pending ACK waiters) lives in-memory in `SharedState`. The session token is read on demand from the profile store via `crate::api::jwt::get_session_token` (live-refresh path); there is no `store.rs`.
## Dependencies
src/openhuman/socket/event_handlers.rs
+1 -28
View File
@@ -207,34 +207,6 @@ pub(super) fn handle_sio_event(
}
}
}
// Device tunnel — backend ack for tunnel:register.
"tunnel:registered" => {
log::info!("[socket] tunnel:registered received");
let channel_id = data
.get("channelId")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let pairing_token = data
.get("pairingToken")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let session_token = data
.get("sessionToken")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if !channel_id.is_empty() {
publish_global(DomainEvent::DeviceTunnelRegistered {
channel_id,
pairing_token,
session_token,
});
} else {
log::warn!("[socket] tunnel:registered missing channelId");
}
}
// Device tunnel — backend evicted the channel (TTL / server restart).
"tunnel:evicted" => {
let channel_id = data
@@ -537,6 +509,7 @@ mod tests {
fn make_shared() -> Arc<SharedState> {
Arc::new(SharedState {
webhook_router: RwLock::new(None),
ack_registry: super::super::manager::AckRegistry::default(),
status: RwLock::new(ConnectionStatus::Disconnected),
socket_id: RwLock::new(None),
error: RwLock::new(None),
src/openhuman/socket/manager.rs
+134 -6
View File
@@ -9,11 +9,15 @@
//! - Connection state logging for observability
//! - Automatic reconnection with exponential backoff
use std::sync::{Arc, OnceLock};
use std::collections::HashMap;
use std::sync::{
atomic::{AtomicU64, Ordering},
Arc, OnceLock,
};
use parking_lot::RwLock;
use parking_lot::{Mutex, RwLock};
use serde_json::json;
use tokio::sync::{mpsc, watch};
use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::Duration;
use crate::api::models::socket::{ConnectionStatus, SocketState};
@@ -48,6 +52,8 @@ pub fn global_socket_manager() -> Option<&'static Arc<SocketManager>> {
pub(super) struct SharedState {
/// Router for delivering incoming webhooks to skills.
pub(super) webhook_router: RwLock<Option<Arc<WebhookRouter>>>,
/// Pending Socket.IO ACK callbacks keyed by outbound ack id.
pub(super) ack_registry: AckRegistry,
/// Current connection status.
pub(super) status: RwLock<ConnectionStatus>,
/// Socket ID assigned by the server.
@@ -58,6 +64,46 @@ pub(super) struct SharedState {
pub(super) error: RwLock<Option<String>>,
}
pub(super) struct AckRegistry {
next_id: AtomicU64,
pending: Mutex<HashMap<u64, oneshot::Sender<serde_json::Value>>>,
}
impl Default for AckRegistry {
fn default() -> Self {
Self {
next_id: AtomicU64::new(1),
pending: Mutex::new(HashMap::new()),
}
}
}
impl AckRegistry {
pub(super) fn register(&self) -> (u64, oneshot::Receiver<serde_json::Value>) {
let ack_id = self.next_id.fetch_add(1, Ordering::Relaxed);
let (tx, rx) = oneshot::channel();
self.pending.lock().insert(ack_id, tx);
(ack_id, rx)
}
pub(super) fn resolve(&self, ack_id: u64, data: serde_json::Value) -> bool {
if let Some(tx) = self.pending.lock().remove(&ack_id) {
let _ = tx.send(data);
true
} else {
false
}
}
pub(super) fn remove(&self, ack_id: u64) {
self.pending.lock().remove(&ack_id);
}
pub(super) fn cancel_all(&self) {
self.pending.lock().clear();
}
}
// ---------------------------------------------------------------------------
// SocketManager
// ---------------------------------------------------------------------------
@@ -85,6 +131,7 @@ impl SocketManager {
Self {
shared: Arc::new(SharedState {
webhook_router: RwLock::new(None),
ack_registry: AckRegistry::default(),
status: RwLock::new(ConnectionStatus::Disconnected),
socket_id: RwLock::new(None),
error: RwLock::new(None),
@@ -228,6 +275,7 @@ impl SocketManager {
if let Some(tx) = self.shutdown_tx.lock().await.take() {
let _ = tx.send(true);
}
self.shared.ack_registry.cancel_all();
self.emit_tx.lock().await.take();
if let Some(handle) = self.loop_handle.lock().await.take() {
let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
@@ -243,14 +291,60 @@ impl SocketManager {
/// Emit a Socket.IO event to the server.
pub async fn emit(&self, event: &str, data: serde_json::Value) -> Result<(), String> {
if let Some(ref tx) = *self.emit_tx.lock().await {
let payload =
serde_json::to_string(&json!([event, data])).map_err(|e| format!("{e}"))?;
let msg = format!("42{}", payload);
let msg = encode_sio_event(event, data, None)?;
tx.send(msg).map_err(|_| "Socket not connected".to_string())
} else {
Err("Not connected".to_string())
}
}
/// Emit a Socket.IO event and wait for the backend ACK callback.
pub async fn emit_with_ack(
&self,
event: &str,
data: serde_json::Value,
timeout: Duration,
) -> Result<serde_json::Value, String> {
let tx = self
.emit_tx
.lock()
.await
.clone()
.ok_or_else(|| "Not connected".to_string())?;
let (ack_id, ack_rx) = self.shared.ack_registry.register();
let msg = encode_sio_event(event, data, Some(ack_id))?;
if let Err(e) = tx.send(msg) {
self.shared.ack_registry.remove(ack_id);
return Err(format!("Socket not connected: {e}"));
}
log::debug!("[socket] emit_with_ack sent event={event} ack_id={ack_id}");
match tokio::time::timeout(timeout, ack_rx).await {
Ok(Ok(data)) => {
log::debug!("[socket] emit_with_ack resolved event={event} ack_id={ack_id}");
Ok(data)
}
Ok(Err(_)) => Err(format!(
"Socket ack channel dropped for event {event} ack_id={ack_id}"
)),
Err(_) => {
self.shared.ack_registry.remove(ack_id);
Err(format!(
"Socket ack timeout for event {event} ack_id={ack_id}"
))
}
}
}
}
fn encode_sio_event(
event: &str,
data: serde_json::Value,
ack_id: Option<u64>,
) -> Result<String, String> {
let payload = serde_json::to_string(&json!([event, data])).map_err(|e| format!("{e}"))?;
let ack = ack_id.map(|id| id.to_string()).unwrap_or_default();
Ok(format!("42{ack}{payload}"))
}
impl Default for SocketManager {
@@ -336,6 +430,38 @@ mod tests {
assert_eq!(err, "Not connected");
}
#[tokio::test]
async fn emit_with_ack_without_connection_errors_without_waiting() {
let mgr = SocketManager::new();
let err = mgr
.emit_with_ack("test.event", json!({"k":"v"}), Duration::from_secs(30))
.await
.unwrap_err();
assert_eq!(err, "Not connected");
}
#[tokio::test]
async fn emit_with_ack_uses_emit_queue_while_connecting() {
let mgr = SocketManager::new();
let (tx, mut rx) = mpsc::unbounded_channel::<String>();
*mgr.emit_tx.lock().await = Some(tx);
*mgr.shared.status.write() = ConnectionStatus::Connecting;
let result = mgr
.emit_with_ack("test.event", json!({"k": "v"}), Duration::from_millis(10))
.await;
let queued = rx
.try_recv()
.unwrap_or_else(|_| panic!("expected queued ACK emit, got result={result:?}"));
assert_eq!(queued, r#"421["test.event",{"k":"v"}]"#);
let err = result.unwrap_err();
assert!(
err.starts_with("Socket ack timeout for event test.event ack_id=1"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn disconnect_on_fresh_manager_is_idempotent() {
let mgr = SocketManager::new();
@@ -349,6 +475,7 @@ mod tests {
fn emit_state_change_is_safe_to_call_on_empty_shared() {
let shared = SharedState {
webhook_router: RwLock::new(None),
ack_registry: AckRegistry::default(),
status: RwLock::new(ConnectionStatus::Connecting),
socket_id: RwLock::new(None),
error: RwLock::new(None),
@@ -361,6 +488,7 @@ mod tests {
fn emit_server_event_is_safe_without_subscribers() {
let shared = SharedState {
webhook_router: RwLock::new(None),
ack_registry: AckRegistry::default(),
status: RwLock::new(ConnectionStatus::Connected),
socket_id: RwLock::new(Some("x".into())),
error: RwLock::new(None),
src/openhuman/socket/ws_loop.rs
+39
View File
@@ -15,6 +15,8 @@ use crate::api::models::socket::ConnectionStatus;
use crate::openhuman::util::utf8_safe_prefix_at_byte_boundary;
use super::event_handlers::{handle_sio_event, parse_sio_event};
#[cfg(test)]
use super::manager::AckRegistry;
use super::manager::{emit_state_change, SharedState};
use super::token_provider::{is_invalid_token_error, TokenProvider};
use super::types::{ConnectionOutcome, WsStream};
@@ -155,6 +157,13 @@ pub(super) async fn ws_loop(
)
.await;
// The connection attempt has ended (lost, failed, or shutdown), so any
// in-flight `emit_with_ack` waiter can never receive its ACK now. Cancel
// them here — covering server-driven disconnects (`Lost`) and the
// session-expired escalation below — not just explicit
// `SocketManager::disconnect()` (CodeRabbit #4355).
shared.ack_registry.cancel_all();
match outcome {
ConnectionOutcome::Shutdown => {
log::info!("[socket] Clean shutdown");
@@ -681,6 +690,21 @@ fn handle_sio_packet(
);
}
}
b'3' => {
// Socket.IO ACK: 3<ackId>[ackPayload]
if let Some((ack_id, data)) = parse_sio_ack(&text[1..]) {
if shared.ack_registry.resolve(ack_id, data) {
log::debug!("[socket] SIO ACK resolved ack_id={ack_id}");
} else {
log::warn!("[socket] SIO ACK had no pending waiter ack_id={ack_id}");
}
} else {
log::warn!(
"[socket] Failed to parse SIO ACK: {}",
utf8_safe_prefix_at_byte_boundary(text, 80)
);
}
}
b'0' => {
// Socket.IO CONNECT (re-ack during reconnection) — update sid
log::debug!("[socket] SIO CONNECT re-ack");
@@ -718,6 +742,21 @@ fn handle_sio_packet(
}
}
fn parse_sio_ack(text: &str) -> Option<(u64, serde_json::Value)> {
let json_start = text.find('[')?;
if json_start == 0 {
return None;
}
let ack_id = text[..json_start].parse::<u64>().ok()?;
let mut args: Vec<serde_json::Value> = serde_json::from_str(&text[json_start..]).ok()?;
let data = if args.len() == 1 {
args.pop().unwrap_or(serde_json::Value::Null)
} else {
serde_json::Value::Array(args)
};
Some((ack_id, data))
}
// ---------------------------------------------------------------------------
// Redirect-following connect
// ---------------------------------------------------------------------------
src/openhuman/socket/ws_loop_tests.rs
+21
View File
@@ -8,6 +8,7 @@ use crate::openhuman::socket::token_provider::{is_invalid_token_error, static_to
fn make_shared() -> Arc<SharedState> {
Arc::new(SharedState {
webhook_router: RwLock::new(None),
ack_registry: AckRegistry::default(),
status: RwLock::new(ConnectionStatus::Connected),
socket_id: RwLock::new(None),
error: RwLock::new(None),
@@ -196,6 +197,26 @@ fn handle_sio_packet_event_dispatches_to_event_handler() {
assert_eq!(*shared.status.read(), ConnectionStatus::Connected);
}
#[test]
fn parse_sio_ack_returns_id_and_single_payload_value() {
let (ack_id, data) = parse_sio_ack(r#"7[{"channelId":"ch_123","pairingToken":"pt_123"}]"#)
.expect("valid ack packet");
assert_eq!(ack_id, 7);
assert_eq!(data, json!({"channelId":"ch_123","pairingToken":"pt_123"}));
}
#[tokio::test]
async fn handle_sio_packet_ack_resolves_pending_ack() {
let shared = make_shared();
let (tx, _rx) = mpsc::unbounded_channel::<String>();
let (ack_id, ack_rx) = shared.ack_registry.register();
handle_sio_packet(&format!(r#"3{ack_id}[{{"ok":true}}]"#), &tx, &shared);
let data = ack_rx.await.expect("ack should resolve");
assert_eq!(data, json!({"ok": true}));
}
#[test]
fn handle_sio_packet_event_with_unparseable_payload_is_logged_only() {
let shared = make_shared();