abort observed marker

Author: icecrasher321

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

@@ -27,7 +27,7 @@ import {
   extractEditContent,
   runStreamLoop,
 } from '@/lib/copilot/request/go/stream'
-import { createEvent, hasAbortMarker } from '@/lib/copilot/request/session'
+import { AbortReason, createEvent, hasAbortMarker } from '@/lib/copilot/request/session'
 import { RequestTraceV1Outcome, TraceCollector } from '@/lib/copilot/request/trace'
 import type { ExecutionContext, StreamingContext } from '@/lib/copilot/request/types'
 
@@ -331,6 +331,72 @@ describe('copilot go stream helpers', () => {
     ).toBe(false)
   })
 
+  it('invokes onAbortObserved with MarkerObservedAtBodyClose when reclassifying via the abort marker', 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',
+    }
+    const onAbortObserved = vi.fn()
+
+    await runStreamLoop('https://example.com/mothership/stream', {}, context, execContext, {
+      timeout: 1000,
+      onAbortObserved,
+    })
+
+    expect(onAbortObserved).toHaveBeenCalledTimes(1)
+    expect(onAbortObserved).toHaveBeenCalledWith(AbortReason.MarkerObservedAtBodyClose)
+    expect(context.wasAborted).toBe(true)
+  })
+
+  it('does not invoke onAbortObserved when no abort marker is present at body close', 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(false)
+
+    const context = createStreamingContext()
+    const execContext: ExecutionContext = {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+    }
+    const onAbortObserved = vi.fn()
+
+    await expect(
+      runStreamLoop('https://example.com/mothership/stream', {}, context, execContext, {
+        timeout: 1000,
+        onAbortObserved,
+      })
+    ).rejects.toThrow('Copilot backend stream ended before a terminal event')
+
+    expect(onAbortObserved).not.toHaveBeenCalled()
+  })
+
   it('still fails closed when the body closes without terminal and the abort marker check throws', async () => {
     const textEvent = createEvent({
       streamId: 'stream-1',

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

@@ -30,6 +30,7 @@ import {
 } from '@/lib/copilot/request/handlers/types'
 import { getCopilotTracer } from '@/lib/copilot/request/otel'
 import {
+  AbortReason,
   eventToStreamEvent,
   hasAbortMarker,
   isSubagentSpanStreamEvent,
@@ -448,6 +449,7 @@ export async function runStreamLoop(
       }
 
       if (abortRequested) {
+        options.onAbortObserved?.(AbortReason.MarkerObservedAtBodyClose)
         context.wasAborted = true
         endedOn = CopilotSseCloseReason.Aborted
       } else {

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

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

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

@@ -249,6 +249,11 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
               onEvent: async (event) => {
                 await publisher.publish(event)
               },
+              onAbortObserved: (reason) => {
+                if (!abortController.signal.aborted) {
+                  abortController.abort(reason)
+                }
+              },
             })
 
             lifecycleResult = result

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

@@ -22,6 +22,12 @@ export const AbortReason = {
    * that the node that DID receive it wrote, and aborts on the poll.
    */
   RedisPoller: 'redis_abort_marker:poller',
+  /**
+   * Cross-process stop: same root cause as `RedisPoller`, but observed
+   * by `runStreamLoop` at body close (the Go body ended before the
+   * 250ms poller's next tick) rather than by the polling timer.
+   */
+  MarkerObservedAtBodyClose: 'redis_abort_marker:body_close',
   /** Internal timeout on the outbound explicit-abort fetch to Go. */
   ExplicitAbortFetchTimeout: 'timeout:go_explicit_abort_fetch',
 } as const
@@ -38,5 +44,9 @@ export type AbortReasonValue = (typeof AbortReason)[keyof typeof AbortReason]
  * stops, mirroring `requestctx.IsExplicitUserStop` on the Go side.
  */
 export function isExplicitStopReason(reason: unknown): boolean {
-  return reason === AbortReason.UserStop || reason === AbortReason.RedisPoller
+  return (
+    reason === AbortReason.UserStop ||
+    reason === AbortReason.RedisPoller ||
+    reason === AbortReason.MarkerObservedAtBodyClose
+  )
 }

apps/sim/lib/copilot/request/types.ts

@@ -136,6 +136,14 @@ export interface OrchestratorOptions {
   onComplete?: (result: OrchestratorResult) => void | Promise<void>
   onError?: (error: Error) => void | Promise<void>
   abortSignal?: AbortSignal
+  /**
+   * Invoked when the orchestrator infers that the run was aborted via
+   * an out-of-band signal (currently: a Redis abort marker observed
+   * at SSE body close). Callers wire this to fire their local
+   * `AbortController` so `signal.reason` is set and `recordCancelled`
+   * classifies as `explicit_stop` rather than `unknown`.
+   */
+  onAbortObserved?: (reason: string) => void
   interactive?: boolean
 }