🤖 refactor: populate softInterruptPending via stream-delta events

Change soft-interrupt communication from IPC request/response to
stream-delta event propagation. When user presses Escape the first time,
stopStream() emits an empty stream-delta with softInterruptPending=true,
which the frontend aggregator receives and updates StreamingContext.

Benefits:
- Semantically correct: stream-delta events signal state transitions
- Consistent with existing patterns: uses event flow like other updates
- Simpler frontend: no manual state setting needed
- Type-safe: compiler enforces optional field handling

Changes:
- streamManager.stopStream(): Emit stream-delta with flag on first Escape
- StreamDeltaEvent: Added optional softInterruptPending field
- StreamingMessageAggregator: Handle flag from stream-delta, skip empty deltas
- Removed unused setSoftInterruptPending() method
- Updated all canInterrupt references to use new interruptible field

Generated with `mux`

Author: ethanndickson

src/App.tsx

@@ -13,7 +13,7 @@ import { matchesKeybind, KEYBINDS } from "./utils/ui/keybinds";
 import { useResumeManager } from "./hooks/useResumeManager";
 import { useUnreadTracking } from "./hooks/useUnreadTracking";
 import { useAutoCompactContinue } from "./hooks/useAutoCompactContinue";
-import { useWorkspaceStoreRaw, useWorkspaceRecency } from "./stores/WorkspaceStore";
+import { useWorkspaceStoreRaw, useWorkspaceRecency, canInterrupt } from "./stores/WorkspaceStore";
 import { ChatInput } from "./components/ChatInput/index";
 
 import { useStableReference, compareMaps } from "./hooks/useStableReference";
@@ -490,7 +490,7 @@ function AppInner() {
       const allStates = workspaceStore.getAllStates();
       const streamingModels = new Map<string, string>();
       for (const [workspaceId, state] of allStates) {
-        if (state.canInterrupt && state.currentModel) {
+        if (canInterrupt(state.interruptType) && state.currentModel) {
           streamingModels.set(workspaceId, state.currentModel);
         }
       }

src/components/AIView.tsx

@@ -21,7 +21,7 @@ import { formatKeybind, KEYBINDS } from "@/utils/ui/keybinds";
 import { useAutoScroll } from "@/hooks/useAutoScroll";
 import { usePersistedState } from "@/hooks/usePersistedState";
 import { useThinking } from "@/contexts/ThinkingContext";
-import { useWorkspaceState, useWorkspaceAggregator } from "@/stores/WorkspaceStore";
+import { useWorkspaceState, useWorkspaceAggregator, canInterrupt } from "@/stores/WorkspaceStore";
 import { WorkspaceHeader } from "./WorkspaceHeader";
 import { getModelName } from "@/utils/ai/models";
 import type { DisplayedMessage } from "@/types/message";
@@ -220,15 +220,15 @@ const AIViewInner: React.FC<AIViewProps> = ({
   // Track if last message was interrupted or errored (for RetryBarrier)
   // Uses same logic as useResumeManager for DRY
   const showRetryBarrier = workspaceState
-    ? !workspaceState.canInterrupt &&
+    ? !canInterrupt(workspaceState.interruptType) &&
       hasInterruptedStream(workspaceState.messages, workspaceState.pendingStreamStartTime)
     : false;
 
   // Handle keyboard shortcuts (using optional refs that are safe even if not initialized)
   useAIViewKeybinds({
     workspaceId,
     currentModel: workspaceState?.currentModel ?? null,
-    canInterrupt: workspaceState?.canInterrupt ?? false,
+    canInterrupt: canInterrupt(workspaceState.interruptType),
     showRetryBarrier,
     currentWorkspaceThinking,
     setThinkingLevel,
@@ -278,7 +278,13 @@ const AIViewInner: React.FC<AIViewProps> = ({
   }
 
   // Extract state from workspace state
-  const { messages, canInterrupt, isCompacting, loading, currentModel } = workspaceState;
+  const {
+    messages,
+    interruptType: interruptible,
+    isCompacting,
+    loading,
+    currentModel,
+  } = workspaceState;
 
   // Get active stream message ID for token counting
   const activeStreamMessageId = aggregator.getActiveStreamMessageId();
@@ -290,6 +296,14 @@ const AIViewInner: React.FC<AIViewProps> = ({
   // Merge consecutive identical stream errors
   const mergedMessages = mergeConsecutiveStreamErrors(messages);
 
+  const model = currentModel ? getModelName(currentModel) : "";
+  const interrupting = interruptible === "hard";
+
+  const prefix = interrupting ? "⏸️ Interrupting " : "";
+  const action = interrupting ? "" : isCompacting ? "compacting..." : "streaming...";
+
+  const statusText = `${prefix}${model} ${action}`.trim();
+
   // When editing, find the cutoff point
   const editCutoffHistoryId = editingMessage
     ? mergedMessages.find(
@@ -362,8 +376,8 @@ const AIViewInner: React.FC<AIViewProps> = ({
             onTouchMove={markUserInteraction}
             onScroll={handleScroll}
             role="log"
-            aria-live={canInterrupt ? "polite" : "off"}
-            aria-busy={canInterrupt}
+            aria-live={canInterrupt(interruptible) ? "polite" : "off"}
+            aria-busy={canInterrupt(interruptible)}
             aria-label="Conversation transcript"
             tabIndex={0}
             className="h-full overflow-y-auto p-[15px] leading-[1.5] break-words whitespace-pre-wrap"
@@ -428,21 +442,13 @@ const AIViewInner: React.FC<AIViewProps> = ({
               </>
             )}
             <PinnedTodoList workspaceId={workspaceId} />
-            {canInterrupt && (
+            {canInterrupt(interruptible) && (
               <StreamingBarrier
-                statusText={
-                  isCompacting
-                    ? currentModel
-                      ? `${getModelName(currentModel)} compacting...`
-                      : "compacting..."
-                    : currentModel
-                      ? `${getModelName(currentModel)} streaming...`
-                      : "streaming..."
-                }
+                statusText={statusText}
                 cancelText={
                   isCompacting
                     ? `${formatKeybind(vimEnabled ? KEYBINDS.INTERRUPT_STREAM_VIM : KEYBINDS.INTERRUPT_STREAM_NORMAL)} cancel | ${formatKeybind(KEYBINDS.ACCEPT_EARLY_COMPACTION)} accept early`
-                    : `hit ${formatKeybind(vimEnabled ? KEYBINDS.INTERRUPT_STREAM_VIM : KEYBINDS.INTERRUPT_STREAM_NORMAL)} to cancel`
+                    : `hit ${formatKeybind(vimEnabled ? KEYBINDS.INTERRUPT_STREAM_VIM : KEYBINDS.INTERRUPT_STREAM_NORMAL)} to ${interruptible === "hard" ? "force" : ""} cancel`
                 }
                 tokenCount={
                   activeStreamMessageId
@@ -499,7 +505,7 @@ const AIViewInner: React.FC<AIViewProps> = ({
           editingMessage={editingMessage}
           onCancelEdit={handleCancelEdit}
           onEditLastUserMessage={() => void handleEditLastUserMessage()}
-          canInterrupt={canInterrupt}
+          canInterrupt={canInterrupt(interruptible)}
           onReady={handleChatInputReady}
         />
       </div>

src/components/AgentStatusIndicator.tsx

@@ -1,7 +1,7 @@
 import React, { useCallback, useMemo } from "react";
 import { cn } from "@/lib/utils";
 import { TooltipWrapper, Tooltip } from "./Tooltip";
-import { useWorkspaceSidebarState } from "@/stores/WorkspaceStore";
+import { canInterrupt, useWorkspaceSidebarState } from "@/stores/WorkspaceStore";
 import { getStatusTooltip } from "@/utils/ui/statusTooltip";
 
 interface AgentStatusIndicatorProps {
@@ -22,10 +22,10 @@ export const AgentStatusIndicator: React.FC<AgentStatusIndicatorProps> = ({
   className,
 }) => {
   // Get workspace state
-  const { canInterrupt, currentModel, agentStatus, recencyTimestamp } =
+  const { interruptType, currentModel, agentStatus, recencyTimestamp } =
     useWorkspaceSidebarState(workspaceId);
 
-  const streaming = canInterrupt;
+  const streaming = canInterrupt(interruptType);
 
   // Compute unread status if lastReadTimestamp provided (sidebar only)
   const unread = useMemo(() => {

src/components/ChatInput/index.tsx

@@ -315,7 +315,11 @@ export const ChatInput: React.FC<ChatInputProps> = (props) => {
   // Allow external components (e.g., CommandPalette, Queued message edits) to insert text
   useEffect(() => {
     const handler = (e: Event) => {
-      const customEvent = e as CustomEvent<{ text: string; mode?: "append" | "replace"; imageParts?: ImagePart[] }>;
+      const customEvent = e as CustomEvent<{
+        text: string;
+        mode?: "append" | "replace";
+        imageParts?: ImagePart[];
+      }>;
 
       const { text, mode = "append", imageParts } = customEvent.detail;
 

src/components/Messages/MessageRenderer.tsx

@@ -7,7 +7,6 @@ import { ReasoningMessage } from "./ReasoningMessage";
 import { StreamErrorMessage } from "./StreamErrorMessage";
 import { HistoryHiddenMessage } from "./HistoryHiddenMessage";
 import { InitMessage } from "./InitMessage";
-// Note: QueuedMessage is NOT imported here - it's rendered directly in AIView after StreamingBarrier
 
 interface MessageRendererProps {
   message: DisplayedMessage;

src/hooks/useResumeManager.ts

@@ -1,5 +1,5 @@
 import { useEffect, useRef } from "react";
-import { useWorkspaceStoreRaw, type WorkspaceState } from "@/stores/WorkspaceStore";
+import { canInterrupt, useWorkspaceStoreRaw, type WorkspaceState } from "@/stores/WorkspaceStore";
 import { CUSTOM_EVENTS, type CustomEventType } from "@/constants/events";
 import { getAutoRetryKey, getRetryStateKey } from "@/constants/storage";
 import { getSendOptionsFromStorage } from "@/utils/messages/sendOptions";
@@ -97,7 +97,7 @@ export function useResumeManager() {
     }
 
     // 1. Must have interrupted stream that's eligible for auto-retry (not currently streaming)
-    if (state.canInterrupt) return false; // Currently streaming
+    if (canInterrupt(state.interruptType)) return false; // Currently streaming
 
     if (!isEligibleForAutoRetry(state.messages, state.pendingStreamStartTime)) {
       return false;

src/services/streamManager.ts

@@ -106,6 +106,8 @@ interface WorkspaceStreamInfo {
   partialWritePromise?: Promise<void>;
   // Track background processing promise for guaranteed cleanup
   processingPromise: Promise<void>;
+  // Flag for soft-interrupt: when true, stream will end at next block boundary
+  softInterruptPending: boolean;
   // Temporary directory for tool outputs (auto-cleaned when stream ends)
   runtimeTempDir: string;
   // Runtime for temp directory cleanup
@@ -414,31 +416,63 @@ export class StreamManager extends EventEmitter {
 
       streamInfo.abortController.abort();
 
-      // CRITICAL: Wait for processing to fully complete before cleanup
-      // This prevents race conditions where the old stream is still running
-      // while a new stream starts (e.g., old stream writing to partial.json)
-      await streamInfo.processingPromise;
+      await this.cleanupStream(workspaceId, streamInfo);
+    } catch (error) {
+      console.error("Error during stream cancellation:", error);
+      // Force cleanup even if cancellation fails
+      this.workspaceStreams.delete(workspaceId);
+    }
+  }
+
+  // Checks if a soft interrupt is necessary, and performs one if so
+  // Similar to cancelStreamSafely but performs cleanup without blocking
+  private async checkSoftCancelStream(
+    workspaceId: WorkspaceId,
+    streamInfo: WorkspaceStreamInfo
+  ): Promise<void> {
+    if (!streamInfo.softInterruptPending) return;
+    try {
+      streamInfo.state = StreamState.STOPPING;
 
-      // Get usage and duration metadata (usage may be undefined if aborted early)
-      const { usage, duration } = await this.getStreamMetadata(streamInfo);
+      // Flush any pending partial write immediately (preserves work on interruption)
+      await this.flushPartialWrite(workspaceId, streamInfo);
 
-      // Emit abort event with usage if available
-      this.emit("stream-abort", {
-        type: "stream-abort",
-        workspaceId: workspaceId as string,
-        messageId: streamInfo.messageId,
-        metadata: { usage, duration },
-      });
+      streamInfo.abortController.abort();
 
-      // Clean up immediately
-      this.workspaceStreams.delete(workspaceId);
+      // Return back to the stream loop so we can wait for it to finish before
+      // sending the stream abort event.
+      void this.cleanupStream(workspaceId, streamInfo);
     } catch (error) {
       console.error("Error during stream cancellation:", error);
       // Force cleanup even if cancellation fails
       this.workspaceStreams.delete(workspaceId);
     }
   }
 
+  private async cleanupStream(
+    workspaceId: WorkspaceId,
+    streamInfo: WorkspaceStreamInfo
+  ): Promise<void> {
+    // CRITICAL: Wait for processing to fully complete before cleanup
+    // This prevents race conditions where the old stream is still running
+    // while a new stream starts (e.g., old stream writing to partial.json)
+    await streamInfo.processingPromise;
+
+    // Get usage and duration metadata (usage may be undefined if aborted early)
+    const { usage, duration } = await this.getStreamMetadata(streamInfo);
+
+    // Emit abort event with usage if available
+    this.emit("stream-abort", {
+      type: "stream-abort",
+      workspaceId: workspaceId as string,
+      messageId: streamInfo.messageId,
+      metadata: { usage, duration },
+    });
+
+    // Clean up immediately
+    this.workspaceStreams.delete(workspaceId);
+  }
+
   /**
    * Atomically creates a new stream with all necessary setup
    */
@@ -525,6 +559,7 @@ export class StreamManager extends EventEmitter {
       lastPartialWriteTime: 0, // Initialize to 0 to allow immediate first write
       partialWritePromise: undefined, // No write in flight initially
       processingPromise: Promise.resolve(), // Placeholder, overwritten in startStream
+      softInterruptPending: false, // Initialize to false
       runtimeTempDir, // Stream-scoped temp directory for tool outputs
       runtime, // Runtime for temp directory cleanup
     };
@@ -688,6 +723,7 @@ export class StreamManager extends EventEmitter {
               workspaceId: workspaceId as string,
               messageId: streamInfo.messageId,
             });
+            await this.checkSoftCancelStream(workspaceId, streamInfo);
             break;
           }
 
@@ -742,6 +778,7 @@ export class StreamManager extends EventEmitter {
                 strippedOutput
               );
             }
+            await this.checkSoftCancelStream(workspaceId, streamInfo);
             break;
           }
 
@@ -778,6 +815,7 @@ export class StreamManager extends EventEmitter {
               toolErrorPart.toolName,
               errorOutput
             );
+            await this.checkSoftCancelStream(workspaceId, streamInfo);
             break;
           }
 
@@ -823,9 +861,14 @@ export class StreamManager extends EventEmitter {
           case "start-step":
           case "text-start":
           case "finish":
-          case "finish-step":
             // These events can be logged or handled if needed
             break;
+
+          case "finish-step":
+          case "text-end":
+          case "tool-input-end":
+            await this.checkSoftCancelStream(workspaceId, streamInfo);
+            break;
         }
       }
 
@@ -1196,14 +1239,32 @@ export class StreamManager extends EventEmitter {
 
   /**
    * Stops an active stream for a workspace
+   * First call: Sets soft interrupt and emits delta event → frontend shows "Interrupting..."
+   * Second call: Hard aborts the stream immediately
    */
   async stopStream(workspaceId: string): Promise<Result<void>> {
     const typedWorkspaceId = workspaceId as WorkspaceId;
 
     try {
       const streamInfo = this.workspaceStreams.get(typedWorkspaceId);
-      if (streamInfo) {
+      if (!streamInfo) {
+        return Ok(undefined); // No active stream
+      }
+
+      if (streamInfo.softInterruptPending) {
         await this.cancelStreamSafely(typedWorkspaceId, streamInfo);
+      } else {
+        // First Escape: Soft interrupt - emit delta to notify frontend
+        streamInfo.softInterruptPending = true;
+        this.emit("stream-delta", {
+          type: "stream-delta",
+          workspaceId: workspaceId,
+          messageId: streamInfo.messageId,
+          delta: "",
+          tokens: 0,
+          timestamp: Date.now(),
+          softInterruptPending: true, // Signal to frontend
+        });
       }
       return Ok(undefined);
     } catch (error) {

src/stores/WorkspaceStore.test.ts

@@ -273,7 +273,7 @@ describe("WorkspaceStore", () => {
       // Object.is() comparison and skip re-renders for primitive values.
       // TODO: Optimize aggregator caching in Phase 2
       expect(state1).toEqual(state2);
-      expect(state1.canInterrupt).toBe(state2.canInterrupt);
+      expect(state1.interruptType).toBe(state2.interruptType);
       expect(state1.loading).toBe(state2.loading);
     });
   });
@@ -428,7 +428,7 @@ describe("WorkspaceStore", () => {
 
       const state2 = store.getWorkspaceState("test-workspace");
       expect(state1).not.toBe(state2); // Cache should be invalidated
-      expect(state2.canInterrupt).toBe(true); // Stream started, so can interrupt
+      expect(state2.interruptType).toBeTruthy(); // Stream started, so can interrupt
     });
 
     it("invalidates getAllStates() cache when workspace changes", async () => {