@@ -546,23 +546,55 @@ pub fn slack_dataflow<'a, P: 'a>(
             */),
         );
 
-    // Set an "is thinking" status as soon as a chat message arrives. Replies
-    // inside a subagent thread set the status on the root conversation thread
-    // (that is where the session-level status indicator lives).
-    let status_actions = chat_messages.map(q!(|event: SlackEvent| {
+    // Finalize the interrupted stream (if any) and set an "is thinking"
+    // status as soon as a chat message arrives. The daemon cancels an
+    // in-flight turn when user input lands, but it broadcasts no ToolResult
+    // events for the interrupted calls, and the cancelled turn's ResponseDone
+    // is swallowed by the had-tool-call keep-open rule — so without an
+    // explicit stop the old stream message would stay active (its task cards
+    // spinning forever) and the next turn would continue streaming *above*
+    // the user's message. Emitting the stop here (same branch, before the
+    // status, so in-branch ordering holds) finalizes the old message; the new
+    // turn's output can only be produced after the daemon receives SendInput,
+    // so it starts a fresh stream below. Replies inside a subagent's
+    // dedicated thread finalize that subagent's stream; the status indicator
+    // always lives on the root conversation thread.
+    let status_actions = chat_messages.flat_map_ordered(q!(|event: SlackEvent| {
         let rt = crate::runtime::get();
-        let thread_ts = {
-            let routes = rt.subagent_routes.lock().expect("bug: lock poisoned");
-            routes
-                .get(&event.thread_ts)
-                .map(|r| r.root_thread_ts.clone())
-                .unwrap_or(event.thread_ts)
+        let route = rt
+            .subagent_routes
+            .lock()
+            .expect("bug: lock poisoned")
+            .get(&event.thread_ts)
+            .cloned();
+        let (root_ts, subagent) = match route {
+            Some(r) => (r.root_thread_ts, Some(r.subagent_id)),
+            None => (event.thread_ts, None),
         };
-        SlackAction::SetStatus {
+        let mut actions = Vec::new();
+        let has_session = {
+            let sessions = rt.sessions.lock().expect("bug: lock poisoned");
+            sessions.get(&root_ts).is_some()
+        };
+        // Only finalize for known sessions (a brand-new thread has nothing to
+        // stop). Stale tool bookkeeping must be cleared even when no output
+        // was delivered yet, so the clear runs unconditionally for the key.
+        let was_in_flight =
+            crate::flow::clear_interrupted_turn_state(rt, &root_ts, subagent.as_deref());
+        if has_session && was_in_flight {
+            actions.push(SlackAction::StreamStop {
+                channel: event.channel.clone(),
+                thread_ts: root_ts.clone(),
+                subagent,
+                end_of_response: true,
+            });
+        }
+        actions.push(SlackAction::SetStatus {
             channel: event.channel,
-            thread_ts,
+            thread_ts: root_ts,
             status: "is thinking".to_owned(),
-        }
+        });
+        actions
     }));
 
     // --- Daemon events → Slack actions (streaming responses) ---
@@ -1052,6 +1084,56 @@ pub fn pair_tool_result(
         .and_then(|q| q.pop_front())
 }
 
+/// Clear the per-stream bookkeeping of a turn the user just interrupted by
+/// sending a message, returning whether anything suggested a response was in
+/// flight.
+///
+/// When user input arrives mid-turn, the daemon cancels the in-flight step
+/// and settles unanswered tool calls with *synthetic* results that are
+/// written to history only — no `ToolResult` events are broadcast — so the
+/// stream's FIFO queue entries for them would never be popped and would
+/// mislabel the next turn's completions. The orphan-result sentinels from a
+/// resume are stale for the same reason.
+///
+/// `had_tool_call` is deliberately **left alone**: the cancelled step still
+/// broadcasts exactly one `ResponseDone`, which consumes the flag (and is
+/// swallowed by the keep-open rule — harmless, since the stream was already
+/// finalized by the caller's `StreamStop`). Clearing it here would instead
+/// let that `ResponseDone` emit a second stop that wipes the fresh
+/// "is thinking" status.
+pub fn clear_interrupted_turn_state(
+    rt: &crate::runtime::Runtime,
+    thread_ts: &str,
+    subagent: Option<&str>,
+) -> bool {
+    let skey = crate::slack::stream_key(thread_ts, subagent);
+    let had_tasks = rt
+        .tool_tasks
+        .lock()
+        .expect("bug: lock poisoned")
+        .remove(&skey)
+        .map(|q| !q.is_empty())
+        .unwrap_or(false);
+    let orphans = rt
+        .resume_orphan_results
+        .lock()
+        .expect("bug: lock poisoned")
+        .remove(&skey)
+        .unwrap_or(0);
+    let delivering = rt
+        .stream_progress
+        .lock()
+        .expect("bug: lock poisoned")
+        .contains_key(&(thread_ts.to_owned(), subagent.map(str::to_owned)));
+    if had_tasks || orphans > 0 {
+        tracing::debug!(
+            "cleared interrupted-turn state for {skey} (open tasks: {had_tasks}, \
+             orphan sentinels: {orphans})"
+        );
+    }
+    had_tasks || orphans > 0 || delivering
+}
+
 pub fn subagent_of(
     msg: &infinity_protocol::DaemonMessage,
     root: Option<&infinity_protocol::ThreadRef>,
@@ -1305,6 +1387,68 @@ mod tests {
         assert_eq!(subagent_of(&welcome, Some(&root)), None);
     }
 
+    #[test]
+    fn clear_interrupted_turn_state_scopes_to_the_stream_key() {
+        crate::runtime::ensure_test_init();
+        let rt = crate::runtime::get();
+        // Unique ids: the runtime is shared process-wide across tests.
+        let thread_ts = "991.1";
+        let root_key = crate::slack::stream_key(thread_ts, None);
+        let sub_key = crate::slack::stream_key(thread_ts, Some("subA"));
+
+        // Idle thread: nothing to clear, nothing in flight.
+        assert!(!clear_interrupted_turn_state(rt, thread_ts, None));
+
+        // Root turn in flight (open tool task + delivery cursor + a stale
+        // orphan sentinel); the subagent's stream has its own open task.
+        rt.tool_tasks.lock().expect("lock").insert(
+            root_key.clone(),
+            [("t_root".to_owned(), "bash".to_owned(), "{}".to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        rt.tool_tasks.lock().expect("lock").insert(
+            sub_key.clone(),
+            [("t_sub".to_owned(), "grep".to_owned(), "{}".to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        rt.resume_orphan_results
+            .lock()
+            .expect("lock")
+            .insert(root_key.clone(), 2);
+        rt.stream_progress
+            .lock()
+            .expect("lock")
+            .insert((thread_ts.to_owned(), None), Default::default());
+
+        // Clearing the ROOT stream reports in-flight and leaves the subagent
+        // stream untouched.
+        assert!(clear_interrupted_turn_state(rt, thread_ts, None));
+        assert!(!rt.tool_tasks.lock().expect("lock").contains_key(&root_key));
+        assert!(!rt
+            .resume_orphan_results
+            .lock()
+            .expect("lock")
+            .contains_key(&root_key));
+        assert!(rt.tool_tasks.lock().expect("lock").contains_key(&sub_key));
+
+        // The delivery cursor alone (text-only turn) still reports in-flight;
+        // the cursor itself is owned by the I/O layer's end-of-response path
+        // and is not removed here.
+        assert!(clear_interrupted_turn_state(rt, thread_ts, None));
+
+        // The subagent stream clears independently.
+        assert!(clear_interrupted_turn_state(rt, thread_ts, Some("subA")));
+        assert!(!rt.tool_tasks.lock().expect("lock").contains_key(&sub_key));
+
+        // Clean up shared runtime state.
+        rt.stream_progress
+            .lock()
+            .expect("lock")
+            .remove(&(thread_ts.to_owned(), None));
+    }
+
     #[test]
     fn pair_tool_result_consumes_sentinels_before_the_queue() {
         crate::runtime::ensure_test_init();
 
@@ -182,8 +182,10 @@ fn choice_required(id: &str, prompt: &str) -> infinity_protocol::DaemonMessage {
     }
 }
 
-/// The non-`SetStatus` actions, which all originate from totally-ordered
-/// branches and therefore have a deterministic sequence.
+/// The non-`SetStatus` actions. In most tests these all originate from the
+/// totally-ordered daemon branch; tests that send a chat message while a turn
+/// is in flight additionally filter out `StreamStop` (the message-triggered
+/// finalize rides the slack-events branch and may interleave).
 fn non_status(actions: &[SlackAction]) -> Vec<&SlackAction> {
     actions
         .iter()
@@ -664,6 +666,162 @@ fn session_stop_interrupts_daemon_generation() {
     });
 }
 
+/// A user message that interrupts an in-flight turn must finalize the active
+/// stream (so the old message stops and its cards complete) BEFORE the new
+/// turn streams — and the interrupted turn's never-resolved tool entries must
+/// not desync the next turn's completion pairing.
+///
+/// Mirrors the live incident: the daemon settles interrupted tool calls with
+/// history-only synthetic results (no ToolResult broadcast) and its cancel
+/// ResponseDone is swallowed by the keep-open rule, so without the
+/// message-triggered finalize the old stream stayed open forever.
+#[test]
+fn user_message_mid_turn_finalizes_the_stream() {
+    let mut flow = FlowBuilder::new();
+    let (slack_tx, daemon_tx, actions_out, commands_out) = build_sim(&mut flow);
+
+    flow.sim().unit_test_fuzz_iterations(256).fuzz(async || {
+        let thread_ts = fresh("interrupt");
+        let channel = fresh("C");
+        let session_id = fresh("sess");
+
+        // Establish the session.
+        slack_tx.send(chat("U1", "go", &channel, &thread_ts));
+        assert!(matches!(
+            commands_out.next().await.unwrap(),
+            DaemonCommand::CreateSession { .. }
+        ));
+        daemon_tx.send(devent(
+            &thread_ts,
+            infinity_protocol::DaemonMessage::Connected {
+                root_thread_id: session_id.clone().into(),
+                thread_id: session_id.clone().into(),
+                model_name: String::new(),
+                context_window: 0,
+                title: Some("T".to_owned()),
+                total_tokens_used: 0,
+                provider_id: String::new(),
+            },
+        ));
+        loop {
+            match actions_out.next().await.unwrap() {
+                SlackAction::SetThreadTitle { .. } => break,
+                SlackAction::SetStatus { .. } => {}
+                other => panic!("expected SetThreadTitle, got {other:?}"),
+            }
+        }
+
+        // A turn is in flight: text + a tool call whose result will never be
+        // broadcast (the user's message interrupts it daemon-side).
+        daemon_tx.send_many([
+            devent(&thread_ts, text_chunk("part one")),
+            devent(&thread_ts, tool_call("execute_command")),
+        ]);
+
+        // Mid-turn user message → SendInput (and, in the same slack-branch
+        // tick, the finalize StreamStop).
+        slack_tx.send(chat(
+            "U1",
+            "actually, do this instead",
+            &channel,
+            &thread_ts,
+        ));
+        match commands_out.next().await.unwrap() {
+            DaemonCommand::SendInput { thread_ts: t, .. } => assert_eq!(t, thread_ts),
+            other => panic!("expected SendInput, got {other:?}"),
+        }
+
+        // The daemon's post-interrupt behavior: the cancelled step's
+        // ResponseDone (swallowed: the old turn had a tool call), then the new
+        // turn with its own tool round.
+        daemon_tx.send_many([
+            devent(&thread_ts, response_done()),   // cancel notification
+            devent(&thread_ts, tool_call("bash")), // new turn
+            devent(&thread_ts, response_done()),   // round end (swallowed)
+            devent(&thread_ts, tool_result()),     // bash completes
+            devent(&thread_ts, text_chunk("done")),
+            devent(&thread_ts, response_done()), // final → stop
+        ]);
+
+        let actions: Vec<SlackAction> = actions_out.collect().await;
+
+        // The daemon-branch sequence is totally ordered; the slack-branch
+        // finalize stop may interleave anywhere, so pull the stops out and
+        // assert the rest by order.
+        let stops: Vec<_> = actions
+            .iter()
+            .filter(|a| matches!(a, SlackAction::StreamStop { .. }))
+            .collect();
+        assert_eq!(
+            stops.len(),
+            2,
+            "expected the message-triggered finalize stop and the final \
+             response stop, got {stops:?}"
+        );
+        for stop in &stops {
+            match stop {
+                SlackAction::StreamStop {
+                    thread_ts: t,
+                    subagent,
+                    end_of_response,
+                    ..
+                } => {
+                    assert_eq!(t, &thread_ts);
+                    assert!(subagent.is_none());
+                    assert!(*end_of_response, "both stops end their response");
+                }
+                _ => unreachable!(),
+            }
+        }
+
+        // Everything else (appends and task updates) is daemon-branch ordered.
+        let core: Vec<_> = non_status(&actions)
+            .into_iter()
+            .filter(|a| !matches!(a, SlackAction::StreamStop { .. }))
+            .collect();
+        match &core[..] {
+            [SlackAction::StreamAppend { text: t1, .. }, SlackAction::StreamTaskUpdate {
+                task_id: old_task,
+                title: old_title,
+                status: s1,
+                ..
+            }, SlackAction::StreamTaskUpdate {
+                task_id: new_task,
+                title: new_title,
+                status: s2,
+                ..
+            }, SlackAction::StreamTaskUpdate {
+                task_id: done_task,
+                status: s3,
+                ..
+            }, SlackAction::StreamAppend { text: t2, .. }] => {
+                assert_eq!(t1, "part one");
+                assert_eq!(old_title, "execute_command");
+                assert_eq!(s1, "in_progress");
+                assert_eq!(new_title, "bash");
+                assert_eq!(s2, "in_progress");
+                assert_eq!(s3, "complete");
+                // The completion pairs with the NEW turn's call: the
+                // interrupted execute_command entry was cleared when the user
+                // message arrived, so it cannot swallow bash's result.
+                assert_eq!(
+                    done_task, new_task,
+                    "completion must pair with the new turn's call"
+                );
+                assert_ne!(
+                    done_task, old_task,
+                    "completion must not pair with the interrupted call"
+                );
+                assert_eq!(t2, "done");
+            }
+            other => panic!(
+                "expected [append, task(old in_progress), task(new in_progress), \
+                 task(new complete), append], got {other:?}"
+            ),
+        }
+    });
+}
+
 #[test]
 fn subagent_output_is_tagged_and_tracked_separately() {
     let mut flow = FlowBuilder::new();