fix(session): preserve queue and terminal finality

src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs

@@ -213,6 +213,7 @@ pub async fn process_gateway_message(
         .ok()
         .map(|r| crate::lifecycle::TerminalTurnSignal {
             turn_id: r.turn_id.clone(),
+            turn_intent_id: None,
             status: crate::lifecycle::TurnTerminalStatus::Completed,
             completed_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
         });

src-tauri/crates/agent-core/src/core/session/scheduler.rs

@@ -544,7 +544,8 @@ impl WorkerTask {
                         let error_code = classify_streaming_error_message(err);
                         let streaming_error = StreamingError::new(err.clone(), error_code)
                             .with_details(serde_json::json!({
-                                "messageId": msg.message_id
+                                "messageId": msg.message_id,
+                                "turnIntentId": turn_intent_id,
                             }));
                         broadcast_agent_error_structured(&self.session_id, &streaming_error);
                     }

src-tauri/crates/agent-core/src/lifecycle.rs

@@ -84,6 +84,7 @@ impl TurnTerminalStatus {
 #[derive(Debug, Clone)]
 pub struct TerminalTurnSignal {
     pub turn_id: String,
+    pub turn_intent_id: Option<String>,
     pub status: TurnTerminalStatus,
     pub completed_at: String,
 }
@@ -190,6 +191,7 @@ fn persist_and_emit_terminal_turn(
         serde_json::json!({
             "sessionId": session_id,
             "turnId": terminal_turn.turn_id,
+            "turnIntentId": terminal_turn.turn_intent_id,
             "turnStatus": terminal_turn.status.as_str(),
             "sessionStatus": final_status.as_ref(),
             "completedAt": terminal_turn.completed_at,

src-tauri/crates/agent-core/src/state/commands/session/message/send.rs

@@ -38,6 +38,17 @@ pub(super) fn should_divert_to_mid_turn_steering(
         && is_turn_processing
 }
 
+async fn persist_direct_user_intervention(
+    params: Option<EnterMemberInterventionParams>,
+) -> Result<(), String> {
+    let Some(params) = params else {
+        return Ok(());
+    };
+    tokio::task::spawn_blocking(move || AgentMemberInterventionStore::enter(params).map(|_| ()))
+        .await
+        .map_err(|err| format!("Agent Org intervention worker failed: {err}"))?
+}
+
 /// Implementation of agent_send_message.
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn send_message_impl(
@@ -164,15 +175,13 @@ pub(crate) async fn send_message_impl(
         let _ = org_tasks::resume_paused_run_for_user_message(state, &session_id).await?;
     }
 
-    if mark_direct_user_intervention && !is_resume && !content.trim().is_empty() {
-        let runtime_snapshot = session_handle.runtime.read().await.clone();
-        if let Some(runtime) = runtime_snapshot {
-            if let Some(org_context) = runtime.agent_org_context.as_ref() {
-                let org_run_id = org_context.run_id.clone();
-                let org_context = org_context.clone();
-                let session_id_for_intervention = session_id.clone();
-                tokio::task::spawn_blocking(move || {
-                    let member_id =
+    let direct_user_intervention =
+        if mark_direct_user_intervention && !is_resume && !content.trim().is_empty() {
+            let runtime_snapshot = session_handle.runtime.read().await.clone();
+            match runtime_snapshot.and_then(|runtime| runtime.agent_org_context.clone()) {
+                Some(org_context) => {
+                    let session_id_for_intervention = session_id.clone();
+                    let member_id = tokio::task::spawn_blocking(move || {
                         crate::session::persistence::get_session(&session_id_for_intervention)
                             .map_err(|err| err.to_string())?
                             .and_then(|record| record.org_member_id)
@@ -181,31 +190,34 @@ pub(crate) async fn send_message_impl(
                                     "Agent Org session {} has no canonical member_id",
                                     session_id_for_intervention
                                 )
-                            })?;
-                    if !can_enter_member_intervention(&member_id) {
+                            })
+                    })
+                    .await
+                    .map_err(|err| format!("Agent Org member lookup worker failed: {err}"))??;
+                    if can_enter_member_intervention(&member_id) {
+                        let agent_id = org_context.require_participant_agent_id(&member_id)?;
+                        Some(EnterMemberInterventionParams {
+                            org_run_id: org_context.run_id,
+                            member_id,
+                            agent_id,
+                            session_id: session_id.clone(),
+                            reason: Some("direct_user_chat".to_string()),
+                            ttl_secs: DEFAULT_INTERVENTION_TTL_SECS,
+                        })
+                    } else {
                         tracing::debug!(
-                            org_run_id = %org_run_id,
-                            session_id = %session_id_for_intervention,
+                            org_run_id = %org_context.run_id,
+                            session_id = %session_id,
                             "ordinary coordinator message does not enter member intervention"
                         );
-                        return Ok::<(), String>(());
+                        None
                     }
-                    let agent_id = org_context.require_participant_agent_id(&member_id)?;
-                    AgentMemberInterventionStore::enter(EnterMemberInterventionParams {
-                        org_run_id,
-                        member_id,
-                        agent_id,
-                        session_id: session_id_for_intervention,
-                        reason: Some("direct_user_chat".to_string()),
-                        ttl_secs: DEFAULT_INTERVENTION_TTL_SECS,
-                    })?;
-                    Ok::<(), String>(())
-                })
-                .await
-                .map_err(|err| err.to_string())??;
+                }
+                None => None,
             }
-        }
-    }
+        } else {
+            None
+        };
 
     let app_handle = state.app_handle.clone();
 
@@ -229,6 +241,10 @@ pub(crate) async fn send_message_impl(
         images.as_deref(),
         session_handle.scheduler.is_turn_processing(),
     ) {
+        // Steering mutates an already-running member turn, so intervention is
+        // part of accepting the control action. If the durable takeover row
+        // cannot be written, do not inject a message that Wake may race.
+        persist_direct_user_intervention(direct_user_intervention.clone()).await?;
         crate::foundation::session_bridge::upsert_turn_intent(
             &session_id,
             &effective_turn_intent_id,
@@ -330,6 +346,7 @@ pub(crate) async fn send_message_impl(
     let display_text_for_closure = display_text;
     let workspace_root_for_closure = effective_workspace_root.clone();
     let turn_intent_id_for_closure = effective_turn_intent_id.clone();
+    let direct_user_intervention_for_closure = direct_user_intervention;
     // Resolve durable mode-control rows from exactly the bounded inbox batch
     // this background wake will drain. A control row in a later batch must
     // not change the mode of earlier work; rows become one-shot only when the
@@ -381,9 +398,14 @@ pub(crate) async fn send_message_impl(
         let workspace_root = workspace_root_for_closure;
         let session = session_for_closure;
         let turn_intent_id = turn_intent_id_for_closure;
+        let direct_user_intervention = direct_user_intervention_for_closure;
         let org_wake_run_id = org_wake_run_id;
 
         Box::pin(async move {
+            // The scheduler now owns this accepted turn. Intervention is a
+            // turn-start side effect, not submit preflight: queued work that is
+            // invalidated before execution must never leave a takeover row.
+            persist_direct_user_intervention(direct_user_intervention).await?;
             // Queued and coalesced messages are not running sessions. Promote
             // the DB state only when the scheduler actually begins execution.
             // For Agent Org wakes, re-check the run and update the session in
@@ -452,7 +474,7 @@ pub(crate) async fn send_message_impl(
                 channel: None,
                 chat_id: None,
                 turn_id: Some(turn_id.clone()),
-                turn_intent_id,
+                turn_intent_id: turn_intent_id.clone(),
             };
 
             let response =
@@ -487,6 +509,7 @@ pub(crate) async fn send_message_impl(
                     .ok()
                     .map(|r| crate::lifecycle::TerminalTurnSignal {
                         turn_id: r.turn_id.clone(),
+                        turn_intent_id: Some(turn_intent_id.clone()),
                         status: match final_turn_state {
                             crate::session::DialogTurnState::Cancelled => {
                                 crate::lifecycle::TurnTerminalStatus::Cancelled

src-tauri/crates/agent-core/src/state/commands/session/mod.rs

@@ -182,6 +182,7 @@ pub async fn agent_send_message(
     #[allow(non_snake_case)] clientMessageId: Option<String>,
     #[allow(non_snake_case)] turnIntentId: Option<String>,
     #[allow(non_snake_case)] turnIntentSource: String,
+    #[allow(non_snake_case)] markDirectUserIntervention: Option<bool>,
 ) -> Result<AgentResponse, String> {
     let source = crate::foundation::session_bridge::TurnIntentBridgeSource::parse(
         &turnIntentSource,
@@ -206,11 +207,9 @@ pub async fn agent_send_message(
         images,
         ide_context,
         isResume.unwrap_or(false),
-        // Direct-user intervention is signalled explicitly at the UI submit
-        // or queue-dispatch boundary. Do not infer it from every generic
-        // agent_send_message call: programmatic continuations use this command
-        // too and must not take over an Agent Org worker accidentally.
-        false,
+        // Only real user-authored submit/queue paths set this. Programmatic
+        // continuations, wake turns, and Resume leave it false.
+        markDirectUserIntervention.unwrap_or(false),
         clientMessageId,
         turnIntentId,
         None,

src-tauri/src/api/agent/test/agent_org.rs

@@ -445,6 +445,7 @@ pub async fn test_agent_org_launch_coordinator(
                             true,
                             Some(agent_core::lifecycle::TerminalTurnSignal {
                                 turn_id: response.turn_id.clone(),
+                                turn_intent_id: None,
                                 status: agent_core::lifecycle::TurnTerminalStatus::Completed,
                                 completed_at: chrono::Utc::now()
                                     .to_rfc3339_opts(chrono::SecondsFormat::Millis, true),

src/app/root/e2e/helpers/sessionHelpers/inspectChatState.ts

@@ -15,9 +15,9 @@ import { chatEventsAtom } from "@src/engines/SessionCore/derived/chatEvents";
 import {
   isPendingCancelAtom,
   isSessionActiveAtom,
+  postStopDispatchSessionsAtom,
   sessionRuntimeErrorAtom,
   sessionRuntimeStatusAtom,
-  userInitiatedCancelAtom,
 } from "@src/store/session/cliSessionStatusAtom";
 import {
   fileReviewMapAtom,
@@ -216,7 +216,9 @@ export function createInspectChatStateHelper(store: E2EStore) {
         isSessionActive: store.get(isSessionActiveAtom),
         isPendingCancel: store.get(isPendingCancelAtom),
         isQueueEditing: store.get(queueEditingAtom),
-        userInitiatedCancel: store.get(userInitiatedCancelAtom),
+        userInitiatedCancel: activeSessionId
+          ? store.get(postStopDispatchSessionsAtom)[activeSessionId] === true
+          : false,
         turnPhase: activeSessionId ? getTurnPhase(activeSessionId) : "idle",
         turnGeneration: activeSessionId
           ? getTurnGeneration(activeSessionId)

src/app/root/e2e/helpers/sessions.ts

@@ -32,13 +32,13 @@ import {
   type ContextUsageSnapshot,
   isPendingCancelAtom,
   lastUserMessageAtom,
+  postStopDispatchSessionsAtom,
   restoreToInputAtom,
   sessionContextTokensAtom,
   sessionContextUsageAtom,
   sessionRolledBackAtom,
   sessionRuntimeStatusAtom,
   streamRetryStatusAtom,
-  userInitiatedCancelAtom,
 } from "@src/store/session/cliSessionStatusAtom";
 import {
   pendingPlanApprovalsAtom,
@@ -295,7 +295,7 @@ export function createSessionHelpers(store: E2EStore) {
       resetTurnLifecycleForTests();
       store.set(chatImageAttachmentsAtom, []);
       store.set(isPendingCancelAtom, false);
-      store.set(userInitiatedCancelAtom, false);
+      store.set(postStopDispatchSessionsAtom, {});
       store.set(sessionRuntimeStatusAtom, "idle");
       store.set(sessionContextTokensAtom, 0);
       store.set(sessionContextUsageAtom, null);

src/engines/ChatPanel/hooks/useWorkspaceChat/useMessageDispatch.ts

@@ -38,27 +38,17 @@ import { resolveModelForMessage } from "@src/util/session/resolveModelForMessage
 import { selectionFromSession } from "@src/util/session/selectionFromSession";
 import { isCursorIdeSession } from "@src/util/session/sessionDispatch";
 
-interface UseMessageDispatchOptions {
-  getSessionId: () => string | null;
-}
-
-export function useMessageDispatch(options: UseMessageDispatchOptions) {
-  const { getSessionId } = options;
+export function useMessageDispatch() {
   const setSessionRuntimeStatus = useSetAtom(setSessionRuntimeStatusAtom);
   const setLastUserMessage = useSetAtom(lastUserMessageAtom);
 
   const addUserMessage = useCallback(
     async (
+      sessionId: string,
       content: string,
       imageDataUrls?: string[],
       turnIntentId?: string
-    ): Promise<void> => {
-      const sessionId = getSessionId();
-      if (!sessionId) {
-        throw new Error(
-          "[useMessageDispatch] addUserMessage: no active sessionId"
-        );
-      }
+    ): Promise<string> => {
       const userEvent = createSyntheticUserEvent(sessionId, content, {
         imageDataUrls,
         turnIntentId,
@@ -73,8 +63,9 @@ export function useMessageDispatch(options: UseMessageDispatchOptions) {
         displayContent: content,
         imageDataUrls,
       });
+      return userEvent.id;
     },
-    [getSessionId, setLastUserMessage]
+    [setLastUserMessage]
   );
 
   const dispatchMessageBySessionType = useCallback(
@@ -130,6 +121,7 @@ export function useMessageDispatch(options: UseMessageDispatchOptions) {
           clientMessageId,
           turnIntentId,
           turnIntentSource: "user_submit",
+          directUserIntent: true,
         });
         // Backend accepted the message — the turn is running even if the
         // provider's running ack has not been observed yet.