diff --git a/gitbooks/developing/architecture.md b/gitbooks/developing/architecture.md
index 335a06d81..d56a25372 100644
--- a/gitbooks/developing/architecture.md
+++ b/gitbooks/developing/architecture.md
@@ -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.
 
diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs
index 13f8db44a..18c9e7ddb 100644
--- a/src/core/event_bus/events.rs
+++ b/src/core/event_bus/events.rs
@@ -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",
diff --git a/src/openhuman/devices/README.md b/src/openhuman/devices/README.md
index 12cb4d393..cdf62fd23 100644
--- a/src/openhuman/devices/README.md
+++ b/src/openhuman/devices/README.md
@@ -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.
diff --git a/src/openhuman/devices/bus.rs b/src/openhuman/devices/bus.rs
index f2647f72b..ceefa92d7 100644
--- a/src/openhuman/devices/bus.rs
+++ b/src/openhuman/devices/bus.rs
@@ -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();
diff --git a/src/openhuman/devices/rpc.rs b/src/openhuman/devices/rpc.rs
index f4c5e9757..11c3d97dc 100644
--- a/src/openhuman/devices/rpc.rs
+++ b/src/openhuman/devices/rpc.rs
@@ -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
diff --git a/src/openhuman/devices/tunnel_client.rs b/src/openhuman/devices/tunnel_client.rs
index dab93d40e..6ad737820 100644
--- a/src/openhuman/devices/tunnel_client.rs
+++ b/src/openhuman/devices/tunnel_client.rs
@@ -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();
diff --git a/src/openhuman/devices/types.rs b/src/openhuman/devices/types.rs
index 0e58f714e..1f2642f24 100644
--- a/src/openhuman/devices/types.rs
+++ b/src/openhuman/devices/types.rs
@@ -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.
diff --git a/src/openhuman/socket/README.md b/src/openhuman/socket/README.md
index fe33ca8a1..0d0c51134 100644
--- a/src/openhuman/socket/README.md
+++ b/src/openhuman/socket/README.md
@@ -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
 
diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs
index 1b5fb0e4c..71fb80707 100644
--- a/src/openhuman/socket/event_handlers.rs
+++ b/src/openhuman/socket/event_handlers.rs
@@ -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),
diff --git a/src/openhuman/socket/manager.rs b/src/openhuman/socket/manager.rs
index 5d927d2db..ca7cd8e47 100644
--- a/src/openhuman/socket/manager.rs
+++ b/src/openhuman/socket/manager.rs
@@ -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),
diff --git a/src/openhuman/socket/ws_loop.rs b/src/openhuman/socket/ws_loop.rs
index ba9828e47..05a54a9c4 100644
--- a/src/openhuman/socket/ws_loop.rs
+++ b/src/openhuman/socket/ws_loop.rs
@@ -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
 // ---------------------------------------------------------------------------
diff --git a/src/openhuman/socket/ws_loop_tests.rs b/src/openhuman/socket/ws_loop_tests.rs
index 6b9a3a563..f97895234 100644
--- a/src/openhuman/socket/ws_loop_tests.rs
+++ b/src/openhuman/socket/ws_loop_tests.rs
@@ -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();