diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+Params.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+Params.swift
index d93ebe80d054..95722c90f748 100644
--- a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+Params.swift
+++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+Params.swift
@@ -70,6 +70,7 @@ nonisolated extension AcpmuxPaneMethods {
         // grants nothing (the same probes acpmux runs in the background), so no gesture rule.
         "_acpmux/models": (["refresh"], []),
         "_acpmux/status": ([], []),
+        "_acpmux/peers": ([], []),
         "_acpmux/draft_get": (["sessionId"], []),
         "_acpmux/draft_set": (["sessionId", "text"], []),
         "_acpmux/bookmark_get": (["sessionId"], []),
diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+ReplyShapes.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+ReplyShapes.swift
index fd8861944d44..717208311d04 100644
--- a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+ReplyShapes.swift
+++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods+ReplyShapes.swift
@@ -27,6 +27,10 @@ nonisolated extension AcpmuxPaneMethods {
     /// tell a chat on a machine that is not connected (with when it was last seen) from a deleted
     /// one, and from one whose machine is still connecting (chatUnavailable.ts).
     public static let replyShapes: [String: ReplyShape] = [
+        // The local/app connection's peer state snapshot; never transport URLs or credentials.
+        "_acpmux/peers": .object(["peers": .list(.object([
+            "name": .string, "state": .string, "lastSeen": .number, "error": .string,
+        ]))]),
         "_acpmux/status": .object([
             "peers": .list(.object([
                 "name": .string, "transport": .scheme(from: "url"), "connected": .bool,
diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods.swift
index b34493f4e818..a8808a3c947f 100644
--- a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods.swift
+++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxPaneMethods.swift
@@ -46,7 +46,7 @@ nonisolated enum AcpmuxPaneMethods {
         // Take Over a View Only chat (direct.ts takeOver): the chat runs the conversation from then on.
         "_acpmux/watch_take_over",
         // Read-only, and its reply is filtered to ``replyShapes`` (ad349).
-        "_acpmux/status", "_acpmux/draft_get",
+        "_acpmux/status", "_acpmux/peers", "_acpmux/draft_get",
         // Session-scoped draft writes are kept separate from the filtered replies above.
         "_acpmux/draft_set",
         // The thread minimap's prompt bookmarks of a chat in the pane's scope (direct.ts
diff --git a/Packages/macOS/CmuxNext/Tests/CmuxNextAgentPaneTests/AcpmuxPanePeerSnapshotTests.swift b/Packages/macOS/CmuxNext/Tests/CmuxNextAgentPaneTests/AcpmuxPanePeerSnapshotTests.swift
new file mode 100644
index 000000000000..0fa23846a4cd
--- /dev/null
+++ b/Packages/macOS/CmuxNext/Tests/CmuxNextAgentPaneTests/AcpmuxPanePeerSnapshotTests.swift
@@ -0,0 +1,23 @@
+import Foundation
+import Testing
+@testable import CmuxNextAgentPane
+
+@Suite struct AcpmuxPanePeerSnapshotTests {
+    @Test func peerSnapshotAcceptsNoParameters() {
+        #expect(AcpmuxPaneMethods.requests.contains("_acpmux/peers"))
+        #expect(!AcpmuxPaneMethods.breaksParamsRule(["method": "_acpmux/peers", "params": [:]], modeFields: nil))
+        for params in [["command": "pwd"], ["peer": "other"], ["url": "ssh://private"]] {
+            #expect(AcpmuxPaneMethods.breaksParamsRule(["method": "_acpmux/peers", "params": params], modeFields: nil))
+        }
+    }
+
+    @Test func peerSnapshotKeepsOnlyDisplayFields() throws {
+        let shape = try #require(AcpmuxPaneMethods.replyShapes["_acpmux/peers"])
+        let value: [String: Any] = ["peers": [["name": "studio-mac", "state": "offline", "lastSeen": 5,
+                                              "error": "removed", "url": "ssh://private", "token": "secret"]]]
+        let filtered = try #require(AcpmuxPaneMethods.filtered(value, to: shape) as? [String: Any])
+        let peer = try #require((filtered["peers"] as? [[String: Any]])?.first)
+        #expect(Set(peer.keys) == ["name", "state", "lastSeen", "error"])
+        #expect(peer["error"] as? String == "removed")
+    }
+}
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/policy.json b/cmux-tui/crates/cmux-agent-pane-policy/policy.json
index 97612d9e7faf..a9cc54bef3d5 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/policy.json
+++ b/cmux-tui/crates/cmux-agent-pane-policy/policy.json
@@ -7,7 +7,7 @@
     "_acpmux/watch", "_acpmux/sessions", "_acpmux/events", "_acpmux/attachment", "_acpmux/attach", "_acpmux/detach", "_acpmux/view", "_acpmux/typing", "_acpmux/warm",
     "_acpmux/peer_snapshot", "_acpmux/peer_reconnect", "_acpmux/retry",
     "_acpmux/kill", "_acpmux/prewarm", "_acpmux/harnesses", "_acpmux/models", "_acpmux/permission_respond",
-    "_acpmux/status", "_acpmux/draft_get", "_acpmux/draft_set", "_acpmux/bookmark_get", "_acpmux/bookmark_set", "_acpmux/btw", "_acpmux/queue_remove", "_acpmux/queue_edit",
+    "_acpmux/status", "_acpmux/peers", "_acpmux/draft_get", "_acpmux/draft_set", "_acpmux/bookmark_get", "_acpmux/bookmark_set", "_acpmux/btw", "_acpmux/queue_remove", "_acpmux/queue_edit",
     "_acpmux/handoff_prepare", "_acpmux/handoff_get", "_acpmux/handoff_draft", "_acpmux/handoff_start",
     "_acpmux/handoff_discard",
     "_acpmux/permission_groups", "_acpmux/permission_group_respond", "_acpmux/permission_chat_revoke",
@@ -78,6 +78,7 @@
     "_acpmux/harnesses": {"params": ["cwd"], "acpmux": []},
     "_acpmux/models": {"params": ["refresh"], "acpmux": []},
     "_acpmux/status": {"params": [], "acpmux": []},
+    "_acpmux/peers": {"params": [], "acpmux": []},
     "_acpmux/draft_get": {"params": ["sessionId"], "acpmux": []},
     "_acpmux/draft_set": {"params": ["sessionId", "text"], "acpmux": []},
     "_acpmux/bookmark_get": {"params": ["sessionId"], "acpmux": []},
@@ -126,7 +127,8 @@
   },
   "reply_shapes": {
     "_acpmux/status": {"object": {"peers": {"list": {"object": {"name": "string"}}}}},
-    "_acpmux/peer_reconnect": {"object": {"peers": {"list": {"object": {"name": "string", "state": "string", "error": "string"}}}}}
+    "_acpmux/peer_reconnect": {"object": {"peers": {"list": {"object": {"name": "string", "state": "string", "error": "string"}}}}},
+    "_acpmux/peers": {"object": {"peers": {"list": {"object": {"name": "string", "state": "string", "lastSeen": "number", "error": "string"}}}}}
   },
   "history_replies": ["_acpmux/attach", "_acpmux/events"],
   "path_keys": ["cwd", "path", "additionalDirectories", "directory", "directories", "workingDirectory",
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/src/data.rs b/cmux-tui/crates/cmux-agent-pane-policy/src/data.rs
index 4d97cdd8d503..0c8a74e2aa32 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/src/data.rs
+++ b/cmux-tui/crates/cmux-agent-pane-policy/src/data.rs
@@ -53,6 +53,8 @@ pub struct QuestionAnswers {
 #[derive(Clone, Debug, PartialEq)]
 pub enum ReplyShape {
     String,
+    /// A finite number, never a boolean (Swift's `.number`).
+    Number,
     Object(BTreeMap<String, ReplyShape>),
     List(Box<ReplyShape>),
 }
@@ -62,6 +64,9 @@ impl ReplyShape {
         if v.as_str() == Some("string") {
             return Some(ReplyShape::String);
         }
+        if v.as_str() == Some("number") {
+            return Some(ReplyShape::Number);
+        }
         let o = v.as_object()?;
         if let Some(fields) = o.get("object").and_then(Value::as_object) {
             let fields = fields
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/src/reply.rs b/cmux-tui/crates/cmux-agent-pane-policy/src/reply.rs
index 7a7cce68c89e..b77550f5147e 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/src/reply.rs
+++ b/cmux-tui/crates/cmux-agent-pane-policy/src/reply.rs
@@ -10,6 +10,8 @@ use serde_json::{Map, Value, json};
 pub fn filtered(value: &Value, shape: &ReplyShape) -> Option<Value> {
     match shape {
         ReplyShape::String => value.as_str().map(|s| Value::String(s.to_owned())),
+        // serde_json keeps no NaN or infinity, and a boolean is not a number.
+        ReplyShape::Number => value.is_number().then(|| value.clone()),
         ReplyShape::List(item) => value
             .as_array()
             .map(|a| Value::Array(a.iter().filter_map(|v| filtered(v, item)).collect())),
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/frames.json b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/frames.json
index 475838e5dcf1..4ce37405ca87 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/frames.json
+++ b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/frames.json
@@ -446,6 +446,15 @@
    "send": "unchanged"
   }
  },
+ {
+  "name": "request _acpmux/peers passes",
+  "text": "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"_acpmux/peers\",\"params\":{}}",
+  "first": false,
+  "token": "abc",
+  "expect": {
+   "send": "unchanged"
+  }
+ },
  {
   "name": "request _acpmux/warm passes",
   "text": "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"_acpmux/warm\",\"params\":{}}",
@@ -643,24 +652,24 @@
   }
  },
  {
-  "name": "escaped '_acpmux/peers' refused",
-  "text": "{\"jsonrpc\":\"2.0\",\"id\":505,\"method\":\"\\u005f\\u0061\\u0063\\u0070\\u006d\\u0075\\u0078\\u002f\\u0070\\u0065\\u0065\\u0072\\u0073\",\"params\":{}}",
+  "name": "escaped '_acpmux/peer_remove' refused",
+  "text": "{\"jsonrpc\":\"2.0\",\"id\":505,\"method\":\"\\u005f\\u0061\\u0063\\u0070\\u006d\\u0075\\u0078\\u002f\\u0070\\u0065\\u0065\\u0072\\u005f\\u0072\\u0065\\u006d\\u006f\\u0076\\u0065\",\"params\":{}}",
   "first": false,
   "token": null,
   "expect": {
    "refuse": "transport.method_refused",
-   "method": "_acpmux/peers",
+   "method": "_acpmux/peer_remove",
    "id": "505"
   }
  },
  {
-  "name": "notification '_acpmux/peers' refused",
-  "text": "{\"jsonrpc\":\"2.0\",\"method\":\"_acpmux/peers\",\"params\":{}}",
+  "name": "notification '_acpmux/peer_remove' refused",
+  "text": "{\"jsonrpc\":\"2.0\",\"method\":\"_acpmux/peer_remove\",\"params\":{}}",
   "first": false,
   "token": null,
   "expect": {
    "refuse": "transport.method_refused",
-   "method": "_acpmux/peers",
+   "method": "_acpmux/peer_remove",
    "id": null
   }
  },
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/params.json b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/params.json
index 12a48e84a3b8..49f73e43ee03 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/params.json
+++ b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/params.json
@@ -1980,5 +1980,29 @@
   },
   "mode_fields": null,
   "breaks": true
+ },
+ {
+  "name": "_acpmux/peers every known param",
+  "frame": {
+   "jsonrpc": "2.0",
+   "id": 1,
+   "method": "_acpmux/peers",
+   "params": {}
+  },
+  "mode_fields": null,
+  "breaks": false
+ },
+ {
+  "name": "_acpmux/peers refuses any param",
+  "frame": {
+   "jsonrpc": "2.0",
+   "id": 1,
+   "method": "_acpmux/peers",
+   "params": {
+    "peer": "other"
+   }
+  },
+  "mode_fields": null,
+  "breaks": true
  }
 ]
diff --git a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/replies.json b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/replies.json
index 46f32f2647f6..cce9b2615f32 100644
--- a/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/replies.json
+++ b/cmux-tui/crates/cmux-agent-pane-policy/tests/cases/replies.json
@@ -160,5 +160,52 @@
     ]
    }
   }
+ },
+ {
+  "name": "peer snapshot keeps each peer's name, state, lastSeen and error, never its endpoint or a boolean lastSeen",
+  "method": "_acpmux/peers",
+  "page_id": "9",
+  "reply": {
+   "jsonrpc": "2.0",
+   "id": 42,
+   "result": {
+    "peers": [
+     {
+      "name": "studio",
+      "url": "ssh://studio",
+      "token": "s",
+      "state": "restarting",
+      "lastSeen": 1793357781599,
+      "error": "daemon restarting",
+      "attempts": 2
+     },
+     {
+      "name": "box",
+      "state": "connected",
+      "lastSeen": true,
+      "error": null
+     }
+    ],
+    "token": "s"
+   }
+  },
+  "expected": {
+   "jsonrpc": "2.0",
+   "id": 9,
+   "result": {
+    "peers": [
+     {
+      "name": "studio",
+      "state": "restarting",
+      "lastSeen": 1793357781599,
+      "error": "daemon restarting"
+     },
+     {
+      "name": "box",
+      "state": "connected"
+     }
+    ]
+   }
+  }
  }
 ]
diff --git a/webviews/src/agent-session/acpmux/chatUnavailable.ts b/webviews/src/agent-session/acpmux/chatUnavailable.ts
index bb1ef8329fb7..067bdd8205c8 100644
--- a/webviews/src/agent-session/acpmux/chatUnavailable.ts
+++ b/webviews/src/agent-session/acpmux/chatUnavailable.ts
@@ -15,7 +15,7 @@
 /// One peer as `_acpmux/status` last said.
 export type PeerStatus = {
   connected?: boolean;
-  /// acpmux's link state: `connecting`, `connected`, `degraded`, `reconnecting` or `offline`.
+  /// acpmux's link state: `connecting`, `connected`, `degraded`, `reconnecting`, `restarting` or `offline`.
   state?: string;
   /// When the peer last answered, in milliseconds since the epoch.
   lastSeen?: number;
@@ -66,10 +66,11 @@ function settled(status: PeerStatus): boolean {
 }
 
 /// A peer still on its way up: acpmux's first attempt (`connecting`), a socket whose handshake has
-/// not finished (`degraded`), or a saved peer a restarted daemon has not tried yet.
+/// not finished (`degraded`), a peer whose daemon announced a restart (`restarting`), or a saved
+/// peer a restarted daemon has not tried yet.
 function stillConnecting(status: PeerStatus): boolean {
   if (settled(status)) return false;
-  if (status.state === "connecting" || status.state === "degraded") return true;
+  if (status.state === "connecting" || status.state === "degraded" || status.state === "restarting") return true;
   return status.state === "reconnecting" && (status.attempts ?? 0) === 0;
 }
 
diff --git a/webviews/src/agent-session/acpmux/direct.ts b/webviews/src/agent-session/acpmux/direct.ts
index 9bd0c7fef326..88dbb281a1de 100644
--- a/webviews/src/agent-session/acpmux/direct.ts
+++ b/webviews/src/agent-session/acpmux/direct.ts
@@ -39,6 +39,7 @@ import { carryInline, inlinePrompt, recordedPromptParts, restoreInline, stripFil
 import { liftInlineImages } from "./inlineImages";
 import { primeSentImages } from "./conversation/imageEntries";
 import { resentAttachments, shownPrompt } from "./pastedText";
+import { PeerStateEvents } from "./peerStateEvents";
 import { peerConnectedFromStatus, peerKindsFromStatus, peerStatesFromStatus } from "./environments";
 import { peerResourcesFromStatus } from "./machineResources";
 import { type ChatUnavailable, chatUnavailable, type PeerStatus, peerStatusFromStatus } from "./chatUnavailable";
@@ -1544,6 +1545,7 @@ export class AcpmuxDirectClient {
   /// acpmux lists `acp.session.fork` among the operations it serves.
   private canFork = false;
   private handoffSupported = false;
+  private readonly peerEvents = new PeerStateEvents();
   /// The `_acpmux/*` methods acpmux's `initialize` lists (`_meta.acpmux.extensions`).
   private extensions: readonly string[] = [];
   private features = new Set<string>();
@@ -1841,6 +1843,8 @@ export class AcpmuxDirectClient {
     if (this.opening || this.closed) return;
     this.features.clear();
     this.opening = true;
+    // A new socket must advertise its own feature and proven origin before its notes are accepted.
+    this.peerEvents.configure(undefined);
     // A reconnect owns the attach of the copy on screen from its start: a row that says its machine
     // is back meanwhile waits for it (`copyBack`), never a second attach over the same copy.
     if (this.peerCopy && this.peerCopy.sessionId === this.selectedSessionId) this.peerCopy.attaching = true;
@@ -1907,7 +1911,10 @@ export class AcpmuxDirectClient {
           onLost();
         } else this.scheduleReconnect();
       };
-      socket.onmessage = (message) => acpmuxPerf.timed("client.receive", () => this.receive(String(message.data)));
+      // A socket this client replaced still answers what was asked on it (an export that watches for
+      // the swap rejects on that answer); only its peer notes, about a connection that is gone, drop.
+      socket.onmessage = (message) =>
+        acpmuxPerf.timed("client.receive", () => this.receive(String(message.data), this.socket !== socket));
     }).catch((error: unknown) => {
       this.releaseCopyAttach(mine);
       throw error;
@@ -1935,6 +1942,7 @@ export class AcpmuxDirectClient {
       const origin = initialized?._meta?.acpmux?.origin;
       this.origin = origin === "local" || origin === "remote" || origin === "peer" ? origin : "unknown";
       this.handoffSupported = supportsHandoff(initialized);
+      this.peerEvents.configure(initialized);
       const groupedPermissionsSupported = supportsPermissionGroups(initialized);
       if (!groupedPermissionsSupported) this.groupedPermissions.clear();
       this.permissions.configure(groupedPermissionsSupported);
@@ -1943,13 +1951,17 @@ export class AcpmuxDirectClient {
       // acpmux answers a connection's requests side by side (server/mod.rs spawns each), so the
       // frame waits on one round trip rather than one per read (Leo, 2026-10-29: "as in i want
       // ours to load fast too :(").
+      const peersSince = this.peerEvents.statusStarted();
       const peersRead = this.request("_acpmux/status", {}).catch(() => undefined);
       const listRead = this.sessionPages.first();
       // Awaited below, after the machines: a list that fails meanwhile is not left unobserved.
       listRead.catch(() => undefined);
       early = this.askEarly();
-      this.readPeers(await peersRead);
+      this.readPeers(await peersRead, peersSince);
       const first = await listRead;
+      // After the list's `_acpmux/watch` answered, so no peer note falls between the read and the
+      // watch; not awaited, so the first frame keeps its one round trip (#27755).
+      if (this.peerEvents.supported) void this.refreshPeerStatesAndEmit();
       let laterPages = first?.next ? first : undefined;
       const linkedLater =
         first !== undefined &&
@@ -2584,7 +2596,7 @@ export class AcpmuxDirectClient {
     }, delay);
   }
 
-  private receive(raw: string): void {
+  private receive(raw: string, stale = false): void {
     this.wire.received(raw);
     // A frame that is not a JSON object is dropped like one that is not JSON (`null` would throw at `in`).
     const message = parseJsonObjectFrame(raw) as Reply | Notification | null;
@@ -2618,6 +2630,13 @@ export class AcpmuxDirectClient {
       return;
     }
     const notification = message as Notification;
+    if (notification.method === "_acpmux/peer_state_changed") {
+      if (!stale && !this.closed && this.peerEvents.changed(notification.params)) {
+        this.applyPeerStates();
+        if (!this.snapshotHold) this.emit();
+      }
+      return;
+    }
     // Who else is in the chat, ephemeral (multiplayer/presenceClient.ts): never a record.
     if (notification.method === "_acpmux/presence") {
       this.presence.notification(notification.params);
@@ -2687,6 +2706,9 @@ export class AcpmuxDirectClient {
   /// agent-gui daemons send {sessionIds, watch, dropped}; older ones send only
   /// {dropped}, so a notice without sessionIds resyncs the selected session.
   private resyncAfterLag(params: any): void {
+    // Peer state notices may be among the dropped ones: read the snapshot again, so a machine that
+    // went down while this client lagged never shows as connected (no session needs to be open).
+    if (this.peerEvents.supported) void this.refreshPeerStatesAndEmit();
     // Needs you's asks: notices of new or answered requests may be lost, so every chat with open
     // requests is read again (after the session list, which drops the chats acpmux no longer lists).
     if (params?.watch === true)
@@ -7617,16 +7639,62 @@ export class AcpmuxDirectClient {
     ];
   }
   /// Reads acpmux's peers again (`_acpmux/status`): the environment picker asks as it opens, so a
-  /// machine that connected or dropped since shows as it is now. acpmux sends no peer events.
+  /// machine that connected or dropped since shows as it is now. Local/app peer events keep the
+  /// closed control current when the daemon advertises them; older daemons keep their last snapshot.
   async refreshPeers(): Promise<void> {
+    const since = this.peerEvents.statusStarted();
     const status = await this.request("_acpmux/status", {}).catch(() => undefined);
     if (status === undefined) return;
-    this.readPeers(status);
+    this.readPeers(status, since);
     // A held snapshot (a chat being switched or reattached) draws the peers when it is released;
     // emitting now would record "connected" on the hold before setup says so.
     if (!this.snapshotHold) this.emit();
   }
-  private readPeers(status: any): void {
+  private async refreshPeerStates(): Promise<boolean> {
+    const read = this.peerEvents.readStarted();
+    if (!this.peerEvents.snapshot(await this.request("_acpmux/peers", {}), read)) return false;
+    this.applyPeerStates();
+    return true;
+  }
+  /// Reads the peer snapshot off the caller's path and draws it when it lands; a held snapshot (or a
+  /// connection still opening) draws it with its own emit.
+  private async refreshPeerStatesAndEmit(): Promise<void> {
+    const applied = await this.refreshPeerStates().catch(() => false);
+    if (applied && !this.closed && !this.opening && !this.snapshotHold) this.emit();
+  }
+  private applyPeerStates(): void {
+    const peers = [...this.peerEvents.entries()];
+    // A peer the status read lists but an authoritative snapshot omits (removed since, or a row it
+    // could not read) is not connected: with no entry at all the picker would read it as ready.
+    const unlisted = this.peerEvents.authoritative
+      ? Object.fromEntries(this.peers.map((peer) => [peer, false] as const))
+      : this.peerConnected;
+    this.peerStates = {
+      ...(this.peerEvents.authoritative ? {} : this.peerStates),
+      ...Object.fromEntries(peers.map(([peer, link]) => [peer, link.state])),
+    };
+    this.peerConnected = {
+      ...unlisted,
+      ...Object.fromEntries(peers.map(([peer, link]) => [peer, link.connected])),
+    };
+    this.peerStatus = {
+      ...(this.peerEvents.authoritative
+        ? Object.fromEntries(
+            this.peers.map((peer) => {
+              // Keeps what the status read last said (`lastSeen`, `attempts`) for the offline banner.
+              const { state: _state, ...seen } = this.peerStatus[peer] ?? {};
+              return [peer, { ...seen, connected: false }];
+            }),
+          )
+        : this.peerStatus),
+      ...Object.fromEntries(peers.map(([peer, link]) => [peer, { ...this.peerStatus[peer], ...link }])),
+    };
+  }
+  /// `since`: the status read's tick (`PeerStateEvents.statusStarted`); omitted for a reply that is
+  /// not a fresh read of every peer's state, which then never replaces an event entry.
+  private readPeers(status: any, since?: number): void {
+    // Asked before a status read that already landed: it would put back what that one replaced.
+    if (since !== undefined && !this.peerEvents.statusRead(since)) return;
     this.peers = Array.isArray(status?.peers)
       ? status.peers
           .map((peer: any) => (typeof peer?.name === "string" ? peer.name : undefined))
@@ -7637,6 +7705,7 @@ export class AcpmuxDirectClient {
     this.peerStates = peerStatesFromStatus(status);
     this.peerResources = peerResourcesFromStatus(status);
     this.peerStatus = peerStatusFromStatus(status);
+    if (this.peerEvents.observed) this.applyPeerStates();
   }
   /// Probes every agent's live model list again (`_acpmux/models {refresh: true}`): acpmux forgets
   /// the lists it has and asks each harness before it answers (the picker's Refresh models).
diff --git a/webviews/src/agent-session/acpmux/peerStateEvents.test.ts b/webviews/src/agent-session/acpmux/peerStateEvents.test.ts
new file mode 100644
index 000000000000..30d1799267d1
--- /dev/null
+++ b/webviews/src/agent-session/acpmux/peerStateEvents.test.ts
@@ -0,0 +1,386 @@
+import { afterAll, beforeAll, expect, spyOn, test } from "bun:test";
+import { JSDOM } from "jsdom";
+import type { AcpmuxSnapshot } from "./model";
+
+const dom = new JSDOM("<!doctype html>", { url: "http://localhost/" });
+const scope = globalThis as Record<string, unknown>;
+const saved = new Map(["window", "document"].map((key) => [key, Object.getOwnPropertyDescriptor(scope, key)]));
+beforeAll(() => Object.assign(scope, { window: dom.window, document: dom.window.document }));
+afterAll(() => {
+  for (const [key, descriptor] of saved) {
+    if (descriptor) Object.defineProperty(scope, key, descriptor);
+    else delete scope[key];
+  }
+  dom.window.close();
+});
+
+type Peer = { name: string; state: string; lastSeen?: number | null; error?: string | null };
+async function fixture(origin: string | undefined, supported = true, holdPeers = false) {
+  const { AcpmuxDirectClient } = await import("./direct");
+  const { MockAcpmuxSocket, mockHost } = await import("./mock");
+  const requests: string[] = [];
+  const snapshots: AcpmuxSnapshot[] = [];
+  let peers: Peer[] = [{ name: "studio-mac", state: "connecting", lastSeen: null, error: null }];
+  let status: Record<string, unknown>[] = [
+    { name: "studio-mac", transport: "ssh", connected: true, state: "connected", lastSeen: 99 },
+  ];
+  // A held read answers only once released, with what the daemon says then (`hold`, `release`).
+  const gates: Record<string, { open: Promise<void>; release: () => void } | undefined> = {};
+  const hold = (method: string) => {
+    let release = () => {};
+    const open = new Promise<void>((resolve) => (release = resolve));
+    gates[method] = { open, release };
+    return release;
+  };
+  const release = (method: string) => {
+    gates[method]?.release();
+    gates[method] = undefined;
+  };
+  /// Later reads of `method` answer at once; the ones already held stay held.
+  const free = (method: string) => {
+    gates[method] = undefined;
+  };
+  if (holdPeers) hold("_acpmux/peers");
+  let socket: InstanceType<typeof MockAcpmuxSocket>;
+  let onSnapshot: ((snapshot: AcpmuxSnapshot) => void) | undefined;
+  const openSocket = () => {
+    socket = new MockAcpmuxSocket(() => Promise.resolve());
+    const daemon = socket as unknown as { answer(method: string, params: Record<string, unknown>): Promise<unknown> };
+    const answer = daemon.answer.bind(daemon);
+    daemon.answer = async (method, params) => {
+      requests.push(method);
+      if (method === "initialize")
+        return { protocolVersion: 1, _meta: { acpmux: { origin, features: supported ? ["peerStateEvents"] : [] } } };
+      if (method === "_acpmux/status") {
+        // The daemon answers with what it knew when the read reached it.
+        const said = status;
+        await gates[method]?.open;
+        return { peers: said };
+      }
+      if (method === "_acpmux/peers") {
+        await gates[method]?.open;
+        return { peers };
+      }
+      return answer(method, params);
+    };
+    return socket as unknown as WebSocket;
+  };
+  const client = await AcpmuxDirectClient.connect(
+    mockHost,
+    (snapshot) => {
+      snapshots.push(snapshot);
+      onSnapshot?.(snapshot);
+    },
+    undefined,
+    openSocket,
+  );
+  const notify = (value: Record<string, unknown>) =>
+    socket.onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/peer_state_changed", params: value }),
+    });
+  const f = {
+    client,
+    socket: () => socket,
+    waitForState: (state: string | undefined) =>
+      new Promise<void>((resolve) => {
+        if (snapshots.length > 0 && snapshots.at(-1)!.peerStates?.["studio-mac"] === state) return resolve();
+        onSnapshot = (snapshot) => {
+          if (snapshot.peerStates?.["studio-mac"] === state) {
+            onSnapshot = undefined;
+            resolve();
+          }
+        };
+      }),
+    requests,
+    snapshots,
+    notify,
+    latest: () => snapshots.at(-1)!,
+    peers: (value: Peer[]) => {
+      peers = value;
+    },
+    status: (value: Record<string, unknown>[]) => {
+      status = value;
+    },
+    releasePeers: () => release("_acpmux/peers"),
+    hold,
+    release,
+    free,
+    peerConnected: () => (client as unknown as { peerConnected: Record<string, boolean> }).peerConnected,
+    peerStatus: () => (client as unknown as { peerStatus: Record<string, Record<string, unknown>> }).peerStatus,
+    /// What the client holds now, drawn or not.
+    peerStates: () => (client as unknown as { peerStates: Record<string, string> }).peerStates,
+    /// The client's own `_acpmux/peers` read, awaited (what a lag notice or a reconnect starts).
+    readSnapshot: () => (client as unknown as { refreshPeerStates(): Promise<boolean> }).refreshPeerStates(),
+  };
+  // The initial snapshot lands after the first frame: settle it, unless the test holds it.
+  if (origin === "local" && supported && !holdPeers) await f.waitForState("connecting");
+  return f;
+}
+
+const note = (state: string, extra: Record<string, unknown> = {}) => ({
+  peer: "studio-mac",
+  state,
+  lastSeen: null,
+  error: null,
+  at: 1,
+  ...extra,
+});
+
+test("a local app snapshots after watch, then closed controls receive the daemon's exact states", async () => {
+  const f = await fixture("local");
+  try {
+    expect(f.requests.indexOf("_acpmux/peers")).toBeGreaterThan(f.requests.indexOf("_acpmux/watch"));
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("connecting");
+    for (const state of ["degraded", "connected", "reconnecting", "offline"]) {
+      f.notify(note(state));
+      expect(f.latest().peerStates?.["studio-mac"]).toBe(state);
+      expect(f.latest().peerConnected?.["studio-mac"]).toBe(state === "connected");
+    }
+    f.notify(note("offline", { error: "removed" }));
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("offline");
+  } finally {
+    f.client.close();
+  }
+});
+
+test("identical lag replay is idempotent and the next state still reaches the client", async () => {
+  const f = await fixture("local");
+  try {
+    const offline = note("offline", { lastSeen: 10, error: "lost" });
+    f.notify(offline);
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("offline");
+    const emitted = f.snapshots.length;
+    f.notify({ ...offline, at: 2 });
+    expect(f.snapshots).toHaveLength(emitted);
+    f.notify(note("connected"));
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("connected");
+  } finally {
+    f.client.close();
+  }
+});
+
+test("a status read in flight cannot erase a newer live state and never polls the event snapshot", async () => {
+  const f = await fixture("local");
+  try {
+    const refreshed = f.client.refreshPeers();
+    f.notify(note("offline", { error: "removed" }));
+    await refreshed;
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("offline");
+    await f.client.refreshPeers();
+    expect(f.requests.filter((method) => method === "_acpmux/peers")).toHaveLength(1);
+  } finally {
+    f.client.close();
+  }
+});
+
+test("a lag notice reads the peer snapshot again, so a dropped disconnect is not lost", async () => {
+  const f = await fixture("local");
+  try {
+    f.notify(note("connected"));
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("connected");
+    // The daemon dropped the offline notice and said so with `_acpmux/lagged`.
+    f.peers([{ name: "studio-mac", state: "offline", lastSeen: 5, error: "lost" }]);
+    const offline = f.waitForState("offline");
+    f.socket().onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/lagged", params: { dropped: 3 } }),
+    });
+    await offline;
+    expect(f.latest().peerConnected?.["studio-mac"]).toBe(false);
+    expect(f.requests.filter((method) => method === "_acpmux/peers")).toHaveLength(2);
+  } finally {
+    f.client.close();
+  }
+});
+
+test("a restarting peer in the snapshot stays restarting, so the picker never offers it as ready", async () => {
+  const { environmentReady, knownMachines } = await import("./environments");
+  const f = await fixture("local");
+  try {
+    f.notify(note("connected"));
+    f.peers([{ name: "studio-mac", state: "restarting", lastSeen: 5, error: null }]);
+    const restarting = f.waitForState("restarting");
+    f.socket().onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/lagged", params: { dropped: 1 } }),
+    });
+    await restarting;
+    const latest = f.latest();
+    expect(latest.peerConnected?.["studio-mac"]).toBe(false);
+    const peers = {
+      kinds: { "studio-mac": "ssh" as const },
+      connected: latest.peerConnected,
+      states: latest.peerStates,
+    };
+    const [machine] = knownMachines(["studio-mac"], [], peers.kinds, [], peers.connected, peers.states);
+    expect(machine).toMatchObject({ state: "restarting", connected: false });
+    expect(environmentReady(machine!)).toBe(false);
+  } finally {
+    f.client.close();
+  }
+});
+
+test("a peer the status lists but the snapshot omits is not offered as ready", async () => {
+  const { environmentReady, knownMachines } = await import("./environments");
+  const f = await fixture("local");
+  try {
+    f.notify(note("offline", { lastSeen: 42, error: "lost" }));
+    // Removed between the `_acpmux/status` read and the `_acpmux/peers` read (or an invalid row).
+    f.peers([]);
+    const omitted = f.waitForState(undefined);
+    f.socket().onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/lagged", params: { dropped: 1 } }),
+    });
+    await omitted;
+    const latest = f.latest();
+    expect(latest.peerConnected?.["studio-mac"]).toBe(false);
+    const peers = {
+      kinds: { "studio-mac": "ssh" as const },
+      connected: latest.peerConnected,
+      states: latest.peerStates,
+    };
+    const [machine] = knownMachines(["studio-mac"], [], peers.kinds, [], peers.connected, peers.states);
+    expect(environmentReady(machine!)).toBe(false);
+    // The offline banner (chatUnavailable) still reads what the status read last said of it.
+    const status = (f.client as unknown as { peerStatus: Record<string, unknown> }).peerStatus["studio-mac"];
+    expect(status).toMatchObject({ connected: false, lastSeen: 42, error: "lost" });
+    expect(status).not.toHaveProperty("state");
+  } finally {
+    f.client.close();
+  }
+});
+
+test("the first frame does not wait on the peer snapshot", async () => {
+  // A held `_acpmux/peers` would hold `connect` if the snapshot sat on the first frame's path,
+  // after the one round trip #27755 keeps it to.
+  const f = await fixture("local", true, true);
+  try {
+    expect(f.requests).toContain("_acpmux/peers");
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("connected");
+    const snapshot = f.waitForState("connecting");
+    f.releasePeers();
+    await snapshot;
+  } finally {
+    f.client.close();
+  }
+});
+
+test("a later status read replaces a restarting state no note has updated since", async () => {
+  // A failed restart goes restarting -> reconnecting, both `reconnecting` on the wire, so the daemon
+  // sends no note; the next `_acpmux/status` read is what says so.
+  const f = await fixture("local");
+  try {
+    f.peers([{ name: "studio-mac", state: "restarting", lastSeen: 5, error: null }]);
+    const restarting = f.waitForState("restarting");
+    f.socket().onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/lagged", params: { dropped: 1 } }),
+    });
+    await restarting;
+    f.status([
+      { name: "studio-mac", transport: "ssh", connected: false, state: "reconnecting", lastSeen: 5, attempts: 1 },
+    ]);
+    await f.client.refreshPeers();
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("reconnecting");
+    expect(f.latest().peerConnected?.["studio-mac"]).toBe(false);
+  } finally {
+    f.client.close();
+  }
+});
+
+test("an older snapshot that lands after a newer status read never brings back its state", async () => {
+  const f = await fixture("local");
+  try {
+    f.peers([{ name: "studio-mac", state: "restarting", lastSeen: 5, error: null }]);
+    f.hold("_acpmux/peers");
+    const snapshot = f.readSnapshot();
+    // Started after the snapshot and answered first: the restart failed meanwhile.
+    f.status([{ name: "studio-mac", transport: "ssh", connected: false, state: "reconnecting", lastSeen: 5 }]);
+    await f.client.refreshPeers();
+    f.release("_acpmux/peers");
+    await snapshot;
+    expect(f.peerStates()["studio-mac"]).toBe("reconnecting");
+  } finally {
+    f.client.close();
+  }
+});
+
+test("an older status read that lands after a newer snapshot never erases it", async () => {
+  const f = await fixture("local");
+  try {
+    f.hold("_acpmux/status");
+    const status = f.client.refreshPeers();
+    // Started after the status read and answered first.
+    f.peers([{ name: "studio-mac", state: "restarting", lastSeen: 5, error: null }]);
+    await f.readSnapshot();
+    f.status([{ name: "studio-mac", transport: "ssh", connected: false, state: "reconnecting", lastSeen: 5 }]);
+    f.release("_acpmux/status");
+    await status;
+    expect(f.peerStates()["studio-mac"]).toBe("restarting");
+  } finally {
+    f.client.close();
+  }
+});
+
+test("an older status read that lands after a newer one never overwrites it", async () => {
+  const f = await fixture("local");
+  try {
+    f.status([{ name: "studio-mac", transport: "ssh", connected: false, state: "offline", lastSeen: 1 }]);
+    const resume = f.hold("_acpmux/status");
+    const older = f.client.refreshPeers();
+    f.free("_acpmux/status");
+    // Asked later, answered first: the machine came back.
+    f.status([{ name: "studio-mac", transport: "ssh", connected: true, state: "connected", lastSeen: 50 }]);
+    await f.client.refreshPeers();
+    resume();
+    await older;
+    expect(f.peerStates()["studio-mac"]).toBe("connected");
+    expect(f.peerConnected()["studio-mac"]).toBe(true);
+    expect(f.peerStatus()["studio-mac"]).toMatchObject({ connected: true, state: "connected", lastSeen: 50 });
+  } finally {
+    f.client.close();
+  }
+});
+
+for (const [origin, supported] of [
+  ["peer", true],
+  ["remote", true],
+  [undefined, true],
+  ["local", false],
+] as const)
+  test(`origin=${origin}, feature=${supported} retains its status snapshot without a peer-event subscription`, async () => {
+    const f = await fixture(origin, supported);
+    try {
+      f.notify(note("offline"));
+      expect(f.latest().peerStates?.["studio-mac"]).toBe("connected");
+      expect(f.requests).not.toContain("_acpmux/peers");
+    } finally {
+      f.client.close();
+    }
+  });
+
+test("reconnect reads a fresh snapshot and ignores late notifications from the old socket", async () => {
+  const f = await fixture("local");
+  let reconnect: (() => void) | undefined;
+  const timer = spyOn(dom.window, "setTimeout").mockImplementation((run) => {
+    if (typeof run === "function") reconnect = () => run();
+    return 1;
+  });
+  try {
+    const old = f.socket();
+    f.notify(note("offline", { error: "removed" }));
+    f.peers([{ name: "studio-mac", state: "connected", lastSeen: null, error: null }]);
+    old.onclose?.();
+    const connected = f.waitForState("connected");
+    expect(reconnect).toBeDefined();
+    reconnect!();
+    await connected;
+    expect(f.requests.filter((method) => method === "_acpmux/peers")).toHaveLength(2);
+    const emitted = f.snapshots.length;
+    old.onmessage?.({
+      data: JSON.stringify({ jsonrpc: "2.0", method: "_acpmux/peer_state_changed", params: note("offline") }),
+    });
+    expect(f.latest().peerStates?.["studio-mac"]).toBe("connected");
+    expect(f.snapshots).toHaveLength(emitted);
+  } finally {
+    timer.mockRestore();
+    f.client.close();
+  }
+});
diff --git a/webviews/src/agent-session/acpmux/peerStateEvents.ts b/webviews/src/agent-session/acpmux/peerStateEvents.ts
new file mode 100644
index 000000000000..d642dc6e9b81
--- /dev/null
+++ b/webviews/src/agent-session/acpmux/peerStateEvents.ts
@@ -0,0 +1,124 @@
+import type { PeerLinkState } from "./protocol/generated/types";
+
+/// Every state acpmux sends (peer_link.rs `PeerState::as_str`). Notes carry the protocol's
+/// `PeerLinkState`; `_acpmux/peers` also says `restarting` while the peer's daemon restarts, and a
+/// snapshot that dropped that row would leave the picker no state, so it would read the peer as ready.
+type LinkState = PeerLinkState | "restarting";
+const STATES = new Set<unknown>(["connecting", "connected", "degraded", "reconnecting", "restarting", "offline"]);
+type Link = { state: LinkState; connected: boolean; lastSeen: number | undefined; error: string | undefined };
+type VersionedLink = { link: Link; revision: number };
+const object = (value: unknown): Record<string, unknown> =>
+  value !== null && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
+const same = (a: Link | undefined, b: Link) => a?.state === b.state && a.lastSeen === b.lastSeen && a.error === b.error;
+
+/// One local/app connection's snapshot and pushed state. The daemon names both proven origins
+/// `local`; a peer daemon's connection never hears this host's peer inventory (#26860).
+export class PeerStateEvents {
+  supported = false;
+  authoritative = false;
+  /// One clock for notes and for the start of every read, status and snapshot alike: an entry keeps
+  /// the tick it was written at, and a read that lands is ordered against the others by the tick it
+  /// started at, so neither kind of read brings back or erases what a newer one said.
+  private clock = 0;
+  /// The newest snapshot read started; only that one may land.
+  private latestRead = 0;
+  /// The start of the newest status read and of the newest snapshot that landed.
+  private statusLanded = 0;
+  private snapshotLanded = 0;
+  private links = new Map<string, VersionedLink>();
+
+  configure(initialized: unknown): void {
+    const acpmux = object(object(object(initialized)._meta).acpmux);
+    this.supported =
+      acpmux.origin === "local" && Array.isArray(acpmux.features) && acpmux.features.includes("peerStateEvents");
+    this.authoritative = false;
+    this.links.clear();
+    // The clock never goes back, so no read from before reads as one from this connection.
+    this.latestRead = 0;
+    this.statusLanded = this.clock;
+    this.snapshotLanded = this.clock;
+  }
+
+  /// Whether anything here overlays a status read.
+  get observed(): boolean {
+    return this.authoritative || this.links.size > 0;
+  }
+
+  changed(value: unknown): boolean {
+    const note = object(value);
+    const link = observation(note);
+    if (
+      !this.supported ||
+      typeof note.peer !== "string" ||
+      !note.peer ||
+      !link ||
+      typeof note.at !== "number" ||
+      !Number.isFinite(note.at)
+    )
+      return false;
+    const previous = this.links.get(note.peer)?.link;
+    // Even a repeated note fences an in-flight stale read, without redrawing identical content.
+    this.links.set(note.peer, { link, revision: ++this.clock });
+    return !same(previous, link);
+  }
+
+  /// A `_acpmux/peers` read starts: its tick, which `snapshot` takes when it lands.
+  readStarted(): number {
+    this.latestRead = ++this.clock;
+    return this.latestRead;
+  }
+
+  /// A `_acpmux/status` read starts: its tick, which `statusRead` takes when it lands.
+  statusStarted(): number {
+    return ++this.clock;
+  }
+
+  /// The snapshot read started at `start` landed. A newer snapshot read, or a status read that
+  /// started later and already landed, says more than it does: it is dropped. Entries written after
+  /// it started (a note, a newer status read's) stay.
+  snapshot(value: unknown, start = this.latestRead): boolean {
+    const peers = object(value).peers;
+    if (!this.supported || start !== this.latestRead || start < this.statusLanded || !Array.isArray(peers))
+      return false;
+    const next = new Map<string, VersionedLink>();
+    for (const value of peers) {
+      const peer = object(value);
+      const link = observation(peer);
+      if (typeof peer.name === "string" && peer.name && link) next.set(peer.name, { link, revision: start });
+    }
+    for (const [name, entry] of this.links) if (entry.revision > start) next.set(name, entry);
+    this.links = next;
+    this.snapshotLanded = start;
+    this.authoritative = true;
+    return true;
+  }
+
+  /// The status read started at `start` landed. One older than a status read that already landed
+  /// (or than this connection) says nothing new: false, and the caller drops it whole. Otherwise it
+  /// is newer than every entry written before it started, so those are dropped and the status read
+  /// speaks for them. A failed restart goes restarting -> reconnecting, both `reconnecting` on the
+  /// wire, so the daemon sends no note and only a status read says so. A snapshot that started after
+  /// it stays the authority.
+  statusRead(start: number): boolean {
+    if (start <= this.statusLanded) return false;
+    this.statusLanded = start;
+    for (const [name, entry] of this.links) if (entry.revision < start) this.links.delete(name);
+    if (start > this.snapshotLanded) this.authoritative = false;
+    return true;
+  }
+
+  *entries(): IterableIterator<[string, Link]> {
+    for (const [name, entry] of this.links) yield [name, entry.link];
+  }
+}
+
+function observation(value: Record<string, unknown>): Link | undefined {
+  if (!STATES.has(value.state)) return undefined;
+  return {
+    state: value.state as LinkState,
+    connected: value.state === "connected",
+    // A full observation clears old nullable values when overlaid on an older status read.
+    lastSeen: typeof value.lastSeen === "number" && Number.isFinite(value.lastSeen) ? value.lastSeen : undefined,
+    error: typeof value.error === "string" ? value.error : undefined,
+  };
+}
diff --git a/webviews/src/agent-session/acpmux/peerStateReducer.test.ts b/webviews/src/agent-session/acpmux/peerStateReducer.test.ts
new file mode 100644
index 000000000000..2595d545bbdc
--- /dev/null
+++ b/webviews/src/agent-session/acpmux/peerStateReducer.test.ts
@@ -0,0 +1,105 @@
+import { expect, test } from "bun:test";
+import { PeerStateEvents } from "./peerStateEvents";
+
+const initialized = (origin = "local", features = ["peerStateEvents"]) => ({ _meta: { acpmux: { origin, features } } });
+const note = (state: string, extra: Record<string, unknown> = {}) => ({
+  peer: "studio-mac",
+  state,
+  lastSeen: null,
+  error: null,
+  at: 1,
+  ...extra,
+});
+function fixture() {
+  const events = new PeerStateEvents();
+  events.configure(initialized());
+  return events;
+}
+const peer = (events: PeerStateEvents) => Object.fromEntries(events.entries())["studio-mac"];
+
+test("nullable and absent metadata clear a prior observation when the consumer overlays it", () => {
+  const events = fixture();
+  events.changed(note("offline", { lastSeen: 10, error: "lost", url: "ssh://private" }));
+  const before = peer(events)!;
+  expect(before.error).toBe("lost");
+  for (const metadata of [{ lastSeen: null, error: null }, {}]) {
+    events.changed({ peer: "studio-mac", state: "connected", at: 2, ...metadata });
+    const overlaid = { ...before, ...peer(events) };
+    expect(overlaid.state).toBe("connected");
+    expect(overlaid.lastSeen).toBeUndefined();
+    expect(overlaid.error).toBeUndefined();
+    expect(overlaid).not.toHaveProperty("url");
+    expect(overlaid).not.toHaveProperty("at");
+  }
+});
+
+test("duplicate replay does not redraw but still fences an in-flight stale snapshot", () => {
+  const events = fixture();
+  expect(events.changed(note("offline", { error: "removed" }))).toBe(true);
+  const read = events.readStarted();
+  expect(events.changed(note("offline", { error: "removed", at: 20 }))).toBe(false);
+  events.snapshot({ peers: [{ name: "studio-mac", state: "connected" }] }, read);
+  expect(peer(events)).toMatchObject({ state: "offline", error: "removed" });
+  expect(events.changed(note("offline", { error: "different" }))).toBe(true);
+  expect(events.changed(note("offline", { error: "different", lastSeen: 5 }))).toBe(true);
+});
+
+test("a reconnect and a later read both invalidate the older read", () => {
+  const events = fixture();
+  const old = events.readStarted();
+  events.configure(initialized());
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "offline" }] }, old)).toBe(false);
+  const first = events.readStarted();
+  const second = events.readStarted();
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "connected" }] }, second)).toBe(true);
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "offline" }] }, first)).toBe(false);
+  expect(peer(events)?.state).toBe("connected");
+});
+
+test("a peer-origin reconnect drops local state and rejects inventory and live notes", () => {
+  const events = fixture();
+  events.changed(note("connected"));
+  events.configure(initialized("peer"));
+  expect(events.supported).toBe(false);
+  expect(events.authoritative).toBe(false);
+  expect(events.changed(note("offline"))).toBe(false);
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "connected" }] }, 0)).toBe(false);
+  expect([...events.entries()]).toEqual([]);
+});
+
+test("invalid notes cannot replace the last daemon observation", () => {
+  const events = fixture();
+  events.changed(note("connected"));
+  for (const value of [null, [], note("invented"), note("offline", { at: NaN }), note("offline", { peer: "" })])
+    expect(events.changed(value)).toBe(false);
+  expect(peer(events)?.state).toBe("connected");
+});
+
+test("a complete empty snapshot removes stale state without inventing an offline transition", () => {
+  const events = fixture();
+  events.changed(note("connected"));
+  const read = events.readStarted();
+  expect(events.snapshot({ peers: [] }, read)).toBe(true);
+  expect([...events.entries()]).toEqual([]);
+  expect(events.authoritative).toBe(true);
+});
+
+test("a new connection has no inherited capability while initialize is pending", () => {
+  const events = fixture();
+  const read = events.readStarted();
+  events.configure(undefined);
+  expect(events.changed(note("offline"))).toBe(false);
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "offline" }] }, read)).toBe(false);
+  expect(events.supported).toBe(false);
+  expect([...events.entries()]).toEqual([]);
+});
+
+test("a snapshot keeps a restarting peer marked restarting and not connected", () => {
+  // `_acpmux/peers` says `restarting` (acpmux peer_link.rs `PeerState::as_str`) while the daemon on
+  // the other end restarts; dropping the row would leave the picker no state and read it as ready.
+  const events = fixture();
+  events.changed(note("connected"));
+  const read = events.readStarted();
+  expect(events.snapshot({ peers: [{ name: "studio-mac", state: "restarting", lastSeen: 7 }] }, read)).toBe(true);
+  expect(peer(events)).toMatchObject({ state: "restarting", connected: false, lastSeen: 7 });
+});
diff --git a/webviews/test/chat-unavailable.test.ts b/webviews/test/chat-unavailable.test.ts
index 30a553aa375a..65f821c04ad4 100644
--- a/webviews/test/chat-unavailable.test.ts
+++ b/webviews/test/chat-unavailable.test.ts
@@ -58,6 +58,10 @@ describe("chatUnavailable", () => {
     expect(
       chatUnavailable({ peers: { box: { connected: false, state: "reconnecting", attempts: 0 } }, waitedMs: 0 }).reason,
     ).toBe("connecting");
+    // A peer whose daemon announced a restart (acpmux peer_link.rs `Restarting`) is on its way back.
+    expect(
+      chatUnavailable({ peers: { box: { connected: false, state: "restarting", attempts: 0 } }, waitedMs: 0 }).reason,
+    ).toBe("connecting");
   });
 
   test("a machine that is down names itself and when it was last seen", () => {