agentmux_srv/backend/storage/
background_tasks.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Durable registry of declared long-running (`run_in_background: true`)
5//! tasks attached to a block. See `db_background_tasks` in migrations.rs
6//! (`OBJECT_SCHEMA_VERSION` v20) and
7//! docs/status/STATUS_ATTACHED_TASK_AXIS_AND_DEV_LOOP_2026_08_15.md.
8//!
9//! This exists because the frontend's own signal chain for "is a
10//! long-running task attached to this pane" — transcript replay →
11//! in-memory reducer slice → `DockSnapshotCache` — is entirely ephemeral:
12//! scoped to one open tab, and (for the srv-side `DockSnapshotCache` mirror)
13//! silently evicted after one hour regardless of whether the task is still
14//! genuinely running. A `task dev` session in this repo's own retros has
15//! run 12+ hours. `id` mirrors the frontend's dock `node_id` (normally the
16//! originating tool_use_id) so rows join cleanly with dock data.
17
18use rusqlite::params;
19
20use super::error::StoreError;
21use super::store::Store;
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum BackgroundTaskStatus {
25    Running,
26    Done,
27    Error,
28    Stopped,
29}
30
31impl BackgroundTaskStatus {
32    pub fn as_str(self) -> &'static str {
33        match self {
34            BackgroundTaskStatus::Running => "running",
35            BackgroundTaskStatus::Done => "done",
36            BackgroundTaskStatus::Error => "error",
37            BackgroundTaskStatus::Stopped => "stopped",
38        }
39    }
40
41    /// Unrecognized values fall back to `Running` — same "never claim
42    /// running-forever silently disappears, never claim a live task is
43    /// terminal on a parse hiccup" posture as the rest of this feature
44    /// area (`isAcceptedBackgroundLaunch`'s "unrecognized → stopped, never
45    /// running-forever" rule is about outgoing classification; this is the
46    /// inverse case of reading a row back, where erring toward "still
47    /// live" is the safer default for a status this registry itself wrote).
48    pub fn from_str(s: &str) -> Self {
49        match s {
50            "done" => BackgroundTaskStatus::Done,
51            "error" => BackgroundTaskStatus::Error,
52            "stopped" => BackgroundTaskStatus::Stopped,
53            _ => BackgroundTaskStatus::Running,
54        }
55    }
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct BackgroundTask {
60    pub id: String,
61    pub block_id: String,
62    pub label: String,
63    pub pid: Option<i64>,
64    pub started_at_ms: i64,
65    pub status: BackgroundTaskStatus,
66    pub last_seen_ms: i64,
67    pub ended_at_ms: Option<i64>,
68    /// The Agent call (`tool_use_id`) whose subagent launched this task, or
69    /// `None` for the agent's own. Schema v41.
70    pub owner_tool_use_id: Option<String>,
71}
72
73const COLUMNS: &str =
74    "id, block_id, label, pid, started_at_ms, status, last_seen_ms, ended_at_ms, owner_tool_use_id";
75
76impl Store {
77    /// Create the row if it doesn't exist yet (status `running`), and
78    /// refresh `last_seen_ms` — but only while the row is still `running`.
79    /// Never resurrects a row a concurrent `background_task_complete` call
80    /// already marked terminal; the terminal status is left untouched.
81    pub fn background_task_observe(
82        &self,
83        id: &str,
84        block_id: &str,
85        label: &str,
86        started_at_ms: i64,
87        now_ms: i64,
88    ) -> Result<(), StoreError> {
89        let conn = self.conn.lock().unwrap();
90        conn.execute(
91            "INSERT OR IGNORE INTO db_background_tasks
92                (id, block_id, label, pid, started_at_ms, status, last_seen_ms, ended_at_ms)
93             VALUES (?1, ?2, ?3, NULL, ?4, 'running', ?5, NULL)",
94            params![id, block_id, label, started_at_ms, now_ms],
95        )?;
96        conn.execute(
97            "UPDATE db_background_tasks SET last_seen_ms = ?1 WHERE id = ?2 AND status = 'running'",
98            params![now_ms, id],
99        )?;
100        Ok(())
101    }
102
103    /// Record the OS pid once known (bashwrap/process_tracker learn it
104    /// after the task has already been observed as declared-background).
105    /// No-op (returns `false`) if the row doesn't exist.
106    pub fn background_task_set_pid(&self, id: &str, pid: i64) -> Result<bool, StoreError> {
107        let conn = self.conn.lock().unwrap();
108        let rows = conn.execute(
109            "UPDATE db_background_tasks SET pid = ?1 WHERE id = ?2",
110            params![pid, id],
111        )?;
112        Ok(rows > 0)
113    }
114
115    /// Mark a task terminal. Idempotent — calling this again on an already-
116    /// terminal row just overwrites the status/ended_at (last write wins),
117    /// which only matters if two different terminal signals race, an edge
118    /// case not worth a CAS for a status transition with no live consumer
119    /// depending on "first terminal write wins" today.
120    pub fn background_task_complete(
121        &self,
122        id: &str,
123        status: BackgroundTaskStatus,
124        ended_at_ms: i64,
125    ) -> Result<bool, StoreError> {
126        let conn = self.conn.lock().unwrap();
127        let rows = conn.execute(
128            "UPDATE db_background_tasks
129             SET status = ?1, ended_at_ms = ?2, last_seen_ms = ?2
130             WHERE id = ?3",
131            params![status.as_str(), ended_at_ms, id],
132        )?;
133        Ok(rows > 0)
134    }
135
136    pub fn background_task_get(&self, id: &str) -> Result<Option<BackgroundTask>, StoreError> {
137        let conn = self.conn.lock().unwrap();
138        let mut stmt = conn.prepare(&format!("SELECT {COLUMNS} FROM db_background_tasks WHERE id = ?1"))?;
139        let mut rows = stmt.query_map(params![id], map_row)?;
140        match rows.next() {
141            Some(r) => Ok(Some(r?)),
142            None => Ok(None),
143        }
144    }
145
146    pub fn background_task_list_for_block(&self, block_id: &str) -> Result<Vec<BackgroundTask>, StoreError> {
147        let conn = self.conn.lock().unwrap();
148        let mut stmt = conn.prepare(&format!(
149            "SELECT {COLUMNS} FROM db_background_tasks WHERE block_id = ?1 ORDER BY started_at_ms ASC"
150        ))?;
151        let iter = stmt.query_map(params![block_id], map_row)?;
152        iter.collect::<Result<Vec<_>, _>>().map_err(Into::into)
153    }
154
155    /// Global — every still-`running` row across every block. Used by
156    /// Swarm-style fleet views and reconnect/history-restore, where the
157    /// caller doesn't yet know which block(s) to ask about.
158    pub fn background_task_list_running(&self) -> Result<Vec<BackgroundTask>, StoreError> {
159        let conn = self.conn.lock().unwrap();
160        let mut stmt = conn.prepare(&format!(
161            "SELECT {COLUMNS} FROM db_background_tasks WHERE status = 'running' ORDER BY started_at_ms ASC"
162        ))?;
163        let iter = stmt.query_map([], map_row)?;
164        iter.collect::<Result<Vec<_>, _>>().map_err(Into::into)
165    }
166
167    /// Every block's `running` rows plus those that ended at or after
168    /// `ended_since_ms` — what a fleet view (Swarm) shows: live work, and
169    /// finished work for as long as it keeps showing a finished row.
170    pub fn background_task_list_fleet(&self, ended_since_ms: i64) -> Result<Vec<BackgroundTask>, StoreError> {
171        let conn = self.conn.lock().unwrap();
172        let mut stmt = conn.prepare(&format!(
173            "SELECT {COLUMNS} FROM db_background_tasks
174             WHERE status = 'running' OR ended_at_ms >= ?1
175             ORDER BY started_at_ms ASC"
176        ))?;
177        let iter = stmt.query_map(params![ended_since_ms], map_row)?;
178        iter.collect::<Result<Vec<_>, _>>().map_err(Into::into)
179    }
180
181    /// Record what the CLI says about a task: its description (replacing a
182    /// generic label such as the renderer's "Bash") and the subagent that
183    /// owns it. The owner is only ever filled in, never cleared or changed.
184    /// No-op (`false`) if the row doesn't exist.
185    pub fn background_task_describe(
186        &self,
187        id: &str,
188        label: &str,
189        owner_tool_use_id: Option<&str>,
190    ) -> Result<bool, StoreError> {
191        let conn = self.conn.lock().unwrap();
192        let rows = conn.execute(
193            "UPDATE db_background_tasks
194             SET label = ?1, owner_tool_use_id = COALESCE(owner_tool_use_id, ?2)
195             WHERE id = ?3",
196            params![label, owner_tool_use_id, id],
197        )?;
198        Ok(rows > 0)
199    }
200}
201
202fn map_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<BackgroundTask> {
203    let status: String = row.get(5)?;
204    Ok(BackgroundTask {
205        id: row.get(0)?,
206        block_id: row.get(1)?,
207        label: row.get(2)?,
208        pid: row.get(3)?,
209        started_at_ms: row.get(4)?,
210        status: BackgroundTaskStatus::from_str(&status),
211        last_seen_ms: row.get(6)?,
212        ended_at_ms: row.get(7)?,
213        owner_tool_use_id: row.get(8)?,
214    })
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220
221    fn object_store() -> Store {
222        Store::open(std::path::Path::new(":memory:")).unwrap()
223    }
224
225    #[test]
226    fn observe_creates_a_running_row() {
227        let store = object_store();
228        store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
229        let task = store.background_task_get("t1").unwrap().unwrap();
230        assert_eq!(task.block_id, "block-a");
231        assert_eq!(task.label, "task dev");
232        assert_eq!(task.status, BackgroundTaskStatus::Running);
233        assert_eq!(task.started_at_ms, 1000);
234        assert_eq!(task.last_seen_ms, 1000);
235        assert_eq!(task.pid, None);
236        assert_eq!(task.ended_at_ms, None);
237    }
238
239    #[test]
240    fn observe_is_idempotent_and_refreshes_last_seen() {
241        let store = object_store();
242        store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
243        store.background_task_observe("t1", "block-a", "task dev", 1000, 5000).unwrap();
244        let task = store.background_task_get("t1").unwrap().unwrap();
245        // started_at_ms is NOT overwritten by a repeat observe (INSERT OR
246        // IGNORE leaves the original row alone) — only last_seen_ms moves.
247        assert_eq!(task.started_at_ms, 1000);
248        assert_eq!(task.last_seen_ms, 5000);
249    }
250
251    #[test]
252    fn observe_never_resurrects_a_completed_task() {
253        let store = object_store();
254        store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
255        store.background_task_complete("t1", BackgroundTaskStatus::Done, 2000).unwrap();
256        // A late/duplicate push racing after completion must not flip the
257        // task back to "running" or move its last_seen_ms.
258        store.background_task_observe("t1", "block-a", "task dev", 1000, 9000).unwrap();
259        let task = store.background_task_get("t1").unwrap().unwrap();
260        assert_eq!(task.status, BackgroundTaskStatus::Done);
261        assert_eq!(task.last_seen_ms, 2000);
262    }
263
264    #[test]
265    fn complete_marks_terminal_and_returns_true_when_row_exists() {
266        let store = object_store();
267        store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
268        let changed = store.background_task_complete("t1", BackgroundTaskStatus::Error, 4000).unwrap();
269        assert!(changed);
270        let task = store.background_task_get("t1").unwrap().unwrap();
271        assert_eq!(task.status, BackgroundTaskStatus::Error);
272        assert_eq!(task.ended_at_ms, Some(4000));
273    }
274
275    #[test]
276    fn complete_returns_false_for_unknown_id() {
277        let store = object_store();
278        assert!(!store.background_task_complete("nope", BackgroundTaskStatus::Done, 1).unwrap());
279    }
280
281    #[test]
282    fn set_pid_updates_existing_row_and_no_ops_for_unknown_id() {
283        let store = object_store();
284        store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
285        assert!(store.background_task_set_pid("t1", 4242).unwrap());
286        assert_eq!(store.background_task_get("t1").unwrap().unwrap().pid, Some(4242));
287        assert!(!store.background_task_set_pid("nope", 1).unwrap());
288    }
289
290    #[test]
291    fn get_returns_none_for_unknown_id() {
292        let store = object_store();
293        assert!(store.background_task_get("nope").unwrap().is_none());
294    }
295
296    #[test]
297    fn list_for_block_only_returns_that_blocks_tasks_in_start_order() {
298        let store = object_store();
299        store.background_task_observe("t1", "block-a", "first", 1000, 1000).unwrap();
300        store.background_task_observe("t2", "block-b", "other-block", 1500, 1500).unwrap();
301        store.background_task_observe("t3", "block-a", "second", 2000, 2000).unwrap();
302        let tasks = store.background_task_list_for_block("block-a").unwrap();
303        let ids: Vec<&str> = tasks.iter().map(|t| t.id.as_str()).collect();
304        assert_eq!(ids, vec!["t1", "t3"]);
305    }
306
307    #[test]
308    fn list_running_excludes_terminal_tasks_across_all_blocks() {
309        let store = object_store();
310        store.background_task_observe("t1", "block-a", "still running", 1000, 1000).unwrap();
311        store.background_task_observe("t2", "block-b", "finished", 1000, 1000).unwrap();
312        store.background_task_complete("t2", BackgroundTaskStatus::Done, 2000).unwrap();
313        let running = store.background_task_list_running().unwrap();
314        let ids: Vec<&str> = running.iter().map(|t| t.id.as_str()).collect();
315        assert_eq!(ids, vec!["t1"]);
316    }
317
318    #[test]
319    fn status_round_trips_through_str_conversion() {
320        for s in [
321            BackgroundTaskStatus::Running,
322            BackgroundTaskStatus::Done,
323            BackgroundTaskStatus::Error,
324            BackgroundTaskStatus::Stopped,
325        ] {
326            assert_eq!(BackgroundTaskStatus::from_str(s.as_str()), s);
327        }
328    }
329
330    #[test]
331    fn unrecognized_status_string_defaults_to_running() {
332        assert_eq!(BackgroundTaskStatus::from_str("garbage"), BackgroundTaskStatus::Running);
333    }
334
335    /// Survives a fresh `Store::open` against the same file — proves the
336    /// row is actually durable (the whole point of this table existing
337    /// instead of another in-memory cache), not just readable within one
338    /// connection's lifetime.
339    #[test]
340    fn survives_reopening_the_store() {
341        let tmp = tempfile::NamedTempFile::new().unwrap();
342        {
343            let store = Store::open(tmp.path()).unwrap();
344            store.background_task_observe("t1", "block-a", "task dev", 1000, 1000).unwrap();
345        }
346        let reopened = Store::open(tmp.path()).unwrap();
347        let task = reopened.background_task_get("t1").unwrap().unwrap();
348        assert_eq!(task.label, "task dev");
349        assert_eq!(task.status, BackgroundTaskStatus::Running);
350    }
351
352    #[test]
353    fn describe_replaces_the_label_and_fills_the_owner_once() {
354        let store = object_store();
355        store.background_task_observe("t1", "block-a", "Bash", 1000, 1000).unwrap();
356        assert_eq!(store.background_task_get("t1").unwrap().unwrap().owner_tool_use_id, None);
357
358        assert!(store.background_task_describe("t1", "Background wait", Some("toolu_agent")).unwrap());
359        let task = store.background_task_get("t1").unwrap().unwrap();
360        assert_eq!((task.label.as_str(), task.owner_tool_use_id.as_deref()), ("Background wait", Some("toolu_agent")));
361
362        // A later description without an owner, or with a different one,
363        // never clears or moves the owner.
364        store.background_task_describe("t1", "Background wait", None).unwrap();
365        store.background_task_describe("t1", "Background wait", Some("toolu_other")).unwrap();
366        assert_eq!(store.background_task_get("t1").unwrap().unwrap().owner_tool_use_id.as_deref(), Some("toolu_agent"));
367
368        assert!(!store.background_task_describe("missing", "x", None).unwrap());
369    }
370
371    #[test]
372    fn list_fleet_returns_running_rows_and_recently_ended_ones_across_blocks() {
373        let store = object_store();
374        store.background_task_observe("live", "block-a", "live", 100, 100).unwrap();
375        store.background_task_observe("recent", "block-b", "recent", 110, 110).unwrap();
376        store.background_task_observe("old", "block-b", "old", 50, 50).unwrap();
377        store.background_task_complete("recent", BackgroundTaskStatus::Done, 900).unwrap();
378        store.background_task_complete("old", BackgroundTaskStatus::Error, 200).unwrap();
379
380        let ids: Vec<String> = store.background_task_list_fleet(500).unwrap().into_iter().map(|t| t.id).collect();
381        assert_eq!(ids, vec!["live".to_string(), "recent".to_string()]);
382    }
383}
agentmux_srv/backend/storage/background_tasks.rs

agentmux_srv/backend/storage/
background_tasks.rs
