agentmux_srv/backend/
background_task_feed.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The Claude CLI's own structured task feed → `db_background_tasks`.
5//!
6//! The CLI reports every task it runs on the agent's stdout as `system` lines:
7//!
8//! - `task_started` — `task_id`, `tool_use_id`, `description`, `task_type`
9//!   (`local_bash`, `local_agent`, `local_workflow`), `is_backgrounded`, and
10//!   `owned_by_subagent` when a subagent launched it.
11//! - `task_updated` — `task_id` plus a `patch`: `{is_backgrounded: true}` when a
12//!   running command is moved to the background, `{status, end_time}` when it
13//!   ends (`completed`, `failed`, `killed`).
14//! - `task_notification` — `task_id`, `tool_use_id`, final `status`
15//!   (`completed`, `failed`, `stopped`) and a `summary`.
16//!
17//! These arrive on the PARENT agent's stream even for tasks a subagent owns.
18//! Before this module nothing read them: the registry learned about a
19//! background task only from the renderer spotting the "Command running in
20//! background with ID:" result text, and learned it had ended only from a
21//! `<task-notification>` user message in the same pane. A subagent's task gets
22//! the first but never the second, so its row stayed `running` forever; a
23//! top-level task's result now carries empty `stdout` plus a structured
24//! `backgroundTaskId`, so the text match missed it entirely. See
25//! docs/specs/SPEC_BACKGROUND_TASK_STRUCTURED_FEED_AND_SWARM_OWNERSHIP_2026_09_27.md.
26//!
27//! Scope: backgrounded `local_bash` tasks only, which is what the registry has
28//! always held. Subagents (`local_agent`) are Swarm's subagent watcher's job.
29//! The renderer's existing observe/complete pushes still run; every write here
30//! is the same idempotent `background_task_observe`/`background_task_complete`
31//! call, so whichever arrives first wins and the other is a no-op or a refresh.
32
33use std::collections::HashMap;
34
35use serde_json::Value;
36
37use super::mps::Broker;
38use super::storage::background_tasks::BackgroundTaskStatus;
39use super::storage::store::Store;
40
41/// One task-feed line, reduced to what the registry needs.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub(crate) enum TaskFeedEvent {
44    /// A shell task started. `backgrounded` is false for an ordinary
45    /// foreground Bash call, which can still be backgrounded later.
46    Started {
47        task_id: String,
48        tool_use_id: String,
49        label: String,
50        backgrounded: bool,
51    },
52    /// A running task was moved to the background.
53    Backgrounded { task_id: String },
54    /// A task ended. `tool_use_id` is absent on `task_updated` lines, which
55    /// carry only the `task_id`.
56    Ended {
57        task_id: String,
58        tool_use_id: Option<String>,
59        status: BackgroundTaskStatus,
60    },
61    /// `background_tasks_changed`: the CLI's full list of live tasks.
62    Snapshot { live: Vec<String> },
63    /// A turn boundary (`result`, or a new session segment's `init`), where
64    /// tasks that left the snapshot without an end event are settled.
65    Boundary,
66}
67
68fn str_field<'a>(v: &'a Value, key: &str) -> Option<&'a str> {
69    v.get(key).and_then(Value::as_str).filter(|s| !s.is_empty())
70}
71
72/// Terminal statuses as the CLI spells them. Anything unrecognised ends the
73/// task as `stopped` rather than leaving it `running` — the same lenient rule
74/// the renderer's `parseTaskNotification` applies.
75fn terminal_status(s: &str) -> BackgroundTaskStatus {
76    match s {
77        "completed" => BackgroundTaskStatus::Done,
78        "failed" => BackgroundTaskStatus::Error,
79        _ => BackgroundTaskStatus::Stopped,
80    }
81}
82
83/// Pure parse of one stdout line. `None` for anything that isn't a task-feed
84/// line about a shell task.
85pub(crate) fn parse(line: &Value) -> Option<TaskFeedEvent> {
86    match str_field(line, "type")? {
87        "system" => {}
88        "result" => return Some(TaskFeedEvent::Boundary),
89        _ => return None,
90    }
91    match str_field(line, "subtype")? {
92        "init" => return Some(TaskFeedEvent::Boundary),
93        "background_tasks_changed" => {
94            let live = line
95                .get("tasks")
96                .and_then(Value::as_array)?
97                .iter()
98                .filter_map(|t| str_field(t, "task_id").map(str::to_string))
99                .collect();
100            return Some(TaskFeedEvent::Snapshot { live });
101        }
102        _ => {}
103    }
104    let task_id = str_field(line, "task_id")?.to_string();
105    match str_field(line, "subtype")? {
106        "task_started" => {
107            if str_field(line, "task_type")? != "local_bash" {
108                return None;
109            }
110            let tool_use_id = str_field(line, "tool_use_id")?.to_string();
111            let label = str_field(line, "description").unwrap_or("Bash").to_string();
112            let backgrounded = line
113                .get("is_backgrounded")
114                .and_then(Value::as_bool)
115                .unwrap_or(false);
116            Some(TaskFeedEvent::Started {
117                task_id,
118                tool_use_id,
119                label,
120                backgrounded,
121            })
122        }
123        "task_updated" => {
124            let patch = line.get("patch")?;
125            if let Some(status) = str_field(patch, "status") {
126                if matches!(status, "completed" | "failed" | "killed" | "stopped") {
127                    return Some(TaskFeedEvent::Ended {
128                        task_id,
129                        tool_use_id: None,
130                        status: terminal_status(status),
131                    });
132                }
133                return None;
134            }
135            if patch.get("is_backgrounded").and_then(Value::as_bool) == Some(true) {
136                return Some(TaskFeedEvent::Backgrounded { task_id });
137            }
138            None
139        }
140        "task_notification" => {
141            let status = terminal_status(str_field(line, "status").unwrap_or(""));
142            let tool_use_id = str_field(line, "tool_use_id").map(str::to_string);
143            Some(TaskFeedEvent::Ended {
144                task_id,
145                tool_use_id,
146                status,
147            })
148        }
149        _ => None,
150    }
151}
152
153/// A shell task seen starting on this stream, remembered until it ends so a
154/// later `task_updated` (which names only the `task_id`) can be applied.
155#[derive(Debug, Clone)]
156struct Pending {
157    tool_use_id: String,
158    label: String,
159    started_at_ms: i64,
160    backgrounded: bool,
161    /// The Agent call whose subagent issued this Bash call; `None` for the
162    /// agent's own.
163    owner: Option<String>,
164    /// Listed in at least one `background_tasks_changed` snapshot. Only then
165    /// does its absence from a later one mean anything: most shell tasks are
166    /// never listed at all.
167    in_snapshot: bool,
168    /// Left the snapshot, with no end event yet, at this time. A task that
169    /// ends normally leaves the snapshot one line before its end event, so
170    /// this alone can't end it; it is settled at the next snapshot or turn
171    /// boundary if the end event still hasn't come.
172    vanished_at_ms: Option<i64>,
173}
174
175/// Bound on `TaskFeed::owners`. Entries are normally consumed by the
176/// `task_started` that follows within the same message; this only guards
177/// against a stream shape where that never happens.
178const MAX_UNCLAIMED_OWNERS: usize = 4096;
179
180/// Per-stream state: one per stdout reader. Holds only shell tasks that have
181/// started and not yet ended, plus the owners of subagent Bash calls whose
182/// `task_started` hasn't arrived yet, so it stays small.
183#[derive(Debug, Default)]
184pub(crate) struct TaskFeed {
185    pending: HashMap<String, Pending>,
186    owners: HashMap<String, String>,
187}
188
189impl TaskFeed {
190    /// Apply one parsed stdout line. Returns true when the registry changed,
191    /// in which case `background-task-updated` has been published.
192    pub(crate) fn apply(
193        &mut self,
194        store: &Store,
195        broker: Option<&Broker>,
196        block_id: &str,
197        line: &Value,
198        now_ms: i64,
199    ) -> bool {
200        self.note_owner(line);
201        let Some(event) = parse(line) else {
202            return false;
203        };
204        let changed = match event {
205            TaskFeedEvent::Started {
206                task_id,
207                tool_use_id,
208                label,
209                backgrounded,
210            } => {
211                let owner = self.owners.remove(&tool_use_id);
212                let observed = backgrounded
213                    && observe(
214                        store,
215                        block_id,
216                        &tool_use_id,
217                        &label,
218                        owner.as_deref(),
219                        now_ms,
220                        now_ms,
221                    );
222                self.pending.insert(
223                    task_id,
224                    Pending {
225                        tool_use_id,
226                        label,
227                        started_at_ms: now_ms,
228                        backgrounded,
229                        owner,
230                        in_snapshot: false,
231                        vanished_at_ms: None,
232                    },
233                );
234                observed
235            }
236            TaskFeedEvent::Backgrounded { task_id } => match self.pending.get_mut(&task_id) {
237                Some(p) if !p.backgrounded => {
238                    p.backgrounded = true;
239                    // Started when the CLI first reported it; seen now.
240                    observe(
241                        store,
242                        block_id,
243                        &p.tool_use_id,
244                        &p.label,
245                        p.owner.as_deref(),
246                        p.started_at_ms,
247                        now_ms,
248                    )
249                }
250                _ => false,
251            },
252            TaskFeedEvent::Ended {
253                task_id,
254                tool_use_id,
255                status,
256            } => {
257                let pending = self.pending.remove(&task_id);
258                // Only a task that was ever backgrounded can have a row. A
259                // `task_notification` carries its own tool_use_id, which also
260                // covers a task that started before this reader did (an
261                // srv restart); completing a row that doesn't exist is a no-op.
262                let id = tool_use_id
263                    .or_else(|| pending.filter(|p| p.backgrounded).map(|p| p.tool_use_id));
264                match id {
265                    Some(id) => complete_if_running(store, block_id, &id, status, now_ms),
266                    None => false,
267                }
268            }
269            TaskFeedEvent::Snapshot { live } => {
270                // Settle what already vanished in an earlier snapshot and is
271                // still missing, before marking what vanished in this one.
272                let changed = self.settle_vanished(store, block_id, Some(live.as_slice()));
273                for (task_id, p) in self.pending.iter_mut() {
274                    if live.contains(task_id) {
275                        p.in_snapshot = true;
276                        p.vanished_at_ms = None;
277                    } else if p.in_snapshot && p.vanished_at_ms.is_none() {
278                        p.vanished_at_ms = Some(now_ms);
279                    }
280                }
281                changed
282            }
283            TaskFeedEvent::Boundary => self.settle_vanished(store, block_id, None),
284        };
285        if changed {
286            if let Some(broker) = broker {
287                publish_background_task_updated(broker, block_id);
288            }
289        }
290        changed
291    }
292
293    /// End every task that left the snapshot and got no end event since: the
294    /// CLI stopped tracking it without saying so (a subagent's task, dropped
295    /// when the CLI started a new session segment, per the 2026-09-30 retro).
296    /// With `still_live`, a vanished task listed again is kept. The row ends
297    /// `stopped`, at the time it left the snapshot.
298    fn settle_vanished(
299        &mut self,
300        store: &Store,
301        block_id: &str,
302        still_live: Option<&[String]>,
303    ) -> bool {
304        let settled: Vec<String> = self
305            .pending
306            .iter()
307            .filter(|(task_id, p)| {
308                p.vanished_at_ms.is_some() && !still_live.is_some_and(|live| live.contains(task_id))
309            })
310            .map(|(task_id, _)| task_id.clone())
311            .collect();
312        let mut changed = false;
313        for task_id in settled {
314            let Some(p) = self.pending.remove(&task_id) else {
315                continue;
316            };
317            if p.backgrounded {
318                let at = p.vanished_at_ms.unwrap_or_default();
319                changed |= complete_if_running(
320                    store,
321                    block_id,
322                    &p.tool_use_id,
323                    BackgroundTaskStatus::Stopped,
324                    at,
325                );
326            }
327        }
328        changed
329    }
330
331    /// A subagent's own stream lines carry `parent_tool_use_id` = the Agent
332    /// call that spawned it (also that subagent's `meta.json` `toolUseId`).
333    /// Remember it for each Bash call such a line issues; the call's
334    /// `task_started`, which follows in the same message, claims it.
335    fn note_owner(&mut self, line: &Value) {
336        if str_field(line, "type") != Some("assistant") {
337            return;
338        }
339        let Some(parent) = str_field(line, "parent_tool_use_id") else {
340            return;
341        };
342        let Some(blocks) = line.pointer("/message/content").and_then(Value::as_array) else {
343            return;
344        };
345        for block in blocks {
346            if str_field(block, "type") == Some("tool_use")
347                && str_field(block, "name") == Some("Bash")
348            {
349                if let Some(id) = str_field(block, "id") {
350                    if self.owners.len() >= MAX_UNCLAIMED_OWNERS {
351                        self.owners.clear();
352                    }
353                    self.owners.insert(id.to_string(), parent.to_string());
354                }
355            }
356        }
357    }
358}
359
360/// Complete `tool_use_id`'s row if it is still running. The CLI usually
361/// reports an end twice (`task_updated`, then `task_notification`); only a
362/// still-running row is completed, so the first end time stands and
363/// subscribers hear it once.
364fn complete_if_running(
365    store: &Store,
366    block_id: &str,
367    tool_use_id: &str,
368    status: BackgroundTaskStatus,
369    at_ms: i64,
370) -> bool {
371    let running = matches!(store.background_task_get(tool_use_id),
372        Ok(Some(row)) if row.status == BackgroundTaskStatus::Running);
373    if !running {
374        return false;
375    }
376    match store.background_task_complete(tool_use_id, status, at_ms) {
377        Ok(changed) => changed,
378        Err(e) => {
379            tracing::warn!(target: "background_tasks", block_id, node_id = %tool_use_id, error = %e,
380                "failed to complete a background task from the CLI's task feed");
381            false
382        }
383    }
384}
385
386/// Create the row if needed, then record the CLI's description as its label
387/// (replacing the renderer's generic "Bash" if it got there first) and its
388/// owner.
389fn observe(
390    store: &Store,
391    block_id: &str,
392    tool_use_id: &str,
393    label: &str,
394    owner: Option<&str>,
395    started_at_ms: i64,
396    seen_at_ms: i64,
397) -> bool {
398    let result = store
399        .background_task_observe(tool_use_id, block_id, label, started_at_ms, seen_at_ms)
400        .and_then(|()| {
401            store
402                .background_task_describe(tool_use_id, label, owner)
403                .map(|_| ())
404        });
405    match result {
406        Ok(()) => true,
407        Err(e) => {
408            tracing::warn!(target: "background_tasks", block_id, node_id = %tool_use_id, error = %e,
409                "failed to record a background task from the CLI's task feed");
410            false
411        }
412    }
413}
414
415/// Notify subscribers that `block_id`'s `db_background_tasks` state changed
416/// (observed, pid recorded, or completed), so the frontend can re-query
417/// `COMMAND_LIST_BACKGROUND_TASKS` instead of polling. Live-only (`persist:
418/// 0`) — a late subscriber gets the current state via the mount-time list
419/// query, not event replay. Deliberately carries no task data itself (just an
420/// invalidation signal): the list query is the single source of truth for the
421/// actual rows. See
422/// docs/specs/SPEC_BACKGROUND_TASK_DASHBOARD_INTELLIGENCE_2026_08_20.md §3.2.
423pub(crate) fn publish_background_task_updated(broker: &Broker, block_id: &str) {
424    broker.publish_event::<crate::backend::mps_events::BackgroundTaskUpdated>(&crate::backend::mps_events::BlockRefEvent { block_id: block_id.to_string() }, vec![format!("block:{block_id}")], 0);
425}
426
427#[cfg(test)]
428mod tests {
429    use super::*;
430    use serde_json::json;
431
432    fn store() -> Store {
433        Store::open(std::path::Path::new(":memory:")).unwrap()
434    }
435
436    // Shapes copied from a real Claude CLI stream (2026-09-26), trimmed of
437    // uuid/session_id.
438    fn started(task_id: &str, tool: &str, bg: bool, subagent: bool) -> Value {
439        json!({"type":"system","subtype":"task_started","task_id":task_id,"owned_by_subagent":subagent,
440               "tool_use_id":tool,"description":"Background wait for agents","is_backgrounded":bg,"task_type":"local_bash"})
441    }
442    fn notification(task_id: &str, tool: &str, status: &str) -> Value {
443        json!({"type":"system","subtype":"task_notification","task_id":task_id,"tool_use_id":tool,"status":status,
444               "output_file":"/tmp/x.output","summary":"Background command \"Background wait for agents\" completed (exit code 0)"})
445    }
446    fn updated(task_id: &str, patch: Value) -> Value {
447        json!({"type":"system","subtype":"task_updated","task_id":task_id,"patch":patch})
448    }
449
450    #[test]
451    fn parses_a_backgrounded_shell_start() {
452        assert_eq!(
453            parse(&started("b6p74mfn6", "toolu_1", true, true)),
454            Some(TaskFeedEvent::Started {
455                task_id: "b6p74mfn6".into(),
456                tool_use_id: "toolu_1".into(),
457                label: "Background wait for agents".into(),
458                backgrounded: true,
459            })
460        );
461    }
462
463    #[test]
464    fn ignores_subagent_tasks_and_non_task_lines() {
465        let agent = json!({"type":"system","subtype":"task_started","task_id":"a1","tool_use_id":"toolu_a",
466                           "description":"Map code","task_type":"local_agent","is_backgrounded":true});
467        assert_eq!(parse(&agent), None);
468        // `init` isn't a task line, but it is a turn boundary: see
469        // `TaskFeedEvent::Boundary`.
470        assert_eq!(
471            parse(&json!({"type":"system","subtype":"init","session_id":"s"})),
472            Some(TaskFeedEvent::Boundary)
473        );
474        assert_eq!(parse(&json!({"type":"assistant","message":{}})), None);
475        assert_eq!(parse(&updated("t", json!({"status":"running"}))), None);
476    }
477
478    #[test]
479    fn maps_every_terminal_status() {
480        for (cli, want) in [
481            ("completed", BackgroundTaskStatus::Done),
482            ("failed", BackgroundTaskStatus::Error),
483            ("stopped", BackgroundTaskStatus::Stopped),
484            ("something-new", BackgroundTaskStatus::Stopped),
485        ] {
486            match parse(&notification("t", "toolu_x", cli)) {
487                Some(TaskFeedEvent::Ended { status, .. }) => assert_eq!(status, want, "{cli}"),
488                other => panic!("{cli}: {other:?}"),
489            }
490        }
491        assert!(matches!(
492            parse(&updated("t", json!({"status":"killed","end_time":1}))),
493            Some(TaskFeedEvent::Ended {
494                status: BackgroundTaskStatus::Stopped,
495                tool_use_id: None,
496                ..
497            })
498        ));
499    }
500
501    /// The stuck-row incident: a subagent's background task starts and ends on
502    /// the parent stream, and the row must end with it.
503    #[test]
504    fn a_subagent_owned_background_task_is_recorded_then_completed() {
505        let store = store();
506        let mut feed = TaskFeed::default();
507        assert!(feed.apply(
508            &store,
509            None,
510            "blk",
511            &started("b6p74mfn6", "toolu_1", true, true),
512            1_000
513        ));
514        let row = store.background_task_get("toolu_1").unwrap().unwrap();
515        assert_eq!(
516            (row.status, row.label.as_str(), row.block_id.as_str()),
517            (
518                BackgroundTaskStatus::Running,
519                "Background wait for agents",
520                "blk"
521            )
522        );
523
524        assert!(feed.apply(
525            &store,
526            None,
527            "blk",
528            &notification("b6p74mfn6", "toolu_1", "completed"),
529            181_000
530        ));
531        let row = store.background_task_get("toolu_1").unwrap().unwrap();
532        assert_eq!(
533            (row.status, row.ended_at_ms),
534            (BackgroundTaskStatus::Done, Some(181_000))
535        );
536    }
537
538    #[test]
539    fn a_foreground_call_never_gets_a_row() {
540        let store = store();
541        let mut feed = TaskFeed::default();
542        assert!(!feed.apply(
543            &store,
544            None,
545            "blk",
546            &started("t1", "toolu_fg", false, false),
547            1
548        ));
549        assert!(!feed.apply(
550            &store,
551            None,
552            "blk",
553            &notification("t1", "toolu_fg", "completed"),
554            2
555        ));
556        assert!(store.background_task_get("toolu_fg").unwrap().is_none());
557        assert!(feed.pending.is_empty(), "ended tasks are forgotten");
558    }
559
560    #[test]
561    fn a_command_moved_to_the_background_gets_a_row_and_its_end_by_task_id() {
562        let store = store();
563        let mut feed = TaskFeed::default();
564        feed.apply(
565            &store,
566            None,
567            "blk",
568            &started("t2", "toolu_mv", false, false),
569            10,
570        );
571        assert!(feed.apply(
572            &store,
573            None,
574            "blk",
575            &updated("t2", json!({"is_backgrounded": true})),
576            20
577        ));
578        let row = store.background_task_get("toolu_mv").unwrap().unwrap();
579        // Started when the CLI first reported it, seen when it was backgrounded.
580        assert_eq!(
581            (row.status, row.started_at_ms, row.last_seen_ms),
582            (BackgroundTaskStatus::Running, 10, 20)
583        );
584
585        // task_updated names only the task_id; the feed resolves it.
586        assert!(feed.apply(
587            &store,
588            None,
589            "blk",
590            &updated("t2", json!({"status":"failed","end_time":30})),
591            30
592        ));
593        assert_eq!(
594            store
595                .background_task_get("toolu_mv")
596                .unwrap()
597                .unwrap()
598                .status,
599            BackgroundTaskStatus::Error
600        );
601    }
602
603    /// A subagent's Bash call, as the parent stream carries it (shape from a
604    /// real recording, trimmed).
605    fn subagent_bash_call(tool: &str, parent: &str) -> Value {
606        json!({"type":"assistant","parent_tool_use_id":parent,
607               "message":{"role":"assistant","content":[
608                   {"type":"tool_use","id":tool,"name":"Bash",
609                    "input":{"command":"sleep 180; echo waited","run_in_background":true}}]}})
610    }
611
612    #[test]
613    fn a_subagent_task_records_the_agent_call_that_owns_it() {
614        let store = store();
615        let mut feed = TaskFeed::default();
616        feed.apply(
617            &store,
618            None,
619            "blk",
620            &subagent_bash_call("toolu_1", "toolu_agent"),
621            1,
622        );
623        feed.apply(
624            &store,
625            None,
626            "blk",
627            &started("b1", "toolu_1", true, true),
628            2,
629        );
630        let row = store.background_task_get("toolu_1").unwrap().unwrap();
631        assert_eq!(row.owner_tool_use_id.as_deref(), Some("toolu_agent"));
632        assert!(
633            feed.owners.is_empty(),
634            "the owner is claimed by its task_started"
635        );
636    }
637
638    #[test]
639    fn the_agents_own_task_has_no_owner() {
640        let store = store();
641        let mut feed = TaskFeed::default();
642        let own = json!({"type":"assistant","parent_tool_use_id":null,
643                         "message":{"content":[{"type":"tool_use","id":"toolu_own","name":"Bash","input":{}}]}});
644        feed.apply(&store, None, "blk", &own, 1);
645        feed.apply(
646            &store,
647            None,
648            "blk",
649            &started("b2", "toolu_own", true, false),
650            2,
651        );
652        assert_eq!(
653            store
654                .background_task_get("toolu_own")
655                .unwrap()
656                .unwrap()
657                .owner_tool_use_id,
658            None
659        );
660    }
661
662    #[test]
663    fn the_owner_survives_a_later_move_to_the_background() {
664        let store = store();
665        let mut feed = TaskFeed::default();
666        feed.apply(
667            &store,
668            None,
669            "blk",
670            &subagent_bash_call("toolu_mv", "toolu_agent"),
671            1,
672        );
673        feed.apply(
674            &store,
675            None,
676            "blk",
677            &started("b3", "toolu_mv", false, true),
678            2,
679        );
680        feed.apply(
681            &store,
682            None,
683            "blk",
684            &updated("b3", json!({"is_backgrounded": true})),
685            3,
686        );
687        assert_eq!(
688            store
689                .background_task_get("toolu_mv")
690                .unwrap()
691                .unwrap()
692                .owner_tool_use_id
693                .as_deref(),
694            Some("toolu_agent")
695        );
696    }
697
698    /// The renderer's own push can create the row first, labelled with the
699    /// tool name; the CLI's description replaces it.
700    #[test]
701    fn the_clis_description_replaces_a_generic_label() {
702        let store = store();
703        store
704            .background_task_observe("toolu_1", "blk", "Bash", 1, 1)
705            .unwrap();
706        let mut feed = TaskFeed::default();
707        feed.apply(
708            &store,
709            None,
710            "blk",
711            &started("b1", "toolu_1", true, false),
712            2,
713        );
714        assert_eq!(
715            store.background_task_get("toolu_1").unwrap().unwrap().label,
716            "Background wait for agents"
717        );
718    }
719
720    /// A task that started before this reader existed (srv restart) still ends:
721    /// `task_notification` carries its own tool_use_id.
722    #[test]
723    fn a_notification_completes_a_row_this_reader_never_saw_start() {
724        let store = store();
725        store
726            .background_task_observe("toolu_old", "blk", "task dev", 5, 5)
727            .unwrap();
728        let mut feed = TaskFeed::default();
729        assert!(feed.apply(
730            &store,
731            None,
732            "blk",
733            &notification("gone", "toolu_old", "stopped"),
734            99
735        ));
736        assert_eq!(
737            store
738                .background_task_get("toolu_old")
739                .unwrap()
740                .unwrap()
741                .status,
742            BackgroundTaskStatus::Stopped
743        );
744    }
745
746    // --- `background_tasks_changed` snapshots -------------------------------
747    //
748    // The retro in docs/retro/retro-agent-pane-progress-flicker-and-orphaned-
749    // background-tasks-2026-09-30.md: the CLI dropped a subagent's two
750    // background tasks with no end event; their only trace was leaving the
751    // snapshot. Measured on real streams, a task that ends normally also leaves
752    // the snapshot one line BEFORE its end event, so leaving alone can't mean
753    // "stopped": it is settled at the next snapshot or turn boundary.
754
755    fn snapshot(task_ids: &[&str]) -> Value {
756        let tasks: Vec<Value> = task_ids
757            .iter()
758            .map(|t| json!({"task_id": t, "task_type": "local_bash", "description": "x"}))
759            .collect();
760        json!({"type":"system","subtype":"background_tasks_changed","tasks":tasks})
761    }
762    fn init() -> Value {
763        json!({"type":"system","subtype":"init","cwd":"/w","tools":[]})
764    }
765    fn result() -> Value {
766        json!({"type":"result","subtype":"success","is_error":false})
767    }
768    fn status_of(store: &Store, tool: &str) -> BackgroundTaskStatus {
769        store.background_task_get(tool).unwrap().unwrap().status
770    }
771
772    /// Opaz, 2026-09-29 09:44 UTC: in the snapshot, then gone from it, never
773    /// an end event, then the CLI starts a new session segment.
774    #[test]
775    fn a_task_dropped_from_the_snapshot_without_an_end_event_is_stopped_at_the_next_boundary() {
776        let store = store();
777        let mut feed = TaskFeed::default();
778        feed.apply(
779            &store,
780            None,
781            "blk",
782            &started("brsbiubq6", "toolu_orphan", true, true),
783            10,
784        );
785        feed.apply(
786            &store,
787            None,
788            "blk",
789            &snapshot(&["aead97a", "brsbiubq6"]),
790            20,
791        );
792        feed.apply(&store, None, "blk", &snapshot(&["aead97a"]), 30);
793        assert_eq!(
794            status_of(&store, "toolu_orphan"),
795            BackgroundTaskStatus::Running,
796            "not settled yet"
797        );
798
799        assert!(feed.apply(&store, None, "blk", &init(), 40));
800        let row = store.background_task_get("toolu_orphan").unwrap().unwrap();
801        assert_eq!(
802            (row.status, row.ended_at_ms),
803            (BackgroundTaskStatus::Stopped, Some(30))
804        );
805        assert!(feed.pending.is_empty());
806    }
807
808    /// The normal shape: the snapshot drops the task, and its end event
809    /// follows on the next line. The real status must win.
810    #[test]
811    fn a_normal_end_right_after_the_snapshot_keeps_its_real_status() {
812        let store = store();
813        let mut feed = TaskFeed::default();
814        feed.apply(
815            &store,
816            None,
817            "blk",
818            &started("bzs6opq7c", "toolu_ok", true, false),
819            10,
820        );
821        feed.apply(&store, None, "blk", &snapshot(&["bzs6opq7c"]), 20);
822        feed.apply(&store, None, "blk", &snapshot(&[]), 30);
823        feed.apply(
824            &store,
825            None,
826            "blk",
827            &updated("bzs6opq7c", json!({"status":"completed","end_time":31})),
828            31,
829        );
830        feed.apply(&store, None, "blk", &result(), 40);
831        let row = store.background_task_get("toolu_ok").unwrap().unwrap();
832        assert_eq!(
833            (row.status, row.ended_at_ms),
834            (BackgroundTaskStatus::Done, Some(31))
835        );
836    }
837
838    #[test]
839    fn a_later_snapshot_also_settles_a_vanished_task() {
840        let store = store();
841        let mut feed = TaskFeed::default();
842        feed.apply(
843            &store,
844            None,
845            "blk",
846            &started("t1", "toolu_a", true, false),
847            10,
848        );
849        feed.apply(&store, None, "blk", &snapshot(&["t1"]), 20);
850        feed.apply(&store, None, "blk", &snapshot(&[]), 30);
851        assert!(feed.apply(&store, None, "blk", &snapshot(&[]), 40));
852        assert_eq!(status_of(&store, "toolu_a"), BackgroundTaskStatus::Stopped);
853    }
854
855    #[test]
856    fn a_task_back_in_the_next_snapshot_is_not_settled() {
857        let store = store();
858        let mut feed = TaskFeed::default();
859        feed.apply(
860            &store,
861            None,
862            "blk",
863            &started("t1", "toolu_a", true, false),
864            10,
865        );
866        feed.apply(&store, None, "blk", &snapshot(&["t1"]), 20);
867        feed.apply(&store, None, "blk", &snapshot(&[]), 30);
868        feed.apply(&store, None, "blk", &snapshot(&["t1"]), 40);
869        feed.apply(&store, None, "blk", &result(), 50);
870        assert_eq!(status_of(&store, "toolu_a"), BackgroundTaskStatus::Running);
871    }
872
873    /// Most shell tasks never appear in a snapshot at all (measured: 124 of
874    /// 163). A snapshot that doesn't list a task it never listed says nothing
875    /// about it.
876    #[test]
877    fn a_task_never_listed_in_a_snapshot_is_left_alone() {
878        let store = store();
879        let mut feed = TaskFeed::default();
880        feed.apply(
881            &store,
882            None,
883            "blk",
884            &started("t1", "toolu_a", true, false),
885            10,
886        );
887        feed.apply(&store, None, "blk", &snapshot(&["other"]), 20);
888        feed.apply(&store, None, "blk", &init(), 30);
889        assert_eq!(status_of(&store, "toolu_a"), BackgroundTaskStatus::Running);
890    }
891
892    #[test]
893    fn a_vanished_foreground_task_is_forgotten_without_a_row() {
894        let store = store();
895        let mut feed = TaskFeed::default();
896        feed.apply(
897            &store,
898            None,
899            "blk",
900            &started("t1", "toolu_fg", false, false),
901            10,
902        );
903        feed.apply(&store, None, "blk", &snapshot(&["t1"]), 20);
904        feed.apply(&store, None, "blk", &snapshot(&[]), 30);
905        assert!(!feed.apply(&store, None, "blk", &result(), 40));
906        assert!(store.background_task_get("toolu_fg").unwrap().is_none());
907        assert!(feed.pending.is_empty());
908    }
909}
agentmux_srv/backend/background_task_feed.rs

agentmux_srv/backend/
background_task_feed.rs
