diff --git a/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs b/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs index 6f5e9da3e..ba769e8f5 100644 --- a/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs +++ b/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), }); diff --git a/src-tauri/crates/agent-core/src/core/session/scheduler.rs b/src-tauri/crates/agent-core/src/core/session/scheduler.rs index 628ec470b..29e6fed79 100644 --- a/src-tauri/crates/agent-core/src/core/session/scheduler.rs +++ b/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); } diff --git a/src-tauri/crates/agent-core/src/lifecycle.rs b/src-tauri/crates/agent-core/src/lifecycle.rs index fd80db128..1893c7da6 100644 --- a/src-tauri/crates/agent-core/src/lifecycle.rs +++ b/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, 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, diff --git a/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs b/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs index 093f010cf..70cf84805 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs +++ b/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, +) -> 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 diff --git a/src-tauri/crates/agent-core/src/state/commands/session/mod.rs b/src-tauri/crates/agent-core/src/state/commands/session/mod.rs index 34cddffba..dee3d8cf7 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/mod.rs +++ b/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, #[allow(non_snake_case)] turnIntentId: Option, #[allow(non_snake_case)] turnIntentSource: String, + #[allow(non_snake_case)] markDirectUserIntervention: Option, ) -> Result { 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, diff --git a/src-tauri/src/api/agent/test/agent_org.rs b/src-tauri/src/api/agent/test/agent_org.rs index 14de10dc5..d88e638c5 100644 --- a/src-tauri/src/api/agent/test/agent_org.rs +++ b/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), diff --git a/src/app/root/e2e/helpers/sessionHelpers/inspectChatState.ts b/src/app/root/e2e/helpers/sessionHelpers/inspectChatState.ts index 62726bf60..76665f28e 100644 --- a/src/app/root/e2e/helpers/sessionHelpers/inspectChatState.ts +++ b/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) diff --git a/src/app/root/e2e/helpers/sessions.ts b/src/app/root/e2e/helpers/sessions.ts index 1a2215547..e71512585 100644 --- a/src/app/root/e2e/helpers/sessions.ts +++ b/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); diff --git a/src/engines/ChatPanel/hooks/useWorkspaceChat/useMessageDispatch.ts b/src/engines/ChatPanel/hooks/useWorkspaceChat/useMessageDispatch.ts index c44ffc9af..fa0f16b69 100644 --- a/src/engines/ChatPanel/hooks/useWorkspaceChat/useMessageDispatch.ts +++ b/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 => { - const sessionId = getSessionId(); - if (!sessionId) { - throw new Error( - "[useMessageDispatch] addUserMessage: no active sessionId" - ); - } + ): Promise => { 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. diff --git a/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.intervention.test.ts b/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.intervention.test.ts index 8c39a8b01..50c22660d 100644 --- a/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.intervention.test.ts +++ b/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.intervention.test.ts @@ -3,6 +3,7 @@ import { createElement } from "react"; import { renderToString } from "react-dom/server"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import { postStopDispatchSessionsAtom } from "@src/store/session/cliSessionStatusAtom"; import { messageQueueAtom } from "@src/store/ui/messageQueueAtom"; import { @@ -17,15 +18,11 @@ const mocks = vi.hoisted(() => ({ beginOptimisticTurn: vi.fn(), beginTurnDispatch: vi.fn(), dispatchMessageBySessionType: vi.fn(), - enterIntervention: vi.fn(), failOptimisticTurn: vi.fn(), getTurnPhase: vi.fn(), markTurnTerminal: vi.fn(), mintTurnIntentId: vi.fn(), -})); - -vi.mock("@src/api/tauri/agent", () => ({ - enterAgentOrgSessionIntervention: mocks.enterIntervention, + removeByIdPrefix: vi.fn(), })); vi.mock("@src/engines/SessionCore/control/optimisticTurnStatus", () => ({ @@ -39,6 +36,10 @@ vi.mock("@src/engines/SessionCore/control/turnLifecycle", () => ({ markTurnTerminal: mocks.markTurnTerminal, })); +vi.mock("@src/engines/SessionCore/core/store/EventStoreProxy", () => ({ + eventStoreProxy: { removeByIdPrefix: mocks.removeByIdPrefix }, +})); + vi.mock("@src/engines/SessionCore/sync/adapters/shared/eventFactories", () => ({ mintTurnIntentId: mocks.mintTurnIntentId, })); @@ -77,34 +78,51 @@ function renderSubmitHook(store: ReturnType) { describe("useUserIntentSubmit Agent Org intervention", () => { beforeEach(() => { - mocks.addUserMessage.mockReset().mockResolvedValue(undefined); + mocks.addUserMessage.mockReset().mockResolvedValue("synthetic-user-1"); mocks.beginOptimisticTurn.mockReset(); mocks.beginTurnDispatch.mockReset().mockReturnValue(7); mocks.dispatchMessageBySessionType.mockReset().mockResolvedValue(undefined); - mocks.enterIntervention.mockReset().mockResolvedValue(true); mocks.failOptimisticTurn.mockReset(); mocks.getTurnPhase.mockReset().mockReturnValue("idle"); mocks.markTurnTerminal.mockReset(); mocks.mintTurnIntentId.mockReset().mockReturnValue("turn-intent-1"); + mocks.removeByIdPrefix.mockReset().mockResolvedValue(1); }); - it("marks intervention after persisting the direct user event and before dispatch", async () => { + it("appends the direct user event before dispatching the same intent", async () => { const submit = renderSubmitHook(createStore()); await submit({ sessionId: SESSION_ID, displayContent: "hello worker" }); - expect(mocks.addUserMessage).toHaveBeenCalledOnce(); - expect(mocks.enterIntervention).toHaveBeenCalledOnce(); - expect(mocks.enterIntervention).toHaveBeenCalledWith(SESSION_ID); + expect(mocks.addUserMessage).toHaveBeenCalledWith( + SESSION_ID, + "hello worker", + undefined, + "turn-intent-1" + ); expect(mocks.dispatchMessageBySessionType).toHaveBeenCalledOnce(); expect(mocks.addUserMessage.mock.invocationCallOrder[0]).toBeLessThan( - mocks.enterIntervention.mock.invocationCallOrder[0] - ); - expect(mocks.enterIntervention.mock.invocationCallOrder[0]).toBeLessThan( mocks.dispatchMessageBySessionType.mock.invocationCallOrder[0] ); }); + it("keeps a Stop episode scoped to its own session", async () => { + const store = createStore(); + store.set(postStopDispatchSessionsAtom, { "session-a": true }); + const submit = renderSubmitHook(store); + + await submit({ + sessionId: SESSION_ID, + displayContent: "session b message", + }); + + expect(store.get(messageQueueAtom)).toEqual([]); + expect(mocks.dispatchMessageBySessionType).toHaveBeenCalledOnce(); + expect(store.get(postStopDispatchSessionsAtom)).toEqual({ + "session-a": true, + }); + }); + it("only enqueues while the current turn is busy", async () => { const store = createStore(); const submit = renderSubmitHook(store); @@ -122,33 +140,23 @@ describe("useUserIntentSubmit Agent Org intervention", () => { status: "queued", }), ]); - expect(mocks.enterIntervention).not.toHaveBeenCalled(); expect(mocks.dispatchMessageBySessionType).not.toHaveBeenCalled(); }); - it("does not dispatch when intervention persistence fails", async () => { + it("removes the optimistic user event and rejects when backend dispatch fails", async () => { const submit = renderSubmitHook(createStore()); - mocks.enterIntervention.mockRejectedValue( - new Error("intervention store unavailable") + mocks.dispatchMessageBySessionType.mockRejectedValue( + new Error("backend send unavailable") ); await expect( - submit({ - sessionId: SESSION_ID, - displayContent: "take over this worker", - swallowErrorAfterUserEventAppend: true, - }) - ).resolves.toBeUndefined(); + submit({ sessionId: SESSION_ID, displayContent: "retry me" }) + ).rejects.toThrow("backend send unavailable"); expect(mocks.addUserMessage).toHaveBeenCalledOnce(); - expect(mocks.enterIntervention).toHaveBeenCalledOnce(); - expect(mocks.dispatchMessageBySessionType).not.toHaveBeenCalled(); - expect(mocks.failOptimisticTurn).toHaveBeenCalledWith( - SESSION_ID, - "dispatch" + expect(mocks.removeByIdPrefix).toHaveBeenCalledWith( + "synthetic-user-1", + SESSION_ID ); - expect(mocks.markTurnTerminal).toHaveBeenCalledWith(SESSION_ID, "failed", { - generation: 7, - }); }); }); diff --git a/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.ts b/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.ts index a4c47cfca..7168d0b8c 100644 --- a/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.ts +++ b/src/engines/ChatPanel/hooks/useWorkspaceChat/useUserIntentSubmit.ts @@ -9,7 +9,6 @@ import { useAtomValue, useSetAtom, useStore } from "jotai"; import { useCallback, useEffect } from "react"; -import { enterAgentOrgSessionIntervention } from "@src/api/tauri/agent"; import type { AgentExecMode } from "@src/config/sessionCreatorConfig"; import { beginOptimisticTurn, @@ -21,12 +20,14 @@ import { getTurnPhase, markTurnTerminal, } from "@src/engines/SessionCore/control/turnLifecycle"; +import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; import { mintTurnIntentId } from "@src/engines/SessionCore/sync/adapters/shared/eventFactories"; import { type SessionRuntimeStatusSource, + closePostStopDispatchEpisodeAtom, isSessionActiveAtom, lastUserMessageAtom, - userInitiatedCancelAtom, + postStopDispatchSessionsAtom, } from "@src/store/session/cliSessionStatusAtom"; import { creatorDefaultExecModeAtom } from "@src/store/session/creatorDefaultExecModeAtom"; import { creatorDefaultModelSelectionAtom } from "@src/store/session/creatorDefaultModelAtom"; @@ -79,7 +80,6 @@ export interface SubmitUserIntentOptions { applyStopSubmitGuards?: boolean; dedupeDirectSubmit?: boolean; clearUserInitiatedCancelOnQueue?: boolean; - swallowErrorAfterUserEventAppend?: boolean; onQueued?: () => void; onBeforeDirectDispatch?: () => void; /** Stable caller-owned identity for observing a queued/direct dispatch. */ @@ -98,10 +98,10 @@ export function useUserIntentSubmit({ const enqueueMessage = useSetAtom(enqueueMessageAtom); const setQueueFlushRequest = useSetAtom(queueFlushRequestAtom); const setLastUserMessage = useSetAtom(lastUserMessageAtom); - const setUserInitiatedCancel = useSetAtom(userInitiatedCancelAtom); - const { addUserMessage, dispatchMessageBySessionType } = useMessageDispatch({ - getSessionId, - }); + const closePostStopDispatchEpisode = useSetAtom( + closePostStopDispatchEpisodeAtom + ); + const { addUserMessage, dispatchMessageBySessionType } = useMessageDispatch(); useEffect(() => { if (!isSessionActive) { @@ -120,7 +120,6 @@ export function useUserIntentSubmit({ applyStopSubmitGuards = false, dedupeDirectSubmit = false, clearUserInitiatedCancelOnQueue = false, - swallowErrorAfterUserEventAppend = false, onQueued, onBeforeDirectDispatch, turnIntentId: providedTurnIntentId, @@ -160,7 +159,8 @@ export function useUserIntentSubmit({ }) : false; const explicitPostStopSubmit = - restoredStopDraftSubmit || store.get(userInitiatedCancelAtom); + restoredStopDraftSubmit || + store.get(postStopDispatchSessionsAtom)[sessionId] === true; if ( dedupeDirectSubmit && @@ -195,7 +195,7 @@ export function useUserIntentSubmit({ store.get(creatorDefaultExecModeAtom); if (clearUserInitiatedCancelOnQueue && explicitPostStopSubmit) { - setUserInitiatedCancel(false); + closePostStopDispatchEpisode(sessionId); } enqueueMessage({ @@ -237,15 +237,16 @@ export function useUserIntentSubmit({ sharedSubmitPayload.current = submitPayloadKey; } - let userEventAppended = false; + let userEventId: string | null = null; let dispatchStarted = false; try { onBeforeDirectDispatch?.(); - await addUserMessage(displayContent, imageDataUrls, turnIntentId); - userEventAppended = true; - // This hook is the canonical user-intent boundary. The backend resolves - // the member from the session and ignores coordinator/non-org sessions. - await enterAgentOrgSessionIntervention(sessionId); + userEventId = await addUserMessage( + sessionId, + displayContent, + imageDataUrls, + turnIntentId + ); const displayTextForDispatch = contentForAgent !== displayContent ? displayContent : undefined; dispatchStarted = true; @@ -270,19 +271,25 @@ export function useUserIntentSubmit({ generation: dispatchGeneration, }); } - if (!userEventAppended || !swallowErrorAfterUserEventAppend) { - throw error; + if (userEventId) { + try { + await eventStoreProxy.removeByIdPrefix(userEventId, sessionId); + } catch { + // Preserve the original dispatch error. A failed cleanup must not + // turn an already-failed submit into a misleading success. + } } + throw error; } }, [ addUserMessage, + closePostStopDispatchEpisode, dispatchMessageBySessionType, enqueueMessage, getSessionId, setLastUserMessage, setQueueFlushRequest, - setUserInitiatedCancel, store, ] ); diff --git a/src/engines/ChatPanel/hooks/useWorkspaceChat/useWorkspaceChat.ts b/src/engines/ChatPanel/hooks/useWorkspaceChat/useWorkspaceChat.ts index f2c122f86..89b0d0da1 100644 --- a/src/engines/ChatPanel/hooks/useWorkspaceChat/useWorkspaceChat.ts +++ b/src/engines/ChatPanel/hooks/useWorkspaceChat/useWorkspaceChat.ts @@ -140,7 +140,6 @@ const useWorkspaceChat = (options: UseWorkspaceChatOptions = {}) => { applyStopSubmitGuards: true, dedupeDirectSubmit: true, clearUserInitiatedCancelOnQueue: true, - swallowErrorAfterUserEventAppend: true, onQueued: () => setSessChatInput(""), onBeforeDirectDispatch: () => setSessChatInput(""), }); diff --git a/src/engines/SessionCore/control/sessionTimelineBoundary.ts b/src/engines/SessionCore/control/sessionTimelineBoundary.ts index 1c5eb719e..2d2279557 100644 --- a/src/engines/SessionCore/control/sessionTimelineBoundary.ts +++ b/src/engines/SessionCore/control/sessionTimelineBoundary.ts @@ -15,10 +15,11 @@ import { markSessionStreamingStopped } from "@src/engines/SessionCore/sync/adapt import { createLogger } from "@src/hooks/logger"; import { killAgentShellProcess } from "@src/services/terminal"; import { + closePostStopDispatchEpisodeAtom, isPendingCancelAtom, + openPostStopDispatchEpisodeAtom, setSessionRuntimeStatusAtom, streamRetryStatusAtom, - userInitiatedCancelAtom, } from "@src/store/session/cliSessionStatusAtom"; import { shellProcessMapAtom } from "@src/store/session/shellProcessAtom"; import { @@ -190,13 +191,13 @@ export function beginTimelineBoundary( } if (effect.isUserStop) { - store.set(userInitiatedCancelAtom, true); + store.set(openPostStopDispatchEpisodeAtom, sessionId); store.set(isPendingCancelAtom, true); // Stop parks every queued follow-up of this session: the natural drain // skips them permanently; only an explicit Send Now dispatches them. store.set(holdSessionQueueForStopAtom, sessionId); } else { - store.set(userInitiatedCancelAtom, false); + store.set(closePostStopDispatchEpisodeAtom, sessionId); store.set(isPendingCancelAtom, false); } diff --git a/src/engines/SessionCore/hooks/session/__tests__/messageQueuePersistence.test.ts b/src/engines/SessionCore/hooks/session/__tests__/messageQueuePersistence.test.ts new file mode 100644 index 000000000..cd42da6ce --- /dev/null +++ b/src/engines/SessionCore/hooks/session/__tests__/messageQueuePersistence.test.ts @@ -0,0 +1,89 @@ +// @vitest-environment jsdom +import { createStore } from "jotai/vanilla"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { + type QueuedMessage, + messageQueueAtom, + messageQueueHydratedAtom, +} from "@src/store/ui/messageQueueAtom"; + +import { hydrateMessageQueue } from "../messageQueuePersistence"; + +const mocks = vi.hoisted(() => ({ + load: vi.fn(), + persist: vi.fn(), +})); + +vi.mock("@src/store/ui/messageQueueRepository", () => ({ + loadDurableMessageQueue: mocks.load, + persistDurableMessageQueue: mocks.persist, +})); + +function message( + id: string, + overrides: Partial = {} +): QueuedMessage { + return { + id, + turnIntentId: `intent-${id}`, + sessionId: "session-1", + content: id, + displayContent: id, + priority: "next", + status: "queued", + createdAt: `2026-07-23T00:00:0${id.length}.000Z`, + ...overrides, + }; +} + +describe("messageQueuePersistence", () => { + beforeEach(() => { + mocks.load.mockReset().mockResolvedValue([]); + mocks.persist.mockReset().mockResolvedValue(undefined); + }); + + it("hydrates before opening the dispatch gate", async () => { + const durable = message("durable"); + mocks.load.mockResolvedValue([durable]); + const store = createStore(); + + expect(store.get(messageQueueHydratedAtom)).toBe(false); + await hydrateMessageQueue(store); + + const recovered = { + ...durable, + priority: "next" as const, + requiresExplicitDispatch: true, + }; + expect(store.get(messageQueueAtom)).toEqual([recovered]); + expect(store.get(messageQueueHydratedAtom)).toBe(true); + expect(mocks.persist).toHaveBeenCalledWith([recovered]); + }); + + it("deduplicates by turn intent and lets live mutations win hydration races", async () => { + const durable = message("durable", { turnIntentId: "shared-intent" }); + const live = message("live", { + turnIntentId: "shared-intent", + content: "edited while loading", + }); + const store = createStore(); + store.set(messageQueueAtom, [live]); + mocks.load.mockResolvedValue([durable]); + + await hydrateMessageQueue(store); + + expect(store.get(messageQueueAtom)).toEqual([live]); + }); + + it("persists queue mutations after hydration", async () => { + const store = createStore(); + await hydrateMessageQueue(store); + mocks.persist.mockClear(); + + const next = message("next"); + store.set(messageQueueAtom, [next]); + + expect(mocks.persist).toHaveBeenCalledWith([next]); + }); +}); diff --git a/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts b/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts index 128aefcf3..709378b92 100644 --- a/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts +++ b/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts @@ -19,7 +19,6 @@ const mocks = vi.hoisted(() => ({ beginTurnDispatch: vi.fn(), cancelTurn: vi.fn(), confirmTurnRunning: vi.fn(), - enterIntervention: vi.fn(), failOptimisticTurn: vi.fn(), getSession: vi.fn(), getTurnPhase: vi.fn(), @@ -27,11 +26,11 @@ const mocks = vi.hoisted(() => ({ markTurnTerminal: vi.fn(), messageError: vi.fn(), messageWarning: vi.fn(), + removeByIdPrefix: vi.fn(), sendMessage: vi.fn(), })); vi.mock("@src/api/tauri/agent", () => ({ - enterAgentOrgSessionIntervention: mocks.enterIntervention, getSession: mocks.getSession, })); @@ -63,7 +62,10 @@ vi.mock("@src/engines/SessionCore/control/turnLifecycle", async () => { }); vi.mock("@src/engines/SessionCore/core/store/EventStoreProxy", () => ({ - eventStoreProxy: { append: mocks.append }, + eventStoreProxy: { + append: mocks.append, + removeByIdPrefix: mocks.removeByIdPrefix, + }, })); vi.mock("@src/engines/SessionCore/services/SessionService", () => ({ @@ -141,7 +143,6 @@ describe("useQueueDispatch Agent Org intervention", () => { mocks.beginTurnDispatch.mockReset().mockReturnValue(11); mocks.cancelTurn.mockReset().mockResolvedValue(undefined); mocks.confirmTurnRunning.mockReset(); - mocks.enterIntervention.mockReset().mockResolvedValue(true); mocks.failOptimisticTurn.mockReset(); mocks.getSession.mockReset().mockResolvedValue(null); mocks.getTurnPhase.mockReset().mockReturnValue("idle"); @@ -149,6 +150,7 @@ describe("useQueueDispatch Agent Org intervention", () => { mocks.markTurnTerminal.mockReset(); mocks.messageError.mockReset(); mocks.messageWarning.mockReset(); + mocks.removeByIdPrefix.mockReset().mockResolvedValue(1); mocks.sendMessage.mockReset().mockResolvedValue(undefined); store = createStore(); root = createSmokeRoot(); @@ -158,59 +160,83 @@ describe("useQueueDispatch Agent Org intervention", () => { await root.unmount(); }); - async function mountWithQueuedMessage(): Promise { - store.set(messageQueueAtom, [makeQueuedMessage()]); + async function mountWithMessages(messages: QueuedMessage[]): Promise { + store.set(messageQueueAtom, messages); await root.render( createElement(Provider, { store }, createElement(QueueDispatchHarness)) ); } - it("persists the queued event, marks intervention, and then dispatches", async () => { + async function mountWithQueuedMessage(): Promise { + await mountWithMessages([makeQueuedMessage()]); + } + + it("persists the queued event and dispatches it as direct user intent", async () => { await mountWithQueuedMessage(); await vi.waitFor(() => expect(mocks.sendMessage).toHaveBeenCalledOnce()); expect(mocks.append).toHaveBeenCalledOnce(); - expect(mocks.enterIntervention).toHaveBeenCalledWith(SESSION_ID); expect(mocks.sendMessage).toHaveBeenCalledWith( expect.objectContaining({ sessionId: SESSION_ID, content: "queued worker follow-up", turnIntentId: "turn-intent-queued-1", turnIntentSource: "force_send", + directUserIntent: true, }) ); expect(mocks.append.mock.invocationCallOrder[0]).toBeLessThan( - mocks.enterIntervention.mock.invocationCallOrder[0] - ); - expect(mocks.enterIntervention.mock.invocationCallOrder[0]).toBeLessThan( mocks.sendMessage.mock.invocationCallOrder[0] ); expect(store.get(messageQueueAtom)).toEqual([]); }); - it("parks the queued message without sending when intervention persistence fails", async () => { - mocks.enterIntervention.mockRejectedValue( - new Error("intervention store unavailable") + it("does not let a blocked Send Now freeze another idle session", async () => { + const blocked = makeQueuedMessage(); + const ready: QueuedMessage = { + ...makeQueuedMessage(), + id: "queued-other-session", + turnIntentId: "turn-intent-other-session", + sessionId: "agent-builtin:sde-other-session", + content: "independent follow-up", + displayContent: "independent follow-up", + priority: "next", + }; + mocks.getTurnPhase.mockImplementation((sessionId: string) => + sessionId === SESSION_ID ? "working" : "idle" + ); + + await mountWithMessages([blocked, ready]); + + await vi.waitFor(() => + expect(mocks.sendMessage).toHaveBeenCalledWith( + expect.objectContaining({ sessionId: ready.sessionId }) + ) ); + expect(mocks.cancelTurn).toHaveBeenCalledWith(SESSION_ID, "force-send"); + expect(store.get(messageQueueAtom)).toEqual([ + expect.objectContaining({ id: blocked.id }), + ]); + }); + + it("removes the optimistic queued event when backend dispatch fails", async () => { + mocks.sendMessage.mockRejectedValue(new Error("backend send unavailable")); await mountWithQueuedMessage(); await vi.waitFor(() => - expect(mocks.failOptimisticTurn).toHaveBeenCalledOnce() + expect(mocks.removeByIdPrefix).toHaveBeenCalledWith( + "synthetic-user-event", + SESSION_ID + ) ); - expect(mocks.sendMessage).not.toHaveBeenCalled(); - expect(mocks.markTurnTerminal).toHaveBeenCalledWith(SESSION_ID, "failed", { - generation: 11, - }); expect(store.get(messageQueueAtom)).toEqual([ expect.objectContaining({ id: "queued-intervention-1", - priority: "next", requiresExplicitDispatch: true, }), ]); - expect(mocks.messageError).toHaveBeenCalledOnce(); }); }); diff --git a/src/engines/SessionCore/hooks/session/backendDispatchVerdict.ts b/src/engines/SessionCore/hooks/session/backendDispatchVerdict.ts index 215f255a5..faa9c0f15 100644 --- a/src/engines/SessionCore/hooks/session/backendDispatchVerdict.ts +++ b/src/engines/SessionCore/hooks/session/backendDispatchVerdict.ts @@ -24,7 +24,7 @@ const BACKEND_DEAD_STATUSES = new Set([ "archived", ]); -export type BackendDispatchVerdict = "busy" | "dead" | "ready"; +export type BackendDispatchVerdict = "busy" | "dead" | "ready" | "unknown"; export function classifyBackendSessionStatus( status: string | undefined | null diff --git a/src/engines/SessionCore/hooks/session/messageQueuePersistence.ts b/src/engines/SessionCore/hooks/session/messageQueuePersistence.ts new file mode 100644 index 000000000..df6d9fa53 --- /dev/null +++ b/src/engines/SessionCore/hooks/session/messageQueuePersistence.ts @@ -0,0 +1,75 @@ +import type { Store } from "jotai/vanilla/store"; + +import { + type QueuedMessage, + messageQueueAtom, + messageQueueHydratedAtom, +} from "@src/store/ui/messageQueueAtom"; +import { + loadDurableMessageQueue, + persistDurableMessageQueue, +} from "@src/store/ui/messageQueueRepository"; + +const hydrationByStore = new WeakMap>(); +const unsubscribeByStore = new WeakMap void>(); + +function mergeQueues( + durable: readonly QueuedMessage[], + live: readonly QueuedMessage[] +): QueuedMessage[] { + const byIntent = new Map(); + // A persisted row may have crossed the backend-ACK/dequeue crash window. On + // recovery we cannot prove whether it was accepted, so never auto-replay it: + // keep it visible and require an explicit Send Now. Live rows created during + // hydration are known to belong to this renderer and therefore retain their + // natural dispatch policy. + for (const message of durable) { + byIntent.set(message.turnIntentId, { + ...message, + priority: "next", + requiresExplicitDispatch: true, + }); + } + // Live mutations made while the async disk read was pending win. + for (const message of live) byIntent.set(message.turnIntentId, message); + return [...byIntent.values()].sort((left, right) => + left.createdAt.localeCompare(right.createdAt) + ); +} + +/** + * Hydrate then subscribe one Jotai store. The WeakMap ownership supports test + * stores and multiple windows without app-lifetime listener leaks. + */ +export function hydrateMessageQueue(store: Store): Promise { + const existing = hydrationByStore.get(store); + if (existing) return existing; + + const hydration = loadDurableMessageQueue() + .then((durable) => { + store.set(messageQueueAtom, (live) => mergeQueues(durable, live)); + store.set(messageQueueHydratedAtom, true); + void persistDurableMessageQueue(store.get(messageQueueAtom)); + if (!unsubscribeByStore.has(store)) { + const unsubscribe = store.sub(messageQueueAtom, () => { + void persistDurableMessageQueue(store.get(messageQueueAtom)); + }); + unsubscribeByStore.set(store, unsubscribe); + } + }) + .catch(() => { + // The repository already logs the root error. Keep the queue usable in + // memory rather than blocking all sends when persistence is unavailable. + store.set(messageQueueHydratedAtom, true); + }); + + hydrationByStore.set(store, hydration); + return hydration; +} + +export function disposeMessageQueuePersistence(store: Store): void { + unsubscribeByStore.get(store)?.(); + unsubscribeByStore.delete(store); + hydrationByStore.delete(store); + store.set(messageQueueHydratedAtom, false); +} diff --git a/src/engines/SessionCore/hooks/session/useQueueDispatch.ts b/src/engines/SessionCore/hooks/session/useQueueDispatch.ts index 3ad7f5466..3aa766e21 100644 --- a/src/engines/SessionCore/hooks/session/useQueueDispatch.ts +++ b/src/engines/SessionCore/hooks/session/useQueueDispatch.ts @@ -25,10 +25,7 @@ import type { Atom } from "jotai"; import { useStore } from "jotai"; import { useCallback, useEffect, useRef } from "react"; -import { - enterAgentOrgSessionIntervention, - getSession, -} from "@src/api/tauri/agent"; +import { getSession } from "@src/api/tauri/agent"; import { Message } from "@src/components/Message"; import type { AgentExecMode } from "@src/config/sessionCreatorConfig"; import { @@ -50,9 +47,9 @@ import { createSyntheticUserEvent } from "@src/engines/SessionCore/sync/adapters import { createLogger } from "@src/hooks/logger"; import { markSessionActive } from "@src/store/session"; import { + closePostStopDispatchEpisodeAtom, lastUserMessageAtom, setSessionRuntimeStatusAtom, - userInitiatedCancelAtom, } from "@src/store/session/cliSessionStatusAtom"; import { creatorDefaultExecModeAtom } from "@src/store/session/creatorDefaultExecModeAtom"; import { @@ -63,6 +60,7 @@ import { sessionMapAtom } from "@src/store/session/sessionAtom"; import { type QueuedMessage, messageQueueAtom, + messageQueueHydratedAtom, queueEditingAtom, queueFlushRequestAtom, } from "@src/store/ui/messageQueueAtom"; @@ -78,6 +76,10 @@ import { type BackendDispatchVerdict, classifyBackendSessionStatus, } from "./backendDispatchVerdict"; +import { + disposeMessageQueuePersistence, + hydrateMessageQueue, +} from "./messageQueuePersistence"; const log = createLogger("useQueueDispatch"); @@ -107,9 +109,9 @@ const QUEUE_BACKEND_RECHECK_MS = 3_000; * session-status broadcasts). Dispatching on a falsely-idle FSM injects the * queued message into the middle of a still-running turn — or into a session * that already died. This asks the backend — the only authority on execution - * — before letting a natural drain proceed. Fail-open ("ready") on RPC - * errors: if the backend is unreachable the dispatch itself will fail and - * park the message. + * — before letting a natural drain proceed. Fail closed ("unknown") on RPC + * errors: a status-read failure does not prove that a turn is idle, so keep + * the durable queue row visible and retry instead of risking overlap. */ async function getBackendDispatchVerdict( sessionId: string @@ -126,13 +128,18 @@ async function getBackendDispatchVerdict( } return "ready"; } catch { - return "ready"; + return "unknown"; } } export function useQueueDispatch(): void { const store = useStore(); + useEffect(() => { + void hydrateMessageQueue(store); + return () => disposeMessageQueuePersistence(store); + }, [store]); + // ── Dispatch lock ───────────────────────────────────────────────────────── // One dispatch at a time, globally. The in-flight id additionally guards // the window between a successful send and the dequeue write. @@ -193,7 +200,7 @@ export function useQueueDispatch(): void { // An explicit dispatch concludes any pending stop episode. if (msg.priority === "now") { - store.set(userInitiatedCancelAtom, false); + store.set(closePostStopDispatchEpisodeAtom, sessionId); } // Capture the payload for Stop-restore before the async append. @@ -206,6 +213,7 @@ export function useQueueDispatch(): void { beginOptimisticTurn(sessionId, "queue"); void (async () => { + let userEventId: string | null = null; try { const userEvent = createSyntheticUserEvent( sessionId, @@ -215,10 +223,8 @@ export function useQueueDispatch(): void { turnIntentId: msg.turnIntentId, } ); + userEventId = userEvent.id; await eventStoreProxy.append([userEvent], sessionId); - // A queued user turn becomes a takeover only when it is actually - // dispatched. Merely waiting in the queue must not suppress Wake. - await enterAgentOrgSessionIntervention(sessionId); // Pass displayContent as displayText when it differs from content // (i.e. skill pills were expanded) so the persisted event stores // the pill format and re-editing shows the pill, not the YAML. @@ -235,6 +241,7 @@ export function useQueueDispatch(): void { clientMessageId: `queued:${sessionId}:${msg.id}`, turnIntentId: msg.turnIntentId, turnIntentSource: msg.priority === "now" ? "force_send" : "queue", + directUserIntent: true, }); // Backend accepted the message — confirm the turn as running. confirmTurnRunning(sessionId); @@ -260,6 +267,16 @@ export function useQueueDispatch(): void { } } catch (err) { log.error("[useQueueDispatch] dispatch failed:", err); + if (userEventId) { + try { + await eventStoreProxy.removeByIdPrefix(userEventId, sessionId); + } catch (cleanupError) { + log.warn( + "[useQueueDispatch] failed to remove optimistic user event:", + cleanupError + ); + } + } // IPC failed before the backend received the message: close the // reserved turn and park the message so it does not retry in a // tight loop — the user can fix the issue and press Send Now. @@ -292,6 +309,7 @@ export function useQueueDispatch(): void { wakeTimerRef.current = null; } if (dispatchLockRef.current) return; + if (!store.get(messageQueueHydratedAtom)) return; if (store.get(queueEditingAtom)) return; const queue = store.get(messageQueueAtom); @@ -303,9 +321,12 @@ export function useQueueDispatch(): void { !sentQueuedMessageIdsRef.current.has(msg.id) ); - // ── Explicit "now" dispatches take absolute precedence ───────────────── - const explicitMsg = candidates.find((msg) => msg.priority === "now"); - if (explicitMsg) { + // ── Explicit "now" dispatches take absolute precedence per session ─────── + // A blocked Send Now for session A must not freeze an idle session B. Scan + // every explicit candidate, dispatch the first idle one, and request at + // most one interrupt for each active message while continuing the pass. + const explicitMessages = candidates.filter((msg) => msg.priority === "now"); + for (const explicitMsg of explicitMessages) { const phase = getTurnPhase(explicitMsg.sessionId); if (phase === "idle") { dispatchLockRef.current = true; @@ -324,22 +345,25 @@ export function useQueueDispatch(): void { !interruptRequestedByMessageIdRef.current.has(explicitMsg.id) ) { // Send Now against an active turn: interrupt it once. The provider's - // cancelled terminal flips the FSM idle, which re-triggers this pass - // and dispatches the message above. + // cancelled terminal flips the FSM idle, which re-triggers this pass. interruptRequestedByMessageIdRef.current.add(explicitMsg.id); void cancelTurnForTimelineBoundary( explicitMsg.sessionId, "force-send" ).catch((error) => { + // A failed interrupt must be retryable. Keeping the id in this set + // would strand the message until an unrelated lifecycle signal. + interruptRequestedByMessageIdRef.current.delete(explicitMsg.id); log.warn("[useQueueDispatch] force-send interrupt failed:", error); }); } - // stopping (or interrupt already requested): wait for the terminal. - return; + // `stopping` and already-requested interrupts wait for their own + // terminal, but do not block dispatchable work in another session. } // ── Natural FIFO drain ────────────────────────────────────────────────── for (const msg of candidates) { + if (msg.priority === "now") continue; if (msg.requiresExplicitDispatch) continue; // held by a user Stop if (getTurnPhase(msg.sessionId) !== "idle") continue; // turn active const remainingVisibleMs = MIN_QUEUE_VISIBLE_MS - queuedMessageAgeMs(msg); @@ -357,10 +381,9 @@ export function useQueueDispatch(): void { // backend before injecting a natural follow-up into the session. void getBackendDispatchVerdict(msg.sessionId).then((verdict) => { if (inFlightMessageIdRef.current !== msg.id) return; - if (verdict === "busy") { - // Still executing — back off and re-check. Do NOT mark the FSM: - // presentation state may legitimately disagree; the queue only - // needs to know "not yet". + if (verdict === "busy" || verdict === "unknown") { + // Still executing or backend state is unknown — back off and + // re-check. Never infer idle from a failed status read. inFlightMessageIdRef.current = null; dispatchLockRef.current = false; if (wakeTimerRef.current === null) { @@ -420,6 +443,7 @@ export function useQueueDispatch(): void { useEffect(() => { const unsubscribers = [ store.sub(messageQueueAtom as Atom, tryDispatchNext), + store.sub(messageQueueHydratedAtom as Atom, tryDispatchNext), store.sub(turnLifecycleSignalAtom as Atom, tryDispatchNext), store.sub(queueFlushRequestAtom as Atom, tryDispatchNext), store.sub(queueEditingAtom as Atom, tryDispatchNext), diff --git a/src/engines/SessionCore/hooks/session/useSessionManager.ts b/src/engines/SessionCore/hooks/session/useSessionManager.ts index de9ecfa93..6d97d19f1 100644 --- a/src/engines/SessionCore/hooks/session/useSessionManager.ts +++ b/src/engines/SessionCore/hooks/session/useSessionManager.ts @@ -55,21 +55,16 @@ export function useSessionManager( const repos = useAtomValue(reposAtom); - const isLoadingRef = useRef(false); // Mirror sessions.length in a ref so loadSessions can read it without // being recreated every time the list grows. Without this, sessions.length // in the dep array causes loadSessions to change identity after every load, - // which re-fires the autoLoad useEffect and risks a self-exciting loop if - // the isLoadingRef guard is ever cleared while a fetch is still in-flight - // (e.g. the forceRefresh path calls resetSessionStore() and resets it). + // which re-fires the autoLoad useEffect after every store update. const sessionsLengthRef = useRef(sessions.length); - sessionsLengthRef.current = sessions.length; + useEffect(() => { + sessionsLengthRef.current = sessions.length; + }, [sessions.length]); const loadSessions = useCallback(async () => { - if (isLoadingRef.current) { - return; - } - const invalidationTimestamp = getSessionCacheInvalidationTimestamp(); const cacheWasInvalidated = invalidationTimestamp !== null && @@ -81,25 +76,24 @@ export function useSessionManager( localStorage.removeItem(SESSION_CACHE_INVALIDATION_KEY); } - isLoadingRef.current = true; - try { await centralLoadSessions({ forceRefresh: cacheWasInvalidated || sessionsLengthRef.current === 0, }); } catch (err) { log.error("[useSessionManager] Failed to load sessions:", err); - } finally { - isLoadingRef.current = false; } }, [lastLoadedAt]); const forceRefresh = useCallback(async () => { resetSessionStore(); localStorage.removeItem(SESSION_CACHE_INVALIDATION_KEY); - isLoadingRef.current = false; - await loadSessions(); - }, [loadSessions]); + try { + await centralLoadSessions({ forceRefresh: true }); + } catch (err) { + log.error("[useSessionManager] Failed to refresh sessions:", err); + } + }, []); useEffect(() => { if (autoLoad && repos.length > 0) { diff --git a/src/engines/SessionCore/services/SessionService.ts b/src/engines/SessionCore/services/SessionService.ts index b05c09a70..763a16260 100644 --- a/src/engines/SessionCore/services/SessionService.ts +++ b/src/engines/SessionCore/services/SessionService.ts @@ -301,6 +301,7 @@ export const SessionService = { clientMessageId, turnIntentId, turnIntentSource, + directUserIntent, } = params; // Gate ADE context on the session row's persisted repo so a session // on repo A doesn't ship repo B's editor / git / LSP state when the @@ -359,6 +360,7 @@ export const SessionService = { clientMessageId, turnIntentId, turnIntentSource, + directUserIntent, adeContext, sessionRepoPath: sessionRow?.repoPath ?? null, }); diff --git a/src/engines/SessionCore/services/types.ts b/src/engines/SessionCore/services/types.ts index 29052e6b5..7c7a8ee0c 100644 --- a/src/engines/SessionCore/services/types.ts +++ b/src/engines/SessionCore/services/types.ts @@ -108,6 +108,12 @@ export interface SessionSendMessageParams { * submit so a still-cancelling backend turn cannot absorb it as steering. */ turnIntentSource: TurnIntentSource; + /** + * True only for a real user-authored prompt. Rust-native adapters persist + * Agent Org intervention inside the same backend acceptance boundary; CLI + * adapters apply it immediately after their command accepts the rerun. + */ + directUserIntent?: boolean; /** * When `true`, this is a user-initiated Resume after a failed turn. * Backend runs deletion-based orphan tool-use filter. diff --git a/src/engines/SessionCore/sync/__tests__/sessionSyncStateHelpers.test.ts b/src/engines/SessionCore/sync/__tests__/sessionSyncStateHelpers.test.ts index 25e0c01ba..95ab8a3d5 100644 --- a/src/engines/SessionCore/sync/__tests__/sessionSyncStateHelpers.test.ts +++ b/src/engines/SessionCore/sync/__tests__/sessionSyncStateHelpers.test.ts @@ -19,6 +19,14 @@ import type { SessionEventHandlerStateActions } from "@src/engines/SessionCore/s import { updateSessionStatus } from "@src/store/session"; import { createInstrumentedStore } from "@src/util/core/state/instrumentedStore"; +const mocks = vi.hoisted(() => ({ + getTurnIntentDispatch: vi.fn(), +})); + +vi.mock("@src/engines/SessionCore/control/turnIntentDispatchLifecycle", () => ({ + getTurnIntentDispatch: mocks.getTurnIntentDispatch, +})); + createInstrumentedStore(); vi.mock("@src/engines/SessionCore/core/store/EventStoreProxy", () => ({ @@ -70,6 +78,7 @@ function createActions(): SessionEventHandlerStateActions & { describe("session sync state callbacks", () => { beforeEach(() => { vi.clearAllMocks(); + mocks.getTurnIntentDispatch.mockReturnValue(undefined); }); it("clears live streaming content before completed status can leave Stop UI stuck", () => { @@ -144,7 +153,53 @@ describe("session sync state callbacks", () => { turnStatus: "completed", }); - expect(markTurnTerminal).toHaveBeenCalledWith("session-1", "completed"); + expect(markTurnTerminal).toHaveBeenCalledWith("session-1", "completed", { + generation: undefined, + }); + }); + + it("passes the exact dispatched generation for an attributed terminal", () => { + mocks.getTurnIntentDispatch.mockReturnValue({ + sessionId: "session-1", + generation: 17, + }); + const callbacks = createSessionEventHandlerCallbacks( + "session-1", + createActions(), + vi.fn() + ); + + callbacks.onStatusChange?.("completed", undefined, { + turnIntentId: "intent-17", + turnStatus: "completed", + }); + + expect(mocks.getTurnIntentDispatch).toHaveBeenCalledWith("intent-17"); + expect(markTurnTerminal).toHaveBeenCalledWith("session-1", "completed", { + generation: 17, + }); + }); + + it("rejects a terminal intent attributed to another session", () => { + mocks.getTurnIntentDispatch.mockReturnValue({ + sessionId: "session-other", + generation: 8, + }); + const actions = createActions(); + const callbacks = createSessionEventHandlerCallbacks( + "session-1", + actions, + vi.fn() + ); + + callbacks.onStatusChange?.("completed", undefined, { + turnIntentId: "cross-session-intent", + }); + + expect(markTurnTerminal).not.toHaveBeenCalled(); + expect(actions.setSessionRuntimeStatus).not.toHaveBeenCalled(); + expect(actions.setPendingCancel).not.toHaveBeenCalled(); + expect(updateSessionStatus).not.toHaveBeenCalled(); }); it("does NOT mark the FSM terminal for intermediate status signals", () => { diff --git a/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts b/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts index 5b6438444..e65cffbd2 100644 --- a/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts +++ b/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts @@ -6,6 +6,7 @@ const mocks = vi.hoisted(() => ({ enterIntervention: vi.fn(), message: vi.fn(), registerReceipt: vi.fn(), + warn: vi.fn(), })); vi.mock("@src/api/tauri/agent", () => ({ @@ -17,6 +18,9 @@ vi.mock("@src/api/tauri/rpc", () => ({ vi.mock("@src/hooks/cliSession/cliTurnLifecycleCoordinator", () => ({ cliTurnLifecycleCoordinator: { registerReceipt: mocks.registerReceipt }, })); +vi.mock("@src/hooks/logger", () => ({ + createLogger: () => ({ warn: mocks.warn }), +})); describe("sendCliMessage acceptance boundary", () => { beforeEach(() => { @@ -26,7 +30,7 @@ describe("sendCliMessage acceptance boundary", () => { turnIntentId: "intent-1", status: "running", }); - mocks.enterIntervention.mockResolvedValue(undefined); + mocks.enterIntervention.mockReturnValue(new Promise(() => undefined)); }); it("resolves from the receipt without status or history reconciliation", async () => { @@ -37,6 +41,7 @@ describe("sendCliMessage acceptance boundary", () => { turnIntentId: "intent-1", clientMessageId: "message-1", turnIntentSource: "user_submit", + directUserIntent: true, }) ).resolves.toBeUndefined(); @@ -51,16 +56,16 @@ describe("sendCliMessage acceptance boundary", () => { turnIntentId: "intent-1", status: "running", }); + expect(mocks.enterIntervention).toHaveBeenCalledWith("cliagent-worker"); }); - it("rejects when the backend command rejects", async () => { + it("rejects only when the backend command rejects", async () => { mocks.message.mockRejectedValue(new Error("ipc unavailable")); await expect( sendCliMessage({ sessionId: "cliagent-worker", content: "retry", - isResume: true, turnIntentId: "intent-2", clientMessageId: "message-2", turnIntentSource: "user_submit", @@ -68,5 +73,6 @@ describe("sendCliMessage acceptance boundary", () => { ).rejects.toThrow("ipc unavailable"); expect(mocks.registerReceipt).not.toHaveBeenCalled(); + expect(mocks.enterIntervention).not.toHaveBeenCalled(); }); }); diff --git a/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts b/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts index b19d70e49..7f132a744 100644 --- a/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts +++ b/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts @@ -2,9 +2,12 @@ import { enterAgentOrgSessionIntervention } from "@src/api/tauri/agent"; import type { CancelReason } from "@src/api/tauri/agent/session"; import { rpc } from "@src/api/tauri/rpc"; import { cliTurnLifecycleCoordinator } from "@src/hooks/cliSession/cliTurnLifecycleCoordinator"; +import { createLogger } from "@src/hooks/logger"; import type { AdapterSendInput } from "../../types"; +const log = createLogger("CliTransport"); + function newMessageId(): string { return crypto.randomUUID(); } @@ -18,11 +21,8 @@ export async function sendCliMessage(input: AdapterSendInput): Promise { mode, imageDataUrls, adeContext, - isResume, + directUserIntent, } = input; - if (!isResume && content.trim()) { - await enterAgentOrgSessionIntervention(sessionId); - } const turnIntentId = input.turnIntentId ?? newMessageId(); const clientMessageId = input.clientMessageId ?? newMessageId(); const receipt = await rpc.cli.message({ @@ -38,7 +38,16 @@ export async function sendCliMessage(input: AdapterSendInput): Promise { : {}), ...(adeContext ? { ideContext: adeContext } : {}), }); + cliTurnLifecycleCoordinator.registerReceipt(receipt); + if (directUserIntent) { + void enterAgentOrgSessionIntervention(sessionId).catch((error) => { + log.warn( + "[sendCliMessage] accepted CLI turn but failed to persist intervention:", + error + ); + }); + } } export async function stopCliSession( diff --git a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.streaming.test.ts b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.streaming.test.ts new file mode 100644 index 000000000..2708120c3 --- /dev/null +++ b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.streaming.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it, vi } from "vitest"; + +import { createStreamingEdgeController } from "./createRustAgentAdapter"; + +describe("Rust Agent streaming edge controller", () => { + it("coalesces 1,000 deltas into one true and one terminal false write", () => { + const write = vi.fn(); + const controller = createStreamingEdgeController(write); + + for (let index = 0; index < 1_000; index += 1) { + controller.set(true); + } + controller.set(false); + controller.set(false); + controller.set(false); + + expect(write.mock.calls).toEqual([[true], [false]]); + expect(controller.value).toBe(false); + }); + + it("forces an unknown owner to converge to false only once", () => { + const write = vi.fn(); + const controller = createStreamingEdgeController(write); + + controller.set(false); + controller.set(false); + + expect(write).toHaveBeenCalledOnce(); + expect(write).toHaveBeenCalledWith(false); + }); +}); diff --git a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts index d8c3d51d5..ad88e8e66 100644 --- a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts +++ b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts @@ -144,6 +144,27 @@ const LIVE_STREAM_EVENTS_IGNORED_AFTER_STOP = new Set([ "agent:streaming_complete", ]); +export interface StreamingEdgeController { + readonly value: boolean; + set(value: boolean): void; +} + +export function createStreamingEdgeController( + write: (value: boolean) => void +): StreamingEdgeController { + let lastValue: boolean | undefined; + return { + get value(): boolean { + return lastValue ?? false; + }, + set(value: boolean): void { + if (lastValue === value) return; + lastValue = value; + write(value); + }, + }; +} + interface TokenUsageRecord { inputTokens: number; contextTokens: number; @@ -224,6 +245,17 @@ export function createRustAgentAdapter( transformUserText, features, } = config; + const streamingControllers = new Map(); + + const getStreamingController = (sessionId: string) => { + const existing = streamingControllers.get(sessionId); + if (existing) return existing; + const created = createStreamingEdgeController((value) => { + void eventStoreProxy.setStreaming(value, sessionId); + }); + streamingControllers.set(sessionId, created); + return created; + }; return { category, @@ -332,7 +364,7 @@ export function createRustAgentAdapter( sessionId: string, callbacks: EventHandlerCallbacks ): SessionEventHandler { - let _streaming = false; + const streamingController = getStreamingController(sessionId); // Two-flag system for status signaling: // @@ -461,12 +493,7 @@ export function createRustAgentAdapter( } : undefined, setStreaming: (value: boolean) => { - // Token/thinking deltas all assert the same active state. Crossing the - // Tauri boundary for every chunk only rewrites one bool, yet it can - // generate hundreds of IPC calls per minute during a long reply. - if (_streaming === value) return; - _streaming = value; - void eventStoreProxy.setStreaming(value, sessionId); + streamingController.set(value); }, }); @@ -630,20 +657,22 @@ export function createRustAgentAdapter( ctx.trackedCodingSessionsRef?.current.clear(); - _streaming = false; _runningSignaled = false; _turnCompleted = false; _consecutiveDispatchFailures = 0; - eventStoreProxy.setStreaming(false, sessionId); + streamingController.set(false); }, get isStreaming(): boolean { - return _streaming; + return streamingController.value; }, dispose(): void { _disposed = true; this.reset(); + if (streamingControllers.get(sessionId) === streamingController) { + streamingControllers.delete(sessionId); + } }, }; }, @@ -660,7 +689,11 @@ export function createRustAgentAdapter( async stopSession(sessionId: string, reason: CancelReason): Promise { markSessionStreamingStopped(sessionId); - void eventStoreProxy.setStreaming(false, sessionId); + const existingController = streamingControllers.get(sessionId); + const streamingController = + existingController ?? getStreamingController(sessionId); + streamingController.set(false); + if (!existingController) streamingControllers.delete(sessionId); await cancel(sessionId, reason); }, }; diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/forkTerminalBridge.test.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/forkTerminalBridge.test.ts index 6934cd939..d0eb701ab 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/forkTerminalBridge.test.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/forkTerminalBridge.test.ts @@ -72,7 +72,9 @@ describe("relay fork terminals are never swallowed by the subagent bridge", () = ctx ); - expect(onStatusChange).toHaveBeenCalledWith("completed"); + expect(onStatusChange).toHaveBeenCalledWith("completed", undefined, { + intermediate: true, + }); }); it("still dispatches normally when an untracked spawned id has no active parent call", async () => { diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/sessionHandlers.test.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/sessionHandlers.test.ts index 90b1c2d30..192365768 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/sessionHandlers.test.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/sessionHandlers.test.ts @@ -200,6 +200,7 @@ describe("Rust Agent session handlers", () => { type: "agent:turn_completed", sessionId: "session-1", turnId: "turn-1", + turnIntentId: "intent-1", turnStatus: "completed", sessionStatus: "idle", }, @@ -213,6 +214,7 @@ describe("Rust Agent session handlers", () => { undefined, { turnId: "turn-1", + turnIntentId: "intent-1", turnStatus: "completed", } ); diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/streamHandlers.test.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/streamHandlers.test.ts index 34ea54e7e..a98be53f3 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/streamHandlers.test.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/__tests__/streamHandlers.test.ts @@ -10,7 +10,11 @@ import { } from "../streamHandlers"; import { clearSessionStreamingStopped, + disposeSessionStreamingState, + getActiveSessionStreamingTurn, + isSessionStreamingStopped, markSessionStreamingStopped, + noteSessionStreamingTurn, } from "../streamHelpers"; import type { EventHandlerContext } from "../types"; @@ -377,7 +381,11 @@ describe("Rust Agent stream handlers", () => { content: "", }); expect(ctx.setStreaming).toHaveBeenCalledWith(false); - expect(ctx.onStatusChangeRef.current).toHaveBeenCalledWith("completed"); + expect(ctx.onStatusChangeRef.current).toHaveBeenCalledWith( + "completed", + undefined, + { intermediate: true } + ); await dispatchAgentEvent( { @@ -472,3 +480,56 @@ describe("Rust Agent stream handlers", () => { expect(removeByIdPrefixSpy).not.toHaveBeenCalled(); }); }); + +describe("streamHelpers session-scoped memory cleanup", () => { + beforeEach(() => { + disposeSessionStreamingState("mem-session"); + }); + + it("disposeSessionStreamingState purges retained turn-level stop markers", () => { + noteSessionStreamingTurn("mem-session", "turn-x"); + markSessionStreamingStopped("mem-session"); + expect(isSessionStreamingStopped("mem-session", "turn-x")).toBe(true); + + disposeSessionStreamingState("mem-session"); + + // The whole session entry is gone — no turn-level suppression survives a + // permanent deletion, so nothing is retained for a re-created session id. + expect(isSessionStreamingStopped("mem-session", "turn-x")).toBe(false); + expect(getActiveSessionStreamingTurn("mem-session")).toBeUndefined(); + }); + + it("clearSessionStreamingStopped preserves turn-level markers (resume contract)", () => { + noteSessionStreamingTurn("mem-session", "turn-x"); + markSessionStreamingStopped("mem-session"); + + clearSessionStreamingStopped("mem-session"); + + // Resume must keep suppressing late events for the already-stopped turn... + expect(isSessionStreamingStopped("mem-session", "turn-x")).toBe(true); + // ...but the active-turn pointer is released. + expect(getActiveSessionStreamingTurn("mem-session")).toBeUndefined(); + + disposeSessionStreamingState("mem-session"); + }); + + it("bounds retained sessions with an LRU safety cap even without disposal", () => { + const SESSIONS = 300; + for (let idx = 0; idx < SESSIONS; idx += 1) { + const id = `cap-session-${idx}`; + noteSessionStreamingTurn(id, "turn-1"); + markSessionStreamingStopped(id); + } + + // The oldest-inserted session is evicted once the cap is exceeded... + expect(isSessionStreamingStopped("cap-session-0", "turn-1")).toBe(false); + // ...while the most recently touched session is retained. + expect( + isSessionStreamingStopped(`cap-session-${SESSIONS - 1}`, "turn-1") + ).toBe(true); + + for (let idx = 0; idx < SESSIONS; idx += 1) { + disposeSessionStreamingState(`cap-session-${idx}`); + } + }); +}); diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/sessionHandlers.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/sessionHandlers.ts index 0afcec52d..443c2a495 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/sessionHandlers.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/sessionHandlers.ts @@ -24,7 +24,7 @@ function settleTerminalRuntime( ctx: EventHandlerContext, status: "completed" | "failed" | "cancelled" = "completed", errorMessage?: string, - meta?: { turnId?: string; turnStatus?: string } + meta?: { turnId?: string; turnIntentId?: string; turnStatus?: string } ): void { resetAllStreamingState(ctx); ctx.setStreaming(false); @@ -77,7 +77,12 @@ export function handleComplete( } : undefined; ctx.onAgentCompleteRef.current?.(tokenUsage); - ctx.onStatusChangeRef.current?.("completed"); + // `agent:complete` carries content/usage, but authoritative finality is the + // following `agent:turn_completed`, which also carries turnIntentId. Treat + // this as intermediate so a delayed complete cannot release a newer turn. + ctx.onStatusChangeRef.current?.("completed", undefined, { + intermediate: true, + }); } export function handleTurnCompleted( @@ -94,6 +99,7 @@ export function handleTurnCompleted( const status = cancelled ? "cancelled" : failed ? "failed" : "completed"; settleTerminalRuntime(sessionId, ctx, status, undefined, { turnId: event.turnId, + turnIntentId: event.turnIntentId, turnStatus: event.turnStatus, }); } @@ -157,7 +163,15 @@ export function handleError( clearStreamRetryStatus(ctx, sessionId); // Status change fires before onAgentComplete so session activity is already // terminal before completion callbacks update derived session state. - ctx.onStatusChangeRef.current?.("failed", event.error); + const turnIntentId = + event.details && "turnIntentId" in event.details + ? String(event.details.turnIntentId) + : undefined; + ctx.onStatusChangeRef.current?.( + "failed", + event.error, + turnIntentId ? { turnIntentId } : undefined + ); ctx.onAgentCompleteRef.current?.(); } diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers.ts index 642e777e1..30a519c6b 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers.ts @@ -9,6 +9,17 @@ import type { EventHandlerContext } from "./types"; const STOPPED_TURNS_PER_SESSION_LIMIT = 20; +/** + * Safety cap on how many sessions `stoppedStreamingTurnsBySession` retains. + * + * Permanent session removal purges entries via `disposeSessionStreamingState`, + * so in practice this map only holds live sessions. This LRU-by-insertion cap + * is defense-in-depth: it keeps the map bounded by a constant even if a future + * deletion path forgets to dispose, so the module can never grow with the + * lifetime session count. + */ +const MAX_STOPPED_TURN_SESSIONS = 256; + const stoppedStreamingSessions = new Set(); const activeStreamingTurnBySession = new Map(); const stoppedStreamingTurnsBySession = new Map>(); @@ -21,6 +32,16 @@ function resetStreamRefs(refs: StreamRefs): void { function stoppedTurnSetForSession(sessionId: string): Set { let stoppedTurns = stoppedStreamingTurnsBySession.get(sessionId); if (!stoppedTurns) { + if (stoppedStreamingTurnsBySession.size >= MAX_STOPPED_TURN_SESSIONS) { + // Evict the oldest-inserted session; its turn-level stop markers are the + // least likely to still be receiving late events. + const oldestSessionId = stoppedStreamingTurnsBySession + .keys() + .next().value; + if (oldestSessionId !== undefined) { + stoppedStreamingTurnsBySession.delete(oldestSessionId); + } + } stoppedTurns = new Set(); stoppedStreamingTurnsBySession.set(sessionId, stoppedTurns); } @@ -67,6 +88,21 @@ export function clearSessionStreamingStopped(sessionId: string): void { activeStreamingTurnBySession.delete(sessionId); } +/** + * Permanently release all retained streaming-stop state for a session. + * + * `clearSessionStreamingStopped` runs on resume/restart and deliberately keeps + * `stoppedStreamingTurnsBySession` so turn-level stop suppression survives a + * resume. On permanent session removal that per-turn set has no further use, so + * purge all three maps. Call this from the session-deletion path — without it + * `stoppedStreamingTurnsBySession` accrues one entry per lifetime session. + */ +export function disposeSessionStreamingState(sessionId: string): void { + stoppedStreamingSessions.delete(sessionId); + activeStreamingTurnBySession.delete(sessionId); + stoppedStreamingTurnsBySession.delete(sessionId); +} + export function isSessionStreamingStopped( sessionId: string, turnId?: string diff --git a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/types.ts b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/types.ts index 5c54cf1da..36f862199 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/types.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/types.ts @@ -64,7 +64,12 @@ export interface EventHandlerContext { | (( status: string, errorMessage?: string, - meta?: { turnId?: string; turnStatus?: string; intermediate?: boolean } + meta?: { + turnId?: string; + turnIntentId?: string; + turnStatus?: string; + intermediate?: boolean; + } ) => void) | undefined >; diff --git a/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts b/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts index fe2a65759..b95c56f71 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts @@ -17,6 +17,7 @@ export function buildRustAgentSendMessageArgs( clientMessageId, turnIntentId, turnIntentSource, + directUserIntent, sessionRepoPath, } = input; const workspacePath = sessionRepoPath ?? undefined; @@ -36,6 +37,7 @@ export function buildRustAgentSendMessageArgs( ...(isResume ? { isResume: true } : {}), ...(clientMessageId ? { clientMessageId } : {}), ...(turnIntentId ? { turnIntentId } : {}), + ...(directUserIntent ? { markDirectUserIntervention: true } : {}), turnIntentSource, }; } diff --git a/src/engines/SessionCore/sync/adapters/shared/types.ts b/src/engines/SessionCore/sync/adapters/shared/types.ts index 5e1de643e..d1bc38811 100644 --- a/src/engines/SessionCore/sync/adapters/shared/types.ts +++ b/src/engines/SessionCore/sync/adapters/shared/types.ts @@ -59,6 +59,10 @@ export interface StreamingErrorDetails { toolName?: string; /** File path if applicable. */ filePath?: string; + /** Queue message identity when a scheduled turn fails before normal finality. */ + messageId?: string; + /** Canonical intent identity for attributing the failure to one turn. */ + turnIntentId?: string; } /** @@ -192,6 +196,8 @@ export interface AgentWSEvent { /** Turn summary text (agent:turn_summary) */ summary?: string; + /** Canonical user-intent id carried by authoritative terminal events. */ + turnIntentId?: string; /** Stable turn id for anchoring post-turn summary events. */ turnId?: string; /** Transcript timestamp for anchoring post-turn summary events. */ diff --git a/src/engines/SessionCore/sync/sessionSyncStateHelpers.ts b/src/engines/SessionCore/sync/sessionSyncStateHelpers.ts index 9e784f309..156c04de8 100644 --- a/src/engines/SessionCore/sync/sessionSyncStateHelpers.ts +++ b/src/engines/SessionCore/sync/sessionSyncStateHelpers.ts @@ -1,6 +1,7 @@ import type { SetStateAction } from "react"; import { wasRecentlyOptimisticallyStarted } from "@src/engines/SessionCore/control/optimisticTurnStatus"; +import { getTurnIntentDispatch } from "@src/engines/SessionCore/control/turnIntentDispatchLifecycle"; import { markTurnRunning, markTurnTerminal, @@ -231,6 +232,14 @@ export function createSessionEventHandlerCallbacks( // the UI mirror, pendingCancel, pin state, and the session row all // leaked the phantom terminal. if (meta?.intermediate) return; + const terminalDispatch = + TERMINAL_HANDLER_STATUSES.has(status) && meta?.turnIntentId + ? getTurnIntentDispatch(meta.turnIntentId) + : undefined; + // Reject a misrouted terminal before it mutates any UI mirror or durable + // session status. Finality attribution and presentation state must move + // together or not at all. + if (terminalDispatch && terminalDispatch.sessionId !== sessionId) return; actions.setSessionRuntimeStatus(toCliSessionStatus(status)); if (status === "failed" && errorMessage) { actions.setSessionRuntimeError(errorMessage); @@ -240,7 +249,8 @@ export function createSessionEventHandlerCallbacks( // here. Intermediate signals already returned above. markTurnTerminal( sessionId, - toTurnTerminalStatus(meta?.turnStatus ?? status) + toTurnTerminalStatus(meta?.turnStatus ?? status), + { generation: terminalDispatch?.generation } ); actions.setPendingCancel(false); eventStoreProxy.unpinSession(sessionId); diff --git a/src/engines/SessionCore/sync/types.ts b/src/engines/SessionCore/sync/types.ts index 38f5151b5..9b0d609e2 100644 --- a/src/engines/SessionCore/sync/types.ts +++ b/src/engines/SessionCore/sync/types.ts @@ -88,7 +88,12 @@ export interface EventHandlerCallbacks { onStatusChange?: ( status: string, error?: string, - meta?: { turnId?: string; turnStatus?: string; intermediate?: boolean } + meta?: { + turnId?: string; + turnIntentId?: string; + turnStatus?: string; + intermediate?: boolean; + } ) => void; /** Called when CLI token usage updates. */ onTokenUpdate?: (tokens: number) => void; @@ -159,6 +164,8 @@ export interface AdapterSendInput { turnIntentId?: string; /** Origin of this turn at the user-intent boundary. */ turnIntentSource: TurnIntentSource; + /** True only for a real user-authored prompt (not resume/wake/continuation). */ + directUserIntent?: boolean; /** * When `true`, this is a user-initiated Resume after a failed turn. * The backend runs deletion-based orphan tool-use filter instead of diff --git a/src/store/session/cliSessionStatusAtom.ts b/src/store/session/cliSessionStatusAtom.ts index 6f7714e6d..95bb8ed4b 100644 --- a/src/store/session/cliSessionStatusAtom.ts +++ b/src/store/session/cliSessionStatusAtom.ts @@ -241,8 +241,35 @@ isPendingCancelAtom.debugLabel = "isPendingCancel"; * Cleared by `useQueueDispatch` after it consumes the restore, or on the next * fresh `status_changed -> running` event, whichever comes first. */ -export const userInitiatedCancelAtom = atom(false); -userInitiatedCancelAtom.debugLabel = "userInitiatedCancel"; +export const postStopDispatchSessionsAtom = atom< + Readonly> +>({}); +postStopDispatchSessionsAtom.debugLabel = "postStopDispatchSessions"; + +/** Open the post-Stop episode for exactly one session. */ +export const openPostStopDispatchEpisodeAtom = atom( + null, + (_get, set, sessionId: string) => { + set(postStopDispatchSessionsAtom, (current) => { + if (current[sessionId]) return current; + return { ...current, [sessionId]: true }; + }); + } +); +openPostStopDispatchEpisodeAtom.debugLabel = "openPostStopDispatchEpisode"; + +/** Close the post-Stop episode without affecting any other active session. */ +export const closePostStopDispatchEpisodeAtom = atom( + null, + (_get, set, sessionId: string) => { + set(postStopDispatchSessionsAtom, (current) => { + if (!current[sessionId]) return current; + const { [sessionId]: _removed, ...remaining } = current; + return remaining; + }); + } +); +closePostStopDispatchEpisodeAtom.debugLabel = "closePostStopDispatchEpisode"; /** * Pending "restore message to input box" signal. diff --git a/src/store/session/sessionAtom/__tests__/mutations.test.ts b/src/store/session/sessionAtom/__tests__/mutations.test.ts index c31e950d2..0874f942e 100644 --- a/src/store/session/sessionAtom/__tests__/mutations.test.ts +++ b/src/store/session/sessionAtom/__tests__/mutations.test.ts @@ -142,3 +142,30 @@ describe("updateSessionStatus", () => { expect(after).toEqual(before); }); }); + +describe("removeSession", () => { + it("drops the session and disposes its rust-agent streaming state", async () => { + const { upsertSession, sessionsAtom, store } = await loadModule(); + const mutations = await import("../mutations"); + const streamHelpers = + await import("@src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers"); + + upsertSession(makeSession({ session_id: "sess-x" })); + + // Seed per-turn streaming-stop state that only the deletion path can free. + streamHelpers.noteSessionStreamingTurn("sess-x", "turn-1"); + streamHelpers.markSessionStreamingStopped("sess-x"); + expect(streamHelpers.isSessionStreamingStopped("sess-x", "turn-1")).toBe( + true + ); + + mutations.removeSession("sess-x"); + + // The single store chokepoint covers every deletion path (sidebar, cloud, + // fork rollback, guest share) — the streaming state must be gone too. + expect(store.get(sessionsAtom)).toHaveLength(0); + expect(streamHelpers.isSessionStreamingStopped("sess-x", "turn-1")).toBe( + false + ); + }); +}); diff --git a/src/store/session/sessionAtom/mutations.ts b/src/store/session/sessionAtom/mutations.ts index 7be7aacd8..b992f6fd4 100644 --- a/src/store/session/sessionAtom/mutations.ts +++ b/src/store/session/sessionAtom/mutations.ts @@ -28,6 +28,7 @@ * intentional escape hatch for "the user just did something, bump * the row". */ +import { disposeSessionStreamingState } from "@src/engines/SessionCore/sync/adapters/rustAgent/eventHandlers/streamHelpers"; import { cursorIdeTurnSummariesAtomFamily } from "@src/store/session/cursorIdeTurnSummariesAtom"; import { tuiModeAtom } from "@src/store/session/tuiModeAtom"; import { clearTodosForSessionAtom } from "@src/store/ui/todoAtom"; @@ -138,6 +139,10 @@ export const removeSession = (sessionId: string) => { } store.set(clearTodosForSessionAtom, sessionId); removeGuestImportedSession(sessionId); + // Rust-agent streaming-stop state (per-turn stop markers etc.). This single + // chokepoint covers every removal path — sidebar delete, cloud remove, fork + // rollback, guest-share remove — so callers need not dispose it themselves. + disposeSessionStreamingState(sessionId); }; /** diff --git a/src/store/ui/messageQueueAtom.ts b/src/store/ui/messageQueueAtom.ts index 9d32bb59b..6fcd1511d 100644 --- a/src/store/ui/messageQueueAtom.ts +++ b/src/store/ui/messageQueueAtom.ts @@ -77,6 +77,10 @@ export interface QueuedMessage { export const messageQueueAtom = atom([]); messageQueueAtom.debugLabel = "messageQueueAtom"; +/** True once the durable queue snapshot has been merged into this Jotai store. */ +export const messageQueueHydratedAtom = atom(false); +messageQueueHydratedAtom.debugLabel = "messageQueueHydratedAtom"; + /** Tracks which queued message is currently being edited in the main input box. */ export interface QueueEditTarget { messageId: string; diff --git a/src/store/ui/messageQueueRepository.ts b/src/store/ui/messageQueueRepository.ts new file mode 100644 index 000000000..92ec05cd0 --- /dev/null +++ b/src/store/ui/messageQueueRepository.ts @@ -0,0 +1,101 @@ +import { type Store, load } from "@tauri-apps/plugin-store"; + +import { createLogger } from "@src/hooks/logger"; + +import type { QueuedMessage } from "./messageQueueAtom"; + +const log = createLogger("messageQueueRepository"); +const STORE_PATH = "chat-message-queue.json"; +const STORE_KEY_PREFIX = "queue"; + +let storePromise: Promise | null = null; +let queueKeyPromise: Promise | null = null; +let writeChain: Promise = Promise.resolve(); + +function isQueuedMessage(value: unknown): value is QueuedMessage { + if (!value || typeof value !== "object") return false; + const item = value as Partial; + return ( + typeof item.id === "string" && + typeof item.turnIntentId === "string" && + typeof item.sessionId === "string" && + typeof item.content === "string" && + typeof item.displayContent === "string" && + (item.priority === "now" || item.priority === "next") && + item.status === "queued" && + typeof item.createdAt === "string" + ); +} + +async function durableStore(): Promise { + if (storePromise) return storePromise; + storePromise = load(STORE_PATH, { + defaults: {}, + autoSave: false, + }).catch((error) => { + log.warn("[messageQueueRepository] durable store unavailable", error); + // Do not memoize a transient startup/plugin failure forever. Queue writes + // remain serialized, and the next mutation gets one fresh load attempt. + storePromise = null; + return null; + }); + return storePromise; +} + +async function queueKey(): Promise { + if (queueKeyPromise) return queueKeyPromise; + queueKeyPromise = import("@tauri-apps/api/window") + .then( + ({ getCurrentWindow }) => + `${STORE_KEY_PREFIX}:${getCurrentWindow().label}` + ) + .catch(() => `${STORE_KEY_PREFIX}:browser`); + return queueKeyPromise; +} + +/** Load this window's durable queue. Invalid rows are ignored, never dispatched. */ +export async function loadDurableMessageQueue(): Promise { + const store = await durableStore(); + if (!store) return []; + try { + const stored = await store.get(await queueKey()); + if (!Array.isArray(stored)) return []; + return stored.filter(isQueuedMessage); + } catch (error) { + log.warn("[messageQueueRepository] failed to load queue", error); + return []; + } +} + +/** + * Serialize writes so a rapid enqueue/reorder/dequeue burst cannot let an older + * async save overwrite a newer snapshot. Writes are explicitly saved before + * the mutation promise resolves, so app shutdown cannot race a deferred + * autosave after the in-memory queue has already changed. + */ +export function persistDurableMessageQueue( + messages: readonly QueuedMessage[] +): Promise { + const snapshot = messages.map((message) => ({ ...message })); + writeChain = writeChain + .catch((error) => { + // A transient failure must not poison the serialization chain. The next + // queue mutation gets a fresh save attempt with its complete snapshot. + log.warn("[messageQueueRepository] previous queue save failed", error); + }) + .then(async () => { + const store = await durableStore(); + if (!store) return; + await store.set(await queueKey(), snapshot); + await store.save(); + }); + return writeChain.catch((error) => { + log.warn("[messageQueueRepository] failed to persist queue", error); + }); +} + +export function resetMessageQueueRepositoryForTests(): void { + storePromise = null; + queueKeyPromise = null; + writeChain = Promise.resolve(); +}