🤖 Add integration tests for stream error recovery (no amnesia) (#333)

## Summary

Adds integration test to verify that stream error recovery preserves
context (no amnesia bug).

## Changes

- **Debug IPC for testing**: Added `DEBUG_TRIGGER_STREAM_ERROR` IPC
channel
- **StreamManager debug method**: `debugTriggerStreamError()` triggers
artificial stream errors that follow the same code path as real errors
- **Integration test**: Single error + resume scenario verifies context
preservation via **structured markers**

## Test Design

**Structured-marker approach** for precise validation:

**Test Flow:**
1. Generate unique nonce for test run (random 10-char identifier)
2. Model counts 1-100 using structured format: `${nonce}-<n>: <word>`
(e.g. `ai7qcnc20g-1: one`)
3. Collect stream deltas until ≥10 complete markers detected
4. Trigger artificial network error mid-stream
5. Resume stream and wait for completion
6. Verify final message has **both** properties:
- **(a) Prefix preservation**: Starts with exact pre-error streamed text
- **(b) Exact continuation**: Contains next sequential marker
(${nonce}-11) shortly after prefix

**Validation:**
- Pre-error content captured from stream-delta events (user-visible data
path)
- Stable prefix truncated to last complete marker line (no partial
markers)
- Assertions directly prove both amnesia-prevention properties
- No coupling to internal storage formats or metadata

**Why this approach:**
- **Precise**: Detects exact continuation (not just "some work done")
- **Unambiguous**: Random nonce makes false positives virtually
impossible
- **Robust**: Structured format less likely to confuse model than
natural language
- **Fast**: Haiku 4.5 completes in ~18-21 seconds

## Bug Fix

Also fixed event collection bug in `collectStreamUntil`: properly track
consumed deltas to avoid returning the same event multiple times.
Previous logic returned first matching event on every poll, causing
duplicate processing.

## Related

Follow-up to #331 which fixed the amnesia bug by preserving accumulated
parts on error.

## Test Results

✅ Test passes reliably in ~18-21 seconds
✅ Validates **exact** prefix preservation and continuation
✅ No flaky failures from timing issues
✅ Integration tests pass: 1 passed, 1 total

_Generated with `cmux`_

Author: ammar-agent

src/constants/ipc-constants.ts

@@ -39,6 +39,9 @@ export const IPC_CHANNELS = {
   // Window channels
   WINDOW_SET_TITLE: "window:setTitle",
 
+  // Debug channels (for testing only)
+  DEBUG_TRIGGER_STREAM_ERROR: "debug:triggerStreamError",
+
   // Dynamic channel prefixes
   WORKSPACE_CHAT_PREFIX: "workspace:chat:",
   WORKSPACE_METADATA: "workspace:metadata",

src/services/ipcMain.ts

@@ -855,6 +855,25 @@ export class IpcMain {
         log.error(`Failed to open terminal: ${message}`);
       }
     });
+
+    // Debug IPC - only for testing
+    ipcMain.handle(
+      IPC_CHANNELS.DEBUG_TRIGGER_STREAM_ERROR,
+      (_event, workspaceId: string, errorMessage: string) => {
+        try {
+          // eslint-disable-next-line @typescript-eslint/dot-notation -- accessing private member for testing
+          const triggered = this.aiService["streamManager"].debugTriggerStreamError(
+            workspaceId,
+            errorMessage
+          );
+          return { success: triggered };
+        } catch (error) {
+          const message = error instanceof Error ? error.message : String(error);
+          log.error(`Failed to trigger stream error: ${message}`);
+          return { success: false, error: message };
+        }
+      }
+    );
   }
 
   /**

src/services/streamManager.ts

@@ -1271,4 +1271,62 @@ export class StreamManager extends EventEmitter {
       this.emitPartAsEvent(typedWorkspaceId, streamInfo.messageId, part);
     }
   }
+
+  /**
+   * DEBUG ONLY: Trigger an artificial stream error for testing
+   * This method allows integration tests to simulate stream errors without
+   * mocking the AI SDK or network layer. It triggers the same error handling
+   * path as genuine stream errors by aborting the stream and manually triggering
+   * the error event (since abort alone doesn't throw, it just sets a flag).
+   */
+  debugTriggerStreamError(workspaceId: string, errorMessage: string): boolean {
+    const typedWorkspaceId = workspaceId as WorkspaceId;
+    const streamInfo = this.workspaceStreams.get(typedWorkspaceId);
+
+    // Only trigger error if stream is actively running
+    if (
+      !streamInfo ||
+      (streamInfo.state !== StreamState.STARTING && streamInfo.state !== StreamState.STREAMING)
+    ) {
+      return false;
+    }
+
+    // Abort the stream first
+    streamInfo.abortController.abort(new Error(errorMessage));
+
+    // Update streamInfo metadata with error (so subsequent flushes preserve it)
+    streamInfo.initialMetadata = {
+      ...streamInfo.initialMetadata,
+      error: errorMessage,
+      errorType: "network",
+    };
+
+    // Write error state to partial.json (same as real error handling)
+    const errorPartialMessage: CmuxMessage = {
+      id: streamInfo.messageId,
+      role: "assistant",
+      metadata: {
+        historySequence: streamInfo.historySequence,
+        timestamp: streamInfo.startTime,
+        model: streamInfo.model,
+        partial: true,
+        error: errorMessage,
+        errorType: "network", // Test errors are network-like
+        ...streamInfo.initialMetadata,
+      },
+      parts: streamInfo.parts,
+    };
+    void this.partialService.writePartial(workspaceId, errorPartialMessage);
+
+    // Emit error event (same as real error handling)
+    this.emit("error", {
+      type: "error",
+      workspaceId,
+      messageId: streamInfo.messageId,
+      error: errorMessage,
+      errorType: "network",
+    } as ErrorEvent);
+
+    return true;
+  }
 }