fix: properly stop streams on ESC interrupt (#457)

Co-authored-by: Letta <noreply@letta.com>

src/cli/helpers/accumulator.ts

@@ -73,6 +73,7 @@ export type Buffers = {
   pendingRefresh?: boolean; // Track throttled refresh state
   interrupted?: boolean; // Track if stream was interrupted by user (skip stale refreshes)
   commitGeneration?: number; // Incremented when resuming from error to invalidate pending refreshes
+  abortGeneration?: number; // Incremented on each interrupt to detect cancellation across async boundaries
   usage: {
     promptTokens: number;
     completionTokens: number;
@@ -92,6 +93,7 @@ export function createBuffers(): Buffers {
     toolCallIdToLineId: new Map(),
     lastOtid: null,
     commitGeneration: 0,
+    abortGeneration: 0,
     usage: {
       promptTokens: 0,
       completionTokens: 0,

src/cli/helpers/stream.ts

@@ -56,12 +56,22 @@ export async function drainStream(
   // Track if we triggered abort via our listener (for eager cancellation)
   let abortedViaListener = false;
 
+  // Capture the abort generation at stream start to detect if handleInterrupt ran
+  const startAbortGen = buffers.abortGeneration || 0;
+
   // Set up abort listener to propagate our signal to SDK's stream controller
   // This immediately cancels the HTTP request instead of waiting for next chunk
   const abortHandler = () => {
     abortedViaListener = true;
     // Abort the SDK's stream controller to cancel the underlying HTTP request
-    if (stream.controller && !stream.controller.signal.aborted) {
+    if (!stream.controller) {
+      debugWarn(
+        "drainStream",
+        "stream.controller is undefined - cannot abort HTTP request",
+      );
+      return;
+    }
+    if (!stream.controller.signal.aborted) {
       stream.controller.abort();
     }
   };
@@ -80,6 +90,15 @@ export async function drainStream(
     for await (const chunk of stream) {
       // console.log("chunk", chunk);
 
+      // Check if abort generation changed (handleInterrupt ran while we were waiting)
+      // This catches cases where the abort signal might not propagate correctly
+      if ((buffers.abortGeneration || 0) !== startAbortGen) {
+        stopReason = "cancelled";
+        // Don't call markIncompleteToolsAsCancelled - handleInterrupt already did
+        queueMicrotask(refresh);
+        break;
+      }
+
       // Check if stream was aborted
       if (abortSignal?.aborted) {
         stopReason = "cancelled";
@@ -321,11 +340,14 @@ export async function drainStreamWithResume(
   );
 
   // If stream ended without proper stop_reason and we have resume info, try once to reconnect
+  // Only resume if we have an abortSignal AND it's not aborted (explicit check prevents
+  // undefined abortSignal from accidentally allowing resume after user cancellation)
   if (
     result.stopReason === "error" &&
     result.lastRunId &&
     result.lastSeqId !== null &&
-    !abortSignal?.aborted
+    abortSignal &&
+    !abortSignal.aborted
   ) {
     // Preserve the original error in case resume fails
     const originalFallbackError = result.fallbackError;