feat(remote): client-side /acp liveness — heartbeat, resume, turn recovery

console/src/log.ts

@@ -62,7 +62,7 @@ export async function bindBackend(activity: Pane, mcp: Pane): Promise<void> {
   const tauri = (globalThis as { __TAURI__?: TauriEventGlobal }).__TAURI__;
   const listen = tauri?.event?.listen;
   if (!listen) {
-    activity.push({ cls: "lv-info", tag: "INFO", msg: "browser build — no core (fixtures)" });
+    activity.push({ cls: "lv-info", tag: "INFO", msg: "app: browser build — no core (fixtures)" });
     return;
   }
   await listen<{ level: Level; msg: string }>("app-log", (e) => {

console/src/main.ts

@@ -161,7 +161,7 @@ async function tick(): Promise<void> {
     prunePending(deployments);
     renderRoster(roster, deployments, pendingKeys());
     if (lastError) {
-      note("info", `roster recovered — ${deployments.length} deployment(s)`);
+      note("info", `roster: recovered — ${deployments.length} deployment(s)`);
       lastError = "";
     }
     if (pollStatus) {
@@ -202,7 +202,7 @@ async function refreshConfig(): Promise<void> {
     fleetConfig = await source.fleetConfig();
     renderFleetConfig(configEl, fleetConfig, activeFleet);
   } catch (e) {
-    note("error", `fleet config: ${errText(e)}`);
+    note("error", `config: fleet load failed — ${errText(e)}`);
     fleetConfig = null;
     renderFleetConfig(configEl, null, activeFleet);
   }
@@ -217,7 +217,7 @@ async function refreshRemote(): Promise<void> {
     remoteConfig = await source.remoteConfig();
     renderRemote(remoteEl, remoteConfig);
   } catch (e) {
-    note("error", `remote config: ${errText(e)}`);
+    note("error", `config: remote load failed — ${errText(e)}`);
     remoteConfig = null;
     renderRemote(remoteEl, null);
   }
@@ -236,7 +236,7 @@ function selectFleet(name: string): void {
   activeCluster = fleet.cluster;
   activeMembers = fleet.members;
   if (clusterLabel) clusterLabel.textContent = `${activeFleet} · ${activeCluster}`;
-  note("info", `switched to fleet "${activeFleet}" (cluster "${activeCluster}")`);
+  note("info", `config: switched to fleet "${activeFleet}" (cluster "${activeCluster}")`);
   if (configEl) renderFleetConfig(configEl, fleetConfig, activeFleet);
   void refreshIdentity();
   void tick();
@@ -295,12 +295,12 @@ async function saveEditor(): Promise<void> {
     if (editorTarget === "remote") {
       remoteConfig = await source.writeRemoteConfig(text);
       if (remoteEl) renderRemote(remoteEl, remoteConfig);
-      note("info", "remote config saved");
+      note("info", "config: remote saved");
       closeEditor();
     } else {
       fleetConfig = await source.writeFleetConfig(text);
       if (configEl) renderFleetConfig(configEl, fleetConfig, activeFleet);
-      note("info", "fleet config saved");
+      note("info", "config: fleet saved");
       closeEditor();
       // A binding change may alter the active fleet's credential — re-observe.
       void refreshIdentity();
@@ -336,13 +336,13 @@ async function remoteAction(kind: "connect" | "disconnect"): Promise<void> {
   try {
     if (kind === "connect") {
       await source.remoteConnect();
-      note("info", "activating remote connection…");
+      note("info", "remote: activating…");
     } else {
       await source.remoteDisconnect();
-      note("info", "remote connection deactivated");
+      note("info", "remote: deactivated");
     }
   } catch (e) {
-    note("error", `remote ${kind}: ${errText(e)}`);
+    note("error", `remote: ${kind} failed — ${errText(e)}`);
   }
   void refreshRemote();
 }
@@ -477,7 +477,7 @@ async function stopTurn(): Promise<void> {
     await source.agentCancel();
     note("info", "chat: cancel sent");
   } catch (e) {
-    note("error", `chat cancel: ${errText(e)}`);
+    note("error", `chat: cancel failed — ${errText(e)}`);
   }
   // The backend still emits a `turn_end` (stopReason `cancelled`), which clears
   // `turnActive` and flushes the queue — no local state change needed here.
@@ -589,11 +589,11 @@ async function scale(
   repaintRoster(); // disable the button immediately
   try {
     await source.scaleDeployment(name, size, namespace, activeCluster);
-    note("info", `${action === "start" ? "started" : "stopped"} ${namespace}/${name}`);
+    note("info", `roster: ${action === "start" ? "started" : "stopped"} ${namespace}/${name}`);
     // tick() observes the new desiredCount and prunes the guard when it flips.
     await tick();
   } catch (e) {
-    note("error", `${action} ${namespace}/${name}: ${errText(e)}`);
+    note("error", `roster: ${action} ${namespace}/${name} failed — ${errText(e)}`);
     clearPending(key);
     repaintRoster();
   }
@@ -643,7 +643,7 @@ async function startCore(): Promise<void> {
   try {
     await invoke("start_core");
   } catch (e) {
-    note("error", `start_core: ${errText(e)}`);
+    note("error", `core: start failed — ${errText(e)}`);
   }
 }
 
@@ -678,15 +678,15 @@ function setupUpdater(): void {
         pending = info;
         btn.textContent = `Update to v${info.version} ↻`;
         btn.classList.add("has-update");
-        note("info", `New version v${info.version} available (current v${info.current}) — click to install and restart`);
+        note("info", `update: v${info.version} available (current v${info.current}) — click to install`);
       } else {
-        note("info", "Already up to date");
+        note("info", "update: already up to date");
         btn.textContent = "Up to date";
         window.setTimeout(reset, 4000);
       }
     } catch (e) {
       reset();
-      note("error", `Update check failed: ${errText(e)}`);
+      note("error", `update: check failed — ${errText(e)}`);
     } finally {
       btn.disabled = false;
     }
@@ -701,7 +701,7 @@ function setupUpdater(): void {
     } catch (e) {
       btn.disabled = false;
       btn.textContent = pending ? `Update to v${pending.version} ↻` : "Check for updates";
-      note("error", `Update install failed: ${errText(e)}`);
+      note("error", `update: install failed — ${errText(e)}`);
     }
   }
 
@@ -747,7 +747,7 @@ async function boot(): Promise<void> {
   renderChat();
   updateChatControls();
   if (clusterLabel) clusterLabel.textContent = activeCluster;
-  note("info", `polling cluster "${activeCluster}" every ${POLL_MS / 1000}s`);
+  note("info", `app: polling cluster "${activeCluster}" every ${POLL_MS / 1000}s`);
   setupUpdater();
   await startCore();
   void refreshConfig();

crates/acp-tunnel/src/lib.rs

@@ -267,6 +267,29 @@ impl Session {
         ))
     }
 
+    /// Build a liveness-probe request (client heartbeat). The method is
+    /// deliberately one the gateway does **not** implement, so its read loop
+    /// answers it **itself** with an immediate `-32601` — the probe never reaches
+    /// the agent runtime, costs no model turn, and is answered even while a
+    /// `session/prompt` is in flight (the gateway spawns prompts, so its read loop
+    /// keeps servicing frames). A response of any kind proves the socket + gateway
+    /// are alive; silence across a probe interval means the socket is half-open.
+    /// Returns `(id, frame)`; the transport treats *any* inbound frame as liveness,
+    /// so it need not correlate this id.
+    pub fn heartbeat(&mut self) -> (u64, Value) {
+        let id = self.alloc_id();
+        (id, request(id, "studio/ping", json!({})))
+    }
+
+    /// Forget the current session id (keeping the declared servers and phase), so
+    /// the next handshake opens a **fresh** session via [`Session::open_session`]
+    /// instead of [`Session::resume`]. Used when the gateway rejects a
+    /// `session/resume` because it has already reaped the session past its grace
+    /// window — the transport then falls back to `session/new`.
+    pub fn forget_session(&mut self) {
+        self.session_id = None;
+    }
+
     /// Replace the declaration set — used on reconnect to swap in fresh per-
     /// connection server ids before a [`Session::resume`], and to reset the phase
     /// to [`Phase::New`] so a full `initialize` runs on the new socket.
@@ -407,10 +430,7 @@ mod tests {
 
     #[test]
     fn bearer_subprotocol_pairs_the_token_with_acp_v1() {
-        assert_eq!(
-            bearer_subprotocol("tok123"),
-            "openab.bearer.tok123, acp.v1"
-        );
+        assert_eq!(bearer_subprotocol("tok123"), "openab.bearer.tok123, acp.v1");
     }
 
     #[test]
@@ -559,7 +579,12 @@ mod tests {
             }
         });
         match parse_inbound(&frame) {
-            Inbound::Message { id, connection_id, method, params } => {
+            Inbound::Message {
+                id,
+                connection_id,
+                method,
+                params,
+            } => {
                 assert_eq!(id, Some(json!(12))); // request → reply owed
                 assert_eq!(connection_id, "conn-1");
                 assert_eq!(method, "tools/call");
@@ -606,15 +631,21 @@ mod tests {
         match parse_inbound(&disc) {
             Inbound::Disconnect { id, connection_id } => {
                 assert_eq!(connection_id, "conn-1");
-                assert_eq!(disconnect_reply(id), json!({ "jsonrpc": "2.0", "id": 3, "result": {} }));
+                assert_eq!(
+                    disconnect_reply(id),
+                    json!({ "jsonrpc": "2.0", "id": 3, "result": {} })
+                );
             }
             other => panic!("expected Disconnect, got {other:?}"),
         }
         // cancel is a notification keyed by the OUTER frame id of the abandoned request
-        let cancel = json!({ "jsonrpc": "2.0", "method": "mcp/cancel", "params": { "requestId": 42 } });
+        let cancel =
+            json!({ "jsonrpc": "2.0", "method": "mcp/cancel", "params": { "requestId": 42 } });
         assert_eq!(
             parse_inbound(&cancel),
-            Inbound::Cancel { request_id: json!(42) }
+            Inbound::Cancel {
+                request_id: json!(42)
+            }
         );
     }
 
@@ -639,15 +670,21 @@ mod tests {
     #[test]
     fn prompt_needs_a_session_then_builds_a_text_turn() {
         let mut fresh = Session::new(vec![oab_server("c")]);
-        assert!(fresh.prompt("hi").is_none(), "no prompt before a session exists");
+        assert!(
+            fresh.prompt("hi").is_none(),
+            "no prompt before a session exists"
+        );
 
         let mut s = active_session();
         let (id, frame) = s.prompt("hello there").expect("session active");
         assert_eq!(frame["jsonrpc"], "2.0");
         assert_eq!(frame["id"], id);
         assert_eq!(frame["method"], "session/prompt");
         assert_eq!(frame["params"]["sessionId"], "sess-1");
-        assert_eq!(frame["params"]["prompt"][0], json!({ "type": "text", "text": "hello there" }));
+        assert_eq!(
+            frame["params"]["prompt"][0],
+            json!({ "type": "text", "text": "hello there" })
+        );
     }
 
     #[test]
@@ -660,6 +697,28 @@ mod tests {
         assert_eq!(f["params"]["sessionId"], "sess-1");
     }
 
+    #[test]
+    fn heartbeat_is_a_request_with_an_unimplemented_method() {
+        let mut s = active_session();
+        let (id, frame) = s.heartbeat();
+        assert_eq!(frame["jsonrpc"], "2.0");
+        assert_eq!(frame["id"], id);
+        // an unimplemented method → the gateway answers -32601 itself, never the agent
+        assert_eq!(frame["method"], "studio/ping");
+        // it carries an id (a request), so a response is owed and proves liveness
+        assert!(frame.get("id").is_some());
+    }
+
+    #[test]
+    fn forget_session_forces_a_fresh_open_not_a_resume() {
+        let mut s = active_session();
+        assert!(s.session_id().is_some());
+        s.forget_session();
+        assert!(s.session_id().is_none());
+        // with no session id, resume is unavailable — the transport must open a new one
+        assert!(s.resume("/w").is_none());
+    }
+
     #[test]
     fn agent_message_chunk_is_classified_others_stay_other() {
         let chunk = json!({
@@ -671,7 +730,9 @@ mod tests {
         });
         assert_eq!(
             parse_inbound(&chunk),
-            Inbound::AgentChunk { text: "the ocean is vast".into() }
+            Inbound::AgentChunk {
+                text: "the ocean is vast".into()
+            }
         );
         // thought / tool_call kinds openab does not emit yet → Other
         let thought = json!({ "method": "session/update", "params": { "update": {

src-tauri/src/lib.rs

@@ -26,7 +26,7 @@ async fn start_core(app: tauri::AppHandle, core: tauri::State<'_, Core>) -> Resu
     if guard.is_some() {
         return Ok(());
     }
-    let _ = app.emit("app-log", json!({ "level": "info", "msg": "OAB Studio starting…" }));
+    // (the `core: spawning…` line from McpClient::spawn covers the "starting" beat)
     match McpClient::spawn(&app, &default_cluster()).await {
         Ok(client) => {
             *guard = Some(client);
@@ -35,7 +35,7 @@ async fn start_core(app: tauri::AppHandle, core: tauri::State<'_, Core>) -> Resu
         Err(e) => {
             let _ = app.emit(
                 "app-log",
-                json!({ "level": "error", "msg": format!("failed to start core: {e}") }),
+                json!({ "level": "error", "msg": format!("core: failed to start — {e}") }),
             );
             Err(e)
         }
@@ -324,15 +324,15 @@ async fn install_update(app: tauri::AppHandle) -> Result<(), String> {
     };
     let _ = app.emit(
         "app-log",
-        json!({ "level": "info", "msg": format!("downloading update v{}…", update.version) }),
+        json!({ "level": "info", "msg": format!("update: downloading v{}…", update.version) }),
     );
     update
         .download_and_install(|_chunk, _total| {}, || {})
         .await
         .map_err(|e| e.to_string())?;
     let _ = app.emit(
         "app-log",
-        json!({ "level": "info", "msg": "update installed — restarting…" }),
+        json!({ "level": "info", "msg": "update: installed — restarting…" }),
     );
     app.restart();
 }