From ef1bc95d4a123c91b73193aa6964e892b5316821 Mon Sep 17 00:00:00 2001 From: Tryanks Date: Sat, 25 Jul 2026 17:29:24 +0800 Subject: [PATCH] fix(runtime): stop a dropped queue head from pinning a session at "Working" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removing a queued message with the strip's ✕ used a bare `retain`, which also removed the head that `dispatch_next_pending` had already submitted to the provider. `accept_turn_delivery` correlates the acknowledgement against `queue.first()`, so with the head gone the matching `TurnAccepted` was rejected and `delivery_in_flight` stayed `Some(_)` forever — while the provider, which had received the request, went on to complete the turn. The session then disagreed with itself: the sidebar read the JSONL fold for the active thread (turn over, no badge) and the lifecycle flags for a parked one (delivery in flight, "Working"). Parking preserves the flags, so opening and leaving the thread toggled the badge instead of clearing it. `drop_queued` now goes through `take_queued`, which already refuses to remove a submitted head, so the later acknowledgement clears the flag normally. `turn_running_for` folds onto a single `ActiveSession::has_work` predicate shared with the quit guard, and the sidebar makes one call instead of branching on selection — the two answers can no longer drift. Co-Authored-By: Claude Opus 5 --- crates/runtime/src/app.rs | 201 +++++++++++++++++++++++++++++--------- crates/ui/src/sidebar.rs | 21 +--- 2 files changed, 162 insertions(+), 60 deletions(-) diff --git a/crates/runtime/src/app.rs b/crates/runtime/src/app.rs index 964e9ae68..ef9ecc8f7 100644 --- a/crates/runtime/src/app.rs +++ b/crates/runtime/src/app.rs @@ -681,6 +681,15 @@ impl ActiveSession { self.turn_in_flight } + /// Whether this session owns work which must stay live and surface as + /// "Working", regardless of whether it is active or parked. + fn has_work(&self) -> bool { + self.turn_in_flight + || self.delivery_in_flight.is_some() + || !self.queue.is_empty() + || self.background_task_count > 0 + } + pub fn queued(&self) -> &[QueuedMessage] { &self.queue } @@ -4586,21 +4595,13 @@ impl AppState { cx.notify(); } - /// Whether a turn is currently running for `session_id` (only the active - /// session can be running). + /// Whether `session_id` owns live or queued work. pub fn turn_running_for(&self, session_id: &str) -> bool { - if let Some(active) = self.active.as_ref().filter(|a| a.meta.id == session_id) { - return active.timeline.turn_running; - } - // A parked session is working when a turn is in flight or its queue - // still has messages to run (the parked timeline is stale by design, so - // the flags are the source of truth). - self.background.get(session_id).is_some_and(|s| { - s.turn_in_flight - || s.delivery_in_flight.is_some() - || !s.queue.is_empty() - || s.background_task_count > 0 - }) + self.active + .as_ref() + .filter(|session| session.meta.id == session_id) + .or_else(|| self.background.get(session_id)) + .is_some_and(ActiveSession::has_work) } /// The first approval currently blocking a session, including parked and @@ -4633,33 +4634,12 @@ impl AppState { /// background tasks. Quitting stops all of it, so the quit guard must gate /// on this rather than on turns alone. pub fn working_sessions_count(&self) -> usize { - fn works( - turn_in_flight: bool, - delivery: bool, - queued: bool, - background_tasks: usize, - ) -> bool { - turn_in_flight || delivery || queued || background_tasks > 0 - } - usize::from(self.active.as_ref().is_some_and(|s| { - works( - s.turn_in_flight, - s.delivery_in_flight.is_some(), - !s.queue.is_empty(), - s.background_task_count, - ) - })) + self - .background - .values() - .filter(|s| { - works( - s.turn_in_flight, - s.delivery_in_flight.is_some(), - !s.queue.is_empty(), - s.background_task_count, - ) - }) - .count() + usize::from(self.active.as_ref().is_some_and(ActiveSession::has_work)) + + self + .background + .values() + .filter(|session| session.has_work()) + .count() } /// Record that a thread has been visited now (clears its unread dot). @@ -5887,10 +5867,11 @@ impl AppState { } /// Queue strip: drop a queued message (the row's ✕). It was never recorded, - /// so nothing needs undoing. + /// so nothing needs undoing. A submitted head cannot be removed until its + /// correlated provider acknowledgement commits it. pub fn drop_queued(&mut self, id: u64, cx: &mut Context) { if let Some(active) = self.active.as_mut() { - active.queue.retain(|m| m.id != id); + let _ = active.take_queued(id); } cx.notify(); } @@ -12127,6 +12108,135 @@ mod tests { let _ = std::fs::remove_dir_all(&data); } + #[gpui::test] + fn submitted_queue_head_cannot_leak_delivery_after_turn_completion( + cx: &mut gpui::TestAppContext, + ) { + let cwd = + std::env::temp_dir().join(format!("tcode-submitted-drop-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&cwd).unwrap(); + let data = std::env::temp_dir().join(format!( + "tcode-submitted-drop-data-{}", + uuid::Uuid::new_v4() + )); + let store = SessionStore::open_at(data.clone()).unwrap(); + let state = cx.new(|_| AppState::new(store)); + let (session, commands) = fake_live_session(cwd.clone()); + let id = session.meta.id.clone(); + + state.update(cx, |state, cx| { + state.store.upsert_meta(&session.meta).unwrap(); + state.sessions = state.store.load_index(); + state.active = Some(session); + state.send_turn("finish this task".into(), Vec::new(), cx); + let delivery_id = match commands.try_recv() { + Ok(SessionCommand::SendTurn { delivery_id, .. }) => delivery_id, + other => panic!("expected SendTurn, got {other:?}"), + }; + + // The submitted head remains in the visible queue strip until its + // provider acknowledgement. Its ✕ must not invalidate correlation. + state.drop_queued(delivery_id, cx); + state.on_event(&id, AgentEvent::TurnAccepted { delivery_id }, cx); + state.on_event( + &id, + AgentEvent::TurnStarted { + turn_id: "turn-1".into(), + }, + cx, + ); + state.on_event( + &id, + AgentEvent::TurnCompleted { + turn_id: "turn-1".into(), + status: TurnStatus::Completed, + usage: None, + }, + cx, + ); + assert!(!state.turn_running_for(&id)); + + state.start_draft("proj".into(), cwd.clone(), cx); + state.select_session(&id, cx); + assert!(!state.turn_running_for(&id)); + state.start_draft("proj".into(), cwd.clone(), cx); + assert!( + !state.turn_running_for(&id), + "the completed delivery became Working again after reparking" + ); + }); + + let _ = std::fs::remove_dir_all(&cwd); + let _ = std::fs::remove_dir_all(&data); + } + + #[gpui::test] + fn turn_running_for_is_independent_of_active_or_parked_location(cx: &mut gpui::TestAppContext) { + let data = std::env::temp_dir().join(format!( + "tcode-working-location-test-{}", + uuid::Uuid::new_v4() + )); + let store = SessionStore::open_at(data.clone()).unwrap(); + let state = cx.new(|_| AppState::new(store)); + let commands = async_channel::unbounded().0; + + let mut idle = live_session(ProviderKind::ClaudeCode, commands.clone()); + idle.meta.id = "idle".into(); + + let mut turn = live_session(ProviderKind::ClaudeCode, commands.clone()); + turn.meta.id = "turn".into(); + turn.turn_in_flight = true; + + let mut delivery = live_session(ProviderKind::ClaudeCode, commands.clone()); + delivery.meta.id = "delivery".into(); + delivery.delivery_in_flight = Some(7); + + let mut queued = live_session(ProviderKind::ClaudeCode, commands.clone()); + queued.meta.id = "queued".into(); + queued.push_queued("waiting".into(), Vec::new()); + + let mut background = live_session(ProviderKind::ClaudeCode, commands.clone()); + background.meta.id = "background".into(); + background.background_task_count = 1; + + let mut stale_timeline = live_session(ProviderKind::ClaudeCode, commands); + stale_timeline.meta.id = "stale-timeline".into(); + stale_timeline.timeline.apply_at( + None, + &AgentEvent::TurnStarted { + turn_id: "stale".into(), + }, + ); + + state.update(cx, |state, _| { + for (label, session, expected) in [ + ("idle", idle, false), + ("turn", turn, true), + ("delivery", delivery, true), + ("queued", queued, true), + ("background", background, true), + ("stale timeline", stale_timeline, false), + ] { + let id = session.meta.id.clone(); + state.active = Some(session); + let active_answer = state.turn_running_for(&id); + + let parked = state.active.take().unwrap(); + state.background.insert(id.clone(), parked); + let parked_answer = state.turn_running_for(&id); + + assert_eq!( + active_answer, parked_answer, + "{label} changed answer when moved between active and parked" + ); + assert_eq!(active_answer, expected, "{label} work predicate"); + state.background.remove(&id); + } + }); + + let _ = std::fs::remove_dir_all(&data); + } + /// The T3 Code session-reaper failure class, our variant: switching to /// another thread must NOT kill a session whose turn is still running. The /// session parks in the background — process and queue alive, events still @@ -12258,7 +12368,10 @@ mod tests { }, cx, ); - assert!(!state.turn_running_for(&id_a) || state.active.is_some()); + assert!( + !state.turn_running_for(&id_a), + "a completed active session must not remain Working" + ); }); let _ = std::fs::remove_dir_all(&cwd); diff --git a/crates/ui/src/sidebar.rs b/crates/ui/src/sidebar.rs index 585fb63b9..f6b4c78a0 100644 --- a/crates/ui/src/sidebar.rs +++ b/crates/ui/src/sidebar.rs @@ -805,7 +805,6 @@ impl SessionsSidebar { &self, group: &ProjectGroup, active_id: Option<&str>, - turn_running: bool, cx: &mut Context, ) -> impl IntoElement + use<> { let project_id = group.project.id.clone(); @@ -955,8 +954,7 @@ impl SessionsSidebar { let is_active = active_id == Some(meta.id.as_str()); // "Working" covers parked sessions too — a thread that keeps // running in the background keeps its green dot. - let working = (is_active && turn_running) - || (!is_active && self.app_state.read(cx).turn_running_for(&meta.id)); + let working = self.app_state.read(cx).turn_running_for(&meta.id); container = container.child(self.render_thread(meta, is_active, working, cx)); } if let Some(toggle_label) = thread_list_toggle_label(total, expanded) { @@ -1399,15 +1397,10 @@ fn proceed_delete( impl Render for SessionsSidebar { fn render(&mut self, window: &mut Window, cx: &mut Context) -> impl IntoElement { - let (groups, active_id, turn_running) = { + let (groups, active_id) = { let state = self.app_state.read(cx); let active_id = state.active_session_id().map(str::to_string); - let turn_running = state - .active - .as_ref() - .map(|a| a.timeline.turn_running) - .unwrap_or(false); - (state.grouped_sessions(), active_id, turn_running) + (state.grouped_sessions(), active_id) }; let mut list_content = v_flex().w_full().px_2().pb_2().gap(px(2.)); @@ -1423,12 +1416,8 @@ impl Render for SessionsSidebar { ); } else { for group in &groups { - list_content = list_content.child(self.render_group( - group, - active_id.as_deref(), - turn_running, - cx, - )); + list_content = + list_content.child(self.render_group(group, active_id.as_deref(), cx)); } }