agentmux_srv/backend/dock_snapshot.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! In-memory, latest-per-block cache of Activity Dock `ToolNode` status
5//! deltas, pushed from the frontend as they happen. Backs `muxspect dock`
6//! (read) and `muxspect dock clear` (write).
7//!
8//! See `docs/specs/SPEC_MUXSPECT_DOCK_DIAGNOSIS_AND_REMEDIATION_2026_08_06.md`
9//! §3.1. Never persisted — this mirrors the dock's own nature: if no
10//! renderer is currently attached to a block, there is nothing live to
11//! report, and losing this cache on restart is correct, not a bug.
12
13use std::collections::HashMap;
14
15use parking_lot::Mutex;
16use serde::Serialize;
17
18/// One `ToolNode`'s last-known status, as reported by the renderer that
19/// owns it. `observed_at` is stamped on receipt by the server (not trusted
20/// from the client) so `muxspect dock`'s age computation is consistent
21/// even if the renderer's clock is skewed.
22#[derive(Debug, Clone, Serialize)]
23pub struct DockNodeSnapshot {
24 pub node_id: String,
25 pub tool_name: String,
26 pub status: String,
27 /// `ToolNode.timestamp` (ms) — when the tool call was initiated,
28 /// per the renderer. `None` if the pushing client didn't have one.
29 pub timestamp: Option<i64>,
30 /// Server-side receipt time (ms) — used for age, not `timestamp`,
31 /// so a slow/backed-up push doesn't understate how stale the
32 /// snapshot itself is.
33 pub observed_at: i64,
34 /// Whether `params.run_in_background` was `true` on this call (Bash
35 /// only; `None` for every other tool). `status` above is the RAW
36 /// `ToolNode` status, which for an accepted background launch is
37 /// terminal ("success") within ~a second — it does NOT reflect
38 /// whether the dock is still showing this row as `running` while it
39 /// awaits a `<task-notification>` (that reclassification happens
40 /// entirely client-side in `tool-adapter.ts`'s `toolActivities`,
41 /// which the server has no visibility into). This flag exists so a
42 /// human reading `muxspect dock` output isn't misled by `status:
43 /// success` into assuming the dock row is finished too — see issue
44 /// #2518, where exactly that gap let 11 of 17 backgrounded calls in
45 /// one session get stuck `running` in the actual UI while this raw
46 /// snapshot showed them as long-since `success`.
47 pub run_in_background: Option<bool>,
48}
49
50/// One block's tracked nodes, keyed by node id so a later delta for the
51/// same node overwrites rather than accumulates.
52type BlockNodes = HashMap<String, DockNodeSnapshot>;
53
54/// Defense-in-depth cap on how long an entry can survive without a fresh
55/// push before `get()` evicts it, regardless of status. The real fix for
56/// "an already-resolved node never gets reported" is pushing the
57/// resolution back (see `agent-document-store.ts`'s `pushResolvedDockNodes`,
58/// wired for every known resolution path — streaming AND the
59/// SessionEnd/HistoryLoaded/HistoryRestored/ScrubOrphanedInProgress scrub
60/// paths). This is the backstop for any path that isn't one of those
61/// (reagentx P1 on PR #2432 flagged the total absence of any bound at
62/// all — `evict_block` existed for whole-block removal but nothing
63/// time-based existed per-node). One hour comfortably exceeds any
64/// legitimate single tool-call duration.
65const MAX_NODE_AGE_MS: i64 = 60 * 60 * 1000;
66
67#[derive(Default)]
68pub struct DockSnapshotCache {
69 inner: Mutex<HashMap<String, BlockNodes>>,
70}
71
72impl DockSnapshotCache {
73 pub fn new() -> Self {
74 Self::default()
75 }
76
77 /// Record (or overwrite) one node's latest status for `block_id`.
78 pub fn push_delta(&self, block_id: &str, node: DockNodeSnapshot) {
79 let mut guard = self.inner.lock();
80 guard.entry(block_id.to_string()).or_default().insert(node.node_id.clone(), node);
81 }
82
83 /// All tracked nodes for `block_id` younger than `MAX_NODE_AGE_MS` as
84 /// of `now_ms`, or empty if none/unknown block. Lazily evicts anything
85 /// older as a side effect — no background sweep needed, and every read
86 /// self-heals the cache instead of accumulating stale entries forever.
87 pub fn get(&self, block_id: &str, now_ms: i64) -> Vec<DockNodeSnapshot> {
88 let mut guard = self.inner.lock();
89 let Some(nodes) = guard.get_mut(block_id) else { return Vec::new() };
90 nodes.retain(|_, n| now_ms.saturating_sub(n.observed_at) < MAX_NODE_AGE_MS);
91 nodes.values().cloned().collect()
92 }
93
94 /// Whether `node_id` is currently tracked for `block_id` — used to
95 /// validate a `dock clear` request before publishing the clear event,
96 /// so muxspect can distinguish "cleared" from "no such node."
97 pub fn has_node(&self, block_id: &str, node_id: &str) -> bool {
98 let guard = self.inner.lock();
99 guard.get(block_id).is_some_and(|m| m.contains_key(node_id))
100 }
101
102 /// Drop a node from the cache — called once a `dock clear` has been
103 /// published, so a subsequent `dock` read doesn't keep showing a node
104 /// that was just cleared (the clearing renderer flips it to
105 /// `"canceled"` locally, but this cache has no way to observe that
106 /// without another push; removing it here is simpler and correct
107 /// either way — a canceled node isn't "stuck" and doesn't need to
108 /// keep appearing in diagnosis output).
109 pub fn remove_node(&self, block_id: &str, node_id: &str) {
110 let mut guard = self.inner.lock();
111 if let Some(nodes) = guard.get_mut(block_id) {
112 nodes.remove(node_id);
113 }
114 }
115
116 /// Drop every tracked node for `block_id` — for wiring into block
117 /// deletion, so the cache doesn't accumulate entries for blocks that
118 /// no longer exist (open question §6.3 of the spec). Not wired to a
119 /// call site yet — deleting a block is a small footprint (a handful
120 /// of node entries) relative to the cache's overall size, so this is
121 /// a nice-to-have bound, not a correctness requirement; wiring it up
122 /// is a follow-up once the block-deletion path is touched for other
123 /// reasons, same as `ScrubOrphanedInProgress`'s own standalone command
124 /// existing with zero live dispatch call sites today.
125 #[allow(dead_code)]
126 pub fn evict_block(&self, block_id: &str) {
127 self.inner.lock().remove(block_id);
128 }
129}
130
131#[cfg(test)]
132mod tests {
133 use super::*;
134
135 fn node(id: &str, status: &str) -> DockNodeSnapshot {
136 DockNodeSnapshot {
137 node_id: id.to_string(),
138 tool_name: "Bash".to_string(),
139 status: status.to_string(),
140 timestamp: Some(1000),
141 observed_at: 2000,
142 run_in_background: None,
143 }
144 }
145
146 #[test]
147 fn get_on_unknown_block_is_empty() {
148 let cache = DockSnapshotCache::new();
149 assert!(cache.get("no-such-block", 2000).is_empty());
150 }
151
152 #[test]
153 fn push_then_get_round_trips() {
154 let cache = DockSnapshotCache::new();
155 cache.push_delta("block-1", node("n1", "running"));
156 let got = cache.get("block-1", 2000);
157 assert_eq!(got.len(), 1);
158 assert_eq!(got[0].node_id, "n1");
159 assert_eq!(got[0].status, "running");
160 }
161
162 #[test]
163 fn run_in_background_round_trips() {
164 let cache = DockSnapshotCache::new();
165 let mut n = node("n1", "success");
166 n.run_in_background = Some(true);
167 cache.push_delta("block-1", n);
168 assert_eq!(cache.get("block-1", 2000)[0].run_in_background, Some(true));
169 }
170
171 #[test]
172 fn later_delta_for_same_node_overwrites() {
173 let cache = DockSnapshotCache::new();
174 cache.push_delta("block-1", node("n1", "running"));
175 cache.push_delta("block-1", node("n1", "success"));
176 let got = cache.get("block-1", 2000);
177 assert_eq!(got.len(), 1);
178 assert_eq!(got[0].status, "success");
179 }
180
181 #[test]
182 fn different_nodes_coexist() {
183 let cache = DockSnapshotCache::new();
184 cache.push_delta("block-1", node("n1", "running"));
185 cache.push_delta("block-1", node("n2", "success"));
186 assert_eq!(cache.get("block-1", 2000).len(), 2);
187 }
188
189 #[test]
190 fn get_evicts_entries_older_than_max_age() {
191 let cache = DockSnapshotCache::new();
192 cache.push_delta("block-1", node("n1", "running")); // observed_at: 2000
193 // Just under the 1h cap — still present.
194 assert_eq!(cache.get("block-1", 2000 + MAX_NODE_AGE_MS - 1).len(), 1);
195 // At/past the cap — evicted, regardless of status.
196 assert!(cache.get("block-1", 2000 + MAX_NODE_AGE_MS).is_empty());
197 }
198
199 #[test]
200 fn get_eviction_is_per_node_not_whole_block() {
201 let cache = DockSnapshotCache::new();
202 cache.push_delta("block-1", node("old", "running")); // observed_at: 2000
203 let mut fresh = node("fresh", "running");
204 fresh.observed_at = 2000 + MAX_NODE_AGE_MS - 1;
205 cache.push_delta("block-1", fresh);
206 let got = cache.get("block-1", 2000 + MAX_NODE_AGE_MS);
207 assert_eq!(got.len(), 1);
208 assert_eq!(got[0].node_id, "fresh");
209 }
210
211 #[test]
212 fn has_node_reflects_current_state() {
213 let cache = DockSnapshotCache::new();
214 assert!(!cache.has_node("block-1", "n1"));
215 cache.push_delta("block-1", node("n1", "running"));
216 assert!(cache.has_node("block-1", "n1"));
217 assert!(!cache.has_node("block-1", "n2"));
218 assert!(!cache.has_node("block-2", "n1"));
219 }
220
221 #[test]
222 fn remove_node_drops_only_that_node() {
223 let cache = DockSnapshotCache::new();
224 cache.push_delta("block-1", node("n1", "running"));
225 cache.push_delta("block-1", node("n2", "running"));
226 cache.remove_node("block-1", "n1");
227 assert!(!cache.has_node("block-1", "n1"));
228 assert!(cache.has_node("block-1", "n2"));
229 }
230
231 #[test]
232 fn remove_node_on_unknown_block_is_a_no_op() {
233 let cache = DockSnapshotCache::new();
234 cache.remove_node("no-such-block", "n1"); // must not panic
235 }
236
237 #[test]
238 fn evict_block_drops_everything_for_that_block_only() {
239 let cache = DockSnapshotCache::new();
240 cache.push_delta("block-1", node("n1", "running"));
241 cache.push_delta("block-2", node("n1", "running"));
242 cache.evict_block("block-1");
243 assert!(cache.get("block-1", 2000).is_empty());
244 assert_eq!(cache.get("block-2", 2000).len(), 1);
245 }
246}