Reconcile Claude background task state

server/src/session/claude.rs

@@ -214,6 +214,7 @@ impl ClaudeDriver {
         session_dir: &Path,
         sink: EventSink,
         subagents: Arc<Subagents>,
+        initial_status: SessionStatus,
     ) -> Result<Self> {
         let state = Arc::new(Mutex::new(Translator::new(
             session_dir.to_path_buf(),
@@ -261,6 +262,9 @@ impl ClaudeDriver {
                 Self::start(meta, provider, transport, session_dir)?
             }
         };
+        if !started_here && matches!(initial_status, SessionStatus::Idle | SessionStatus::Waiting) {
+            state.lock().unwrap().mark_settled();
+        }
 
         // A process this driver has just started has been asked for nothing,
         // which is what idle means. Said here because nothing else will: the
@@ -320,14 +324,22 @@ impl ClaudeDriver {
             format!("{} {}", provider.name, transport.describe()),
         ));
 
-        Ok(Self {
+        let driver = Self {
             sink,
             queue,
             to_child,
             state,
             session_dir: session_dir.to_path_buf(),
             reading,
-        })
+        };
+        if !started_here {
+            // Claude 2.1.261 sends a full background-task snapshot after a
+            // repeated initialize. That repairs a completion edge missed by a
+            // backend that was down while the CLI kept running; older CLIs
+            // accept the request and simply send no snapshot.
+            driver.send_control(json!({"subtype": "initialize"}), None);
+        }
+        Ok(driver)
     }
 
     /// Starts a new CLI for this session, with its streams in the session
@@ -866,8 +878,12 @@ fn translate_line(
         return true;
     };
     let opens_a_model_call = starts_a_model_call(&message);
+    let parent_running = queue.lock().unwrap().running;
     let (events, new_session_id, before) = {
         let mut state = state.lock().unwrap();
+        if parent_running {
+            state.mark_running();
+        }
         let before = state.session_id.clone();
         let events = state.translate(&message);
         let after = state.session_id.clone();

server/src/session/claude/translate.rs

@@ -63,6 +63,14 @@ struct PendingRequest {
     answers: HashMap<String, String>,
 }
 
+/// Where a terminal notification's detail still belongs after the level
+/// signal has already closed the task.
+#[derive(Clone, Copy, PartialEq)]
+enum TaskReport {
+    Subagent,
+    Command,
+}
+
 /// Translation state: stream-json lines in, common events out.
 pub(super) struct Translator {
     pub(super) session_id: Option<String>,
@@ -125,6 +133,16 @@ pub(super) struct Translator {
     /// translator existed, which is every one of them after a backend
     /// restart adopts a running session.
     open_tasks: HashSet<String>,
+    /// Claude's level signal for background work, when this CLI is new enough
+    /// to send one. Unlike `open_tasks`, this is a snapshot: each new value
+    /// replaces the old one, so a missed ending edge cannot leave work open
+    /// forever. See [`Translator::translate_background_tasks`].
+    background_tasks: Option<bool>,
+    /// Tasks the level signal closed before their ordinary notification
+    /// arrived. That notification still owns the useful summary, so it gets
+    /// one chance to update the transcript or tool card after the status was
+    /// already corrected.
+    awaiting_task_summaries: HashMap<String, TaskReport>,
     /// File-edit calls whose successful boilerplate result should not be drawn below their diff.
     /// Each leaves here with its `tool_result`; a failure keeps its text because that is the part a
     /// reader needs to act on.
@@ -139,6 +157,10 @@ pub(super) struct Translator {
     /// narrow: whether a task reporting back means the session has gone idle,
     /// or is merely one of several things happening inside a turn.
     in_turn: bool,
+    /// Whether this translator has observed that the parent is between turns.
+    /// False on construction because an adopted process may be mid-turn; the
+    /// driver sets it only when the persisted session status proves otherwise.
+    settled: bool,
 }
 
 impl Translator {
@@ -155,11 +177,28 @@ impl Translator {
             rate_limited: false,
             tasks: HashMap::new(),
             open_tasks: HashSet::new(),
+            background_tasks: None,
+            awaiting_task_summaries: HashMap::new(),
             patches: HashSet::new(),
             in_turn: false,
+            settled: false,
         }
     }
 
+    /// The persisted parent status says an adopted process is between turns,
+    /// so an initial background-task snapshot is safe to apply immediately.
+    pub(super) fn mark_settled(&mut self) {
+        self.settled = true;
+    }
+
+    /// The driver has a parent turn in flight. This covers the interval before
+    /// its first output proves the same thing, when a background-task update
+    /// from an older turn must not briefly return the session to idle.
+    pub(super) fn mark_running(&mut self) {
+        self.in_turn = true;
+        self.settled = false;
+    }
+
     /// Remembers what a control request was for, so its answer can say so.
     /// Called before the request goes out: the reader thread is already running
     /// and a fast CLI can answer before this side gets back to it.
@@ -263,8 +302,10 @@ impl Translator {
         let events = self.translate_line(message);
         if events.iter().any(closes_a_turn) {
             self.in_turn = false;
+            self.settled = true;
         } else if events.iter().any(super::proves_a_turn) {
             self.in_turn = true;
+            self.settled = false;
         }
         events
     }
@@ -388,6 +429,10 @@ impl Translator {
                 if tokens > 0 {
                     events.push(Event::UsageDelta { tokens, context });
                 }
+                // A level snapshot is authoritative at a turn boundary. In
+                // particular, it repairs a task whose terminal edge was
+                // missed before this backend adopted the still-running CLI.
+                events.extend(self.reconcile_background_tasks());
                 // Idle means "waiting for a person", and a session with a
                 // backgrounded subagent or command still running is not doing
                 // that -- it is waiting for itself, and will speak again with
@@ -450,6 +495,7 @@ impl Translator {
             Some("task_started" | "task_progress" | "task_updated" | "task_notification") => {
                 self.translate_task(message)
             }
+            Some("background_tasks_changed") => self.translate_background_tasks(message),
             Some("compact_boundary") => {
                 let meta = &message["compact_metadata"];
                 vec![Event::Compacted {
@@ -582,6 +628,7 @@ impl Translator {
                 // parent's own message, which arrives first and carries the
                 // title this side shows.
                 if let Some(about) = about {
+                    self.awaiting_task_summaries.remove(&about);
                     self.open_tasks.insert(about);
                 }
                 Vec::new()
@@ -592,19 +639,87 @@ impl Translator {
         }
     }
 
+    /// Claude 2.1.261's authoritative account of whether background work is
+    /// alive. The `tasks` array has replace semantics, but its ids are not
+    /// promised to correlate with task edges, so only its emptiness is used.
+    fn translate_background_tasks(&mut self, message: &Value) -> Vec<Event> {
+        let Some(tasks) = message.get("tasks").and_then(Value::as_array) else {
+            tracing::warn!("background_tasks_changed without a tasks array");
+            return Vec::new();
+        };
+        let was_outstanding = self.work_outstanding();
+        // The CLI explicitly says not to correlate these ids with its edge
+        // stream. Their useful claim here is the level: empty or not.
+        self.background_tasks = Some(!tasks.is_empty());
+
+        // During a turn the snapshot omits a foreground task, so wait for the
+        // result boundary before using it to close anything. Between turns,
+        // every task that can still be alive is background work and the level
+        // can repair a missed notification immediately.
+        if !self.settled || self.in_turn {
+            return Vec::new();
+        }
+        let mut events = self.reconcile_background_tasks();
+        let is_outstanding = self.work_outstanding();
+        if was_outstanding != is_outstanding {
+            events.push(Event::Status {
+                state: if is_outstanding {
+                    SessionStatus::Waiting
+                } else {
+                    SessionStatus::Idle
+                },
+            });
+        }
+        events
+    }
+
+    /// Applies the latest background-task snapshot at a point where no
+    /// foreground task can remain. Returns updates for background commands;
+    /// subagents carry the same correction in their own status transcript.
+    fn reconcile_background_tasks(&mut self) -> Vec<Event> {
+        let Some(false) = self.background_tasks else {
+            return Vec::new();
+        };
+        let mut events = Vec::new();
+        for info in self.subagents.list(true) {
+            if info.status == SessionStatus::Running {
+                self.awaiting_task_summaries
+                    .insert(info.id.clone(), TaskReport::Subagent);
+                self.subagents.finish(&info.id);
+            }
+        }
+        for id in self.open_tasks.drain() {
+            let report = if self.subagents.get(&id).is_some() {
+                TaskReport::Subagent
+            } else {
+                TaskReport::Command
+            };
+            self.awaiting_task_summaries.insert(id.clone(), report);
+            if report == TaskReport::Command {
+                events.push(Event::ToolUpdate {
+                    id,
+                    output: "this background command finished".to_string(),
+                });
+            }
+        }
+        events
+    }
+
     /// Whether the session has work of its own still running: the difference
     /// between `Idle` and [`SessionStatus::Waiting`].
     ///
-    /// Two sources because neither covers the other. `open_tasks` holds what
-    /// this translator watched start, which is the only thing that knows
-    /// about a backgrounded *command* -- it has no subagent. The registry
-    /// holds what is on disk, which is the only thing that knows about a
-    /// subagent that started before this translator did.
+    /// The level is authoritative when a current CLI has supplied it. The
+    /// edge fallback has two sources because neither covers the other:
+    /// `open_tasks` knows about a backgrounded command, which has no
+    /// subagent, while the registry knows about a subagent started before
+    /// this translator existed.
     ///
     /// `session_running` is true by construction: this is only ever asked
     /// while translating a line the session's process just wrote.
     fn work_outstanding(&self) -> bool {
-        !self.open_tasks.is_empty() || self.subagents.any_open(true)
+        self.background_tasks == Some(true)
+            || !self.open_tasks.is_empty()
+            || self.subagents.any_open(true)
     }
 
     /// A task reporting back, from whichever of the two lines got here first.
@@ -643,7 +758,9 @@ impl Translator {
         // `task_started` is behind the offset its session's stdout is read
         // from. `finish` below closes it either way, so a second line for the
         // same task still finds nothing.
-        if !self.open_tasks.remove(&about) && !self.subagents.is_open(&about) {
+        let reconciled = self.awaiting_task_summaries.remove(&about);
+        let was_open = self.open_tasks.remove(&about) || self.subagents.is_open(&about);
+        if !was_open && reconciled.is_none() {
             return Vec::new();
         }
         let mut events = Vec::new();
@@ -661,7 +778,7 @@ impl Translator {
                 self.subagents
                     .record(&about, Event::AssistantText { delta: summary });
             }
-        } else {
+        } else if reconciled != Some(TaskReport::Subagent) {
             events.push(Event::ToolUpdate {
                 id: about.clone(),
                 // A task that stopped without a word still has to say so: the
@@ -677,7 +794,7 @@ impl Translator {
         // over: it has stopped being `Waiting` and nothing else will say so.
         // Inside a turn there is nothing to announce -- the turn's own
         // `result` will decide between the two statuses when it lands.
-        if !self.work_outstanding() && !self.in_turn {
+        if was_open && !self.work_outstanding() && !self.in_turn {
             events.push(Event::Status {
                 state: SessionStatus::Idle,
             });
@@ -1724,6 +1841,126 @@ mod tests {
         );
     }
 
+    /// Claude's task edges are useful detail but not durable state. A
+    /// repeated initialize after adoption sends this level snapshot, whose
+    /// empty set is the authoritative answer even when the old registry says
+    /// a subagent is still running.
+    #[test]
+    fn a_background_snapshot_repairs_an_adopted_waiting_session() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let subagents = test_subagents(&dir);
+        subagents.start("toolu_stale", "an old helper", None);
+        let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
+        translator.mark_settled();
+
+        assert!(
+            translate_lines(
+                &mut translator,
+                &[r#"{"type":"system","subtype":"background_tasks_changed","tasks":["toolu_stale"]}"#],
+            )
+            .is_empty()
+        );
+        assert_eq!(subagents.list(true)[0].status, SessionStatus::Running);
+        assert_eq!(
+            translate_lines(
+                &mut translator,
+                &[r#"{"type":"system","subtype":"background_tasks_changed","tasks":[]}"#],
+            ),
+            vec![Event::Status {
+                state: SessionStatus::Idle
+            }]
+        );
+        assert_eq!(subagents.list(true)[0].status, SessionStatus::Exited);
+    }
+
+    /// An adopted process may be in the middle of a foreground agent when its
+    /// initialize snapshot arrives. Foreground work is absent from that
+    /// snapshot, so it is only safe to reconcile at the result boundary.
+    #[test]
+    fn a_background_snapshot_does_not_close_foreground_work_mid_turn() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let subagents = test_subagents(&dir);
+        subagents.start("toolu_foreground", "foreground helper", None);
+        let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
+        translator.mark_running();
+
+        assert!(
+            translate_lines(
+                &mut translator,
+                &[r#"{"type":"system","subtype":"background_tasks_changed","tasks":[]}"#],
+            )
+            .is_empty()
+        );
+        assert_eq!(subagents.list(true)[0].status, SessionStatus::Running);
+
+        let result = r#"{"type":"result","subtype":"success","is_error":false,"usage":{}}"#;
+        assert_eq!(
+            translate_lines(&mut translator, &[result]).last(),
+            Some(&Event::Status {
+                state: SessionStatus::Idle
+            })
+        );
+        assert_eq!(subagents.list(true)[0].status, SessionStatus::Exited);
+    }
+
+    /// The level edge is deliberately allowed to arrive before the detailed
+    /// notification. Correcting the status must not discard the useful report
+    /// that follows it or emit a second idle transition.
+    #[test]
+    fn a_notification_after_the_level_correction_keeps_its_summary() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let subagents = test_subagents(&dir);
+        subagents.start("toolu_level", "level helper", None);
+        let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
+        translator.mark_settled();
+        translate_lines(
+            &mut translator,
+            &[r#"{"type":"system","subtype":"background_tasks_changed","tasks":[]}"#],
+        );
+
+        assert!(
+            translate_lines(
+                &mut translator,
+                &[r#"{"type":"system","subtype":"task_notification","tool_use_id":"toolu_level","status":"completed","summary":"the useful report"}"#],
+            )
+            .is_empty()
+        );
+        let transcript_path = subagents
+            .get("toolu_level")
+            .expect("subagent")
+            .transcript_path();
+        assert!(
+            std::fs::read_to_string(transcript_path)
+                .expect("read transcript")
+                .contains("the useful report")
+        );
+    }
+
+    #[test]
+    fn a_deleted_reconciled_subagent_is_not_mistaken_for_a_command() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let subagents = test_subagents(&dir);
+        subagents.start("toolu_deleted", "deleted helper", None);
+        let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
+        translator.mark_settled();
+        translate_lines(
+            &mut translator,
+            &[r#"{"type":"system","subtype":"background_tasks_changed","tasks":[]}"#],
+        );
+        subagents
+            .delete(&["toolu_deleted".to_string()], true)
+            .expect("delete corrected subagent");
+
+        assert!(
+            translate_lines(
+                &mut translator,
+                &[r#"{"type":"system","subtype":"task_notification","tool_use_id":"toolu_deleted","status":"completed","summary":"late"}"#],
+            )
+            .is_empty(),
+            "a missing subagent is not a background command"
+        );
+    }
+
     /// A task ending *inside* a turn says nothing about the session's status:
     /// the turn is still running, and its own `result` decides. Without the
     /// `in_turn` guard this reported the session idle in the middle of one,

server/src/session/mod.rs

@@ -1908,6 +1908,7 @@ impl SessionManager {
                     session.transcript_path(),
                     &session.sink,
                     session.subagents(),
+                    status,
                 )?);
             }
             // Nothing is live for this one -- a session whose launch failed
@@ -2463,6 +2464,7 @@ fn launch(
                     &transcript_path,
                     &sink,
                     &subagents,
+                    status,
                 )
             })
             .transpose()?,
@@ -2514,6 +2516,7 @@ fn make_driver(
     transcript_path: &Path,
     sink: &EventSink,
     subagents: &Arc<Subagents>,
+    initial_status: SessionStatus,
 ) -> Result<Arc<dyn Driver>> {
     Ok(match provider.kind {
         DriverKind::Echo => Arc::new(EchoDriver::new(
@@ -2541,6 +2544,7 @@ fn make_driver(
             dir,
             sink.clone(),
             Arc::clone(subagents),
+            initial_status,
         )?),
         DriverKind::CodexCli => Arc::new(CodexDriver::launch(
             meta,