fix(mothership): queue supersede crash

Author: icecrasher321

apps/sim/lib/copilot/request/go/stream.test.ts

@@ -10,13 +10,24 @@ import {
   MothershipStreamV1ToolOutcome,
   MothershipStreamV1ToolPhase,
 } from '@/lib/copilot/generated/mothership-stream-v1'
+
+vi.mock('@/lib/copilot/request/session', async () => {
+  const actual = await vi.importActual<typeof import('@/lib/copilot/request/session')>(
+    '@/lib/copilot/request/session'
+  )
+  return {
+    ...actual,
+    hasAbortMarker: vi.fn().mockResolvedValue(false),
+  }
+})
+
 import {
   buildPreviewContentUpdate,
   decodeJsonStringPrefix,
   extractEditContent,
   runStreamLoop,
 } from '@/lib/copilot/request/go/stream'
-import { createEvent } from '@/lib/copilot/request/session'
+import { createEvent, hasAbortMarker } from '@/lib/copilot/request/session'
 import { RequestTraceV1Outcome, TraceCollector } from '@/lib/copilot/request/trace'
 import type { ExecutionContext, StreamingContext } from '@/lib/copilot/request/types'
 
@@ -285,6 +296,71 @@ describe('copilot go stream helpers', () => {
     ).toBe(true)
   })
 
+  it('reclassifies as aborted when the body closes without terminal but the abort marker is set', async () => {
+    const textEvent = createEvent({
+      streamId: 'stream-1',
+      cursor: '1',
+      seq: 1,
+      requestId: 'req-1',
+      type: MothershipStreamV1EventType.text,
+      payload: {
+        channel: 'assistant',
+        text: 'partial response',
+      },
+    })
+
+    vi.mocked(fetch).mockResolvedValueOnce(createSseResponse([textEvent]))
+    vi.mocked(hasAbortMarker).mockResolvedValueOnce(true)
+
+    const context = createStreamingContext()
+    const execContext: ExecutionContext = {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+    }
+
+    await runStreamLoop('https://example.com/mothership/stream', {}, context, execContext, {
+      timeout: 1000,
+    })
+
+    expect(hasAbortMarker).toHaveBeenCalledWith(context.messageId)
+    expect(context.wasAborted).toBe(true)
+    expect(
+      context.errors.some((message) =>
+        message.includes('Copilot backend stream ended before a terminal event')
+      )
+    ).toBe(false)
+  })
+
+  it('still fails closed when the body closes without terminal and the abort marker check throws', async () => {
+    const textEvent = createEvent({
+      streamId: 'stream-1',
+      cursor: '1',
+      seq: 1,
+      requestId: 'req-1',
+      type: MothershipStreamV1EventType.text,
+      payload: {
+        channel: 'assistant',
+        text: 'partial response',
+      },
+    })
+
+    vi.mocked(fetch).mockResolvedValueOnce(createSseResponse([textEvent]))
+    vi.mocked(hasAbortMarker).mockRejectedValueOnce(new Error('redis unavailable'))
+
+    const context = createStreamingContext()
+    const execContext: ExecutionContext = {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+    }
+
+    await expect(
+      runStreamLoop('https://example.com/mothership/stream', {}, context, execContext, {
+        timeout: 1000,
+      })
+    ).rejects.toThrow('Copilot backend stream ended before a terminal event')
+    expect(context.wasAborted).toBe(false)
+  })
+
   it('fails closed when the shared stream receives an invalid event', async () => {
     vi.mocked(fetch).mockResolvedValueOnce(
       createSseResponse([

apps/sim/lib/copilot/request/go/stream.ts

@@ -31,6 +31,7 @@ import {
 import { getCopilotTracer } from '@/lib/copilot/request/otel'
 import {
   eventToStreamEvent,
+  hasAbortMarker,
   isSubagentSpanStreamEvent,
   parsePersistedStreamEventEnvelope,
 } from '@/lib/copilot/request/session'
@@ -436,16 +437,31 @@ export async function runStreamLoop(
     })
 
     if (!context.streamComplete && !abortSignal?.aborted && !context.wasAborted) {
-      const streamPath = new URL(fetchUrl).pathname
-      const message = `Copilot backend stream ended before a terminal event on ${streamPath}`
-      context.errors.push(message)
-      logger.error('Copilot backend stream ended before a terminal event', {
-        path: streamPath,
-        requestId: context.requestId,
-        messageId: context.messageId,
-      })
-      endedOn = CopilotSseCloseReason.ClosedNoTerminal
-      throw new CopilotBackendError(message, { status: 503 })
+      let abortRequested = false
+      try {
+        abortRequested = await hasAbortMarker(context.messageId)
+      } catch (error) {
+        logger.warn('Failed to read abort marker at body close', {
+          streamId: context.messageId,
+          error: error instanceof Error ? error.message : String(error),
+        })
+      }
+
+      if (abortRequested) {
+        context.wasAborted = true
+        endedOn = CopilotSseCloseReason.Aborted
+      } else {
+        const streamPath = new URL(fetchUrl).pathname
+        const message = `Copilot backend stream ended before a terminal event on ${streamPath}`
+        context.errors.push(message)
+        logger.error('Copilot backend stream ended before a terminal event', {
+          path: streamPath,
+          requestId: context.requestId,
+          messageId: context.messageId,
+        })
+        endedOn = CopilotSseCloseReason.ClosedNoTerminal
+        throw new CopilotBackendError(message, { status: 503 })
+      }
     }
   } catch (error) {
     if (error instanceof FatalSseEventError && !context.errors.includes(error.message)) {

apps/sim/lib/copilot/request/lifecycle/headless.ts

@@ -53,11 +53,12 @@ export async function runHeadlessCopilotLifecycle(
           simRequestId,
           otelContext,
         })
-        outcome = options.abortSignal?.aborted
-          ? RequestTraceV1Outcome.cancelled
-          : result.success
-            ? RequestTraceV1Outcome.success
-            : RequestTraceV1Outcome.error
+        outcome =
+          options.abortSignal?.aborted || result.cancelled
+            ? RequestTraceV1Outcome.cancelled
+            : result.success
+              ? RequestTraceV1Outcome.success
+              : RequestTraceV1Outcome.error
         return result
       } catch (error) {
         outcome = options.abortSignal?.aborted

apps/sim/lib/copilot/request/lifecycle/start.test.ts

@@ -6,7 +6,10 @@ import { propagation, trace } from '@opentelemetry/api'
 import { W3CTraceContextPropagator } from '@opentelemetry/core'
 import { BasicTracerProvider } from '@opentelemetry/sdk-trace-base'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
-import { MothershipStreamV1EventType } from '@/lib/copilot/generated/mothership-stream-v1'
+import {
+  MothershipStreamV1CompletionStatus,
+  MothershipStreamV1EventType,
+} from '@/lib/copilot/generated/mothership-stream-v1'
 
 const {
   runCopilotLifecycle,
@@ -60,6 +63,7 @@ vi.mock('@/lib/copilot/request/session', () => ({
   registerActiveStream: vi.fn(),
   unregisterActiveStream: vi.fn(),
   startAbortPoller: vi.fn().mockReturnValue(setInterval(() => {}, 999999)),
+  isExplicitStopReason: vi.fn().mockReturnValue(false),
   SSE_RESPONSE_HEADERS: {},
   StreamWriter: vi.fn().mockImplementation(() => ({
     attach: vi.fn().mockImplementation((ctrl: ReadableStreamDefaultController) => {
@@ -211,6 +215,46 @@ describe('createSSEStream terminal error handling', () => {
     expect(scheduleBufferCleanup).toHaveBeenCalledWith('stream-1')
   })
 
+  it('publishes a cancelled completion (not an error) when the orchestrator reports cancelled without abortSignal aborted', async () => {
+    runCopilotLifecycle.mockResolvedValue({
+      success: false,
+      cancelled: true,
+      content: '',
+      contentBlocks: [],
+      toolCalls: [],
+    })
+
+    const stream = createSSEStream({
+      requestPayload: { message: 'hello' },
+      userId: 'user-1',
+      streamId: 'stream-1',
+      executionId: 'exec-1',
+      runId: 'run-1',
+      currentChat: null,
+      isNewChat: false,
+      message: 'hello',
+      titleModel: 'gpt-5.4',
+      requestId: 'req-cancelled',
+      orchestrateOptions: {},
+    })
+
+    await drainStream(stream)
+
+    expect(appendEvent).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: MothershipStreamV1EventType.error,
+      })
+    )
+    expect(appendEvent).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: MothershipStreamV1EventType.complete,
+        payload: expect.objectContaining({
+          status: MothershipStreamV1CompletionStatus.cancelled,
+        }),
+      })
+    )
+  })
+
   it('passes an OTel context into the streaming lifecycle', async () => {
     let lifecycleTraceparent = ''
     runCopilotLifecycle.mockImplementation(async (_payload, options) => {

apps/sim/lib/copilot/request/lifecycle/start.ts

@@ -266,7 +266,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
             //   3. Otherwise → error.
             outcome = result.success
               ? RequestTraceV1Outcome.success
-              : abortController.signal.aborted || publisher.clientDisconnected
+              : result.cancelled || abortController.signal.aborted || publisher.clientDisconnected
                 ? RequestTraceV1Outcome.cancelled
                 : RequestTraceV1Outcome.error
             if (outcome === RequestTraceV1Outcome.cancelled) {

apps/sim/lib/copilot/request/session/abort.ts

@@ -17,7 +17,7 @@ const pendingChatStreams = new Map<
   { promise: Promise<void>; resolve: () => void; streamId: string }
 >()
 
-const DEFAULT_ABORT_POLL_MS = 1000
+const DEFAULT_ABORT_POLL_MS = 250
 
 /**
  * TTL for the per-chat stream lock. Kept short so that if the Sim pod