fix: #1190 reconcile streamed function calls when server-managed runs abort

Author: seratch

.changeset/stream-abort-reconciliation.md

@@ -0,0 +1,5 @@
+---
+'@openai/agents-core': patch
+---
+
+fix: #1190 reconcile streamed function calls when server-managed runs abort

packages/agents-core/src/run.ts

@@ -93,6 +93,14 @@ import {
   prepareSandboxInterruptedTurnResume,
   type SandboxMemoryPersistenceContext,
 } from './runner/sandbox';
+import {
+  buildAbortReconciliationInput,
+  createStreamAbortReconciliationState,
+  getAbortReconciliationPreviousResponseId,
+  markAbortReconciliationComplete,
+  recordStreamEventForAbortReconciliation,
+  shouldReconcileStreamAbort,
+} from './runner/streamReconciliation';
 
 export type {
   CallModelInputFilter,
@@ -1247,6 +1255,8 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
           guardrailTracker.throwIfError();
 
           let finalResponse: ModelResponse | undefined = undefined;
+          const abortReconciliationState =
+            createStreamAbortReconciliationState();
           let inputMarked = false;
           const markInputOnce = () => {
             if (inputMarked || !serverConversationTracker) {
@@ -1266,6 +1276,63 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
             );
             inputMarked = true;
           };
+          const reconcileStreamAbortIfNeeded = async () => {
+            if (
+              !serverConversationTracker ||
+              !shouldReconcileStreamAbort(abortReconciliationState)
+            ) {
+              return;
+            }
+
+            const reconciliationInput = buildAbortReconciliationInput(
+              abortReconciliationState,
+            );
+            try {
+              const reconciliationResponse = await getResponseWithRetry(
+                preparedCall.model,
+                {
+                  systemInstructions: preparedCall.modelInput.instructions,
+                  prompt: preparedCall.prompt,
+                  ...(preparedCall.explictlyModelSet
+                    ? { overridePromptModel: true }
+                    : {}),
+                  input: reconciliationInput,
+                  previousResponseId: getAbortReconciliationPreviousResponseId(
+                    abortReconciliationState,
+                    preparedCall,
+                  ),
+                  conversationId: preparedCall.conversationId,
+                  modelSettings: preparedCall.modelSettings,
+                  tools: preparedCall.serializedTools,
+                  toolsExplicitlyProvided: preparedCall.toolsExplicitlyProvided,
+                  handoffs: preparedCall.serializedHandoffs,
+                  outputType: convertAgentOutputTypeToSerializable(
+                    currentAgent.outputType,
+                  ),
+                  tracing: getTracing(
+                    this.config.tracingDisabled,
+                    this.config.traceIncludeSensitiveData,
+                  ),
+                },
+              );
+              markAbortReconciliationComplete(
+                abortReconciliationState,
+                reconciliationResponse,
+              );
+              serverConversationTracker.trackServerItems(
+                reconciliationResponse,
+              );
+              result.state.setConversationContext(
+                serverConversationTracker.conversationId,
+                serverConversationTracker.previousResponseId,
+              );
+            } catch (error) {
+              logger.debug(
+                'Failed to reconcile streamed function calls after abort.',
+                error,
+              );
+            }
+          };
 
           sentInputToModel = true;
           if (!delayStreamInputPersistence) {
@@ -1301,6 +1368,10 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
             )) {
               guardrailTracker.throwIfError();
               markInputOnce();
+              recordStreamEventForAbortReconciliation(
+                abortReconciliationState,
+                event,
+              );
               if (event.type === 'response_done') {
                 const parsed = StreamEventResponseCompleted.parse(event);
                 finalResponse = {
@@ -1315,6 +1386,7 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
                 // When the user's code exits a loop to consume the stream, we need to break
                 // this loop to prevent internal false errors and unnecessary processing
                 await awaitGuardrailsAndPersistInput();
+                await reconcileStreamAbortIfNeeded();
                 return;
               }
               result._addItem(new RunRawModelStreamEvent(event));
@@ -1325,6 +1397,7 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
                 markInputOnce();
               }
               await awaitGuardrailsAndPersistInput();
+              await reconcileStreamAbortIfNeeded();
               return;
             }
             throw error;

packages/agents-core/src/runner/streamReconciliation.ts

@@ -0,0 +1,118 @@
+import type { ModelRequest, ModelResponse } from '../model';
+import type {
+  FunctionCallItem,
+  FunctionCallResultItem,
+  StreamEvent,
+} from '../types/protocol';
+
+type PendingStreamedFunctionCall = Pick<
+  FunctionCallItem,
+  'callId' | 'name' | 'namespace'
+>;
+
+export type StreamAbortReconciliationState = {
+  responseId?: string;
+  pendingFunctionCalls: Map<string, PendingStreamedFunctionCall>;
+};
+
+export function createStreamAbortReconciliationState(): StreamAbortReconciliationState {
+  return {
+    pendingFunctionCalls: new Map(),
+  };
+}
+
+export function recordStreamEventForAbortReconciliation(
+  state: StreamAbortReconciliationState,
+  event: StreamEvent,
+): void {
+  if (event.type === 'response_done') {
+    state.pendingFunctionCalls.clear();
+    state.responseId = event.response.id;
+    return;
+  }
+
+  if (event.type !== 'model' || !isRecord(event.event)) {
+    return;
+  }
+
+  const rawEvent = event.event;
+  if (
+    rawEvent.type === 'response.created' &&
+    isRecord(rawEvent.response) &&
+    typeof rawEvent.response.id === 'string'
+  ) {
+    state.responseId = rawEvent.response.id;
+    return;
+  }
+
+  if (
+    rawEvent.type !== 'response.output_item.done' ||
+    !isRecord(rawEvent.item)
+  ) {
+    return;
+  }
+
+  const item = rawEvent.item;
+  if (item.type === 'function_call' && typeof item.call_id === 'string') {
+    state.pendingFunctionCalls.set(item.call_id, {
+      callId: item.call_id,
+      name: typeof item.name === 'string' ? item.name : item.call_id,
+      ...(typeof item.namespace === 'string'
+        ? { namespace: item.namespace }
+        : {}),
+    });
+    return;
+  }
+
+  if (
+    item.type === 'function_call_output' &&
+    typeof item.call_id === 'string'
+  ) {
+    state.pendingFunctionCalls.delete(item.call_id);
+  }
+}
+
+export function buildAbortReconciliationInput(
+  state: StreamAbortReconciliationState,
+): FunctionCallResultItem[] {
+  return Array.from(state.pendingFunctionCalls.values(), (toolCall) => ({
+    type: 'function_call_result',
+    name: toolCall.name,
+    ...(typeof toolCall.namespace === 'string'
+      ? { namespace: toolCall.namespace }
+      : {}),
+    callId: toolCall.callId,
+    status: 'incomplete',
+    output: { type: 'text', text: 'aborted' },
+  }));
+}
+
+export function getAbortReconciliationPreviousResponseId(
+  state: StreamAbortReconciliationState,
+  request: Pick<ModelRequest, 'conversationId' | 'previousResponseId'>,
+): string | undefined {
+  if (request.conversationId) {
+    return request.previousResponseId;
+  }
+  return state.responseId ?? request.previousResponseId;
+}
+
+export function shouldReconcileStreamAbort(
+  state: StreamAbortReconciliationState,
+): boolean {
+  return state.pendingFunctionCalls.size > 0;
+}
+
+export function markAbortReconciliationComplete(
+  state: StreamAbortReconciliationState,
+  response: ModelResponse | undefined,
+): void {
+  state.pendingFunctionCalls.clear();
+  if (response?.responseId) {
+    state.responseId = response.responseId;
+  }
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null;
+}

packages/agents-core/test/run.stream.test.ts

@@ -60,6 +60,53 @@ function getRequestInputItems(request: ModelRequest): AgentInputItem[] {
   return Array.isArray(request.input) ? request.input : [];
 }
 
+class AbortAfterStreamedFunctionCallModel implements Model {
+  public requests: ModelRequest[] = [];
+
+  constructor(private readonly responseId: string) {}
+
+  async getResponse(request: ModelRequest): Promise<ModelResponse> {
+    this.requests.push(request);
+    return {
+      output: [fakeModelMessage('reconciled')],
+      usage: new Usage(),
+      responseId: 'resp-reconciled',
+    };
+  }
+
+  async *getStreamedResponse(
+    request: ModelRequest,
+  ): AsyncIterable<StreamEvent> {
+    this.requests.push(request);
+    yield {
+      type: 'model',
+      event: {
+        type: 'response.created',
+        response: {
+          id: this.responseId,
+        },
+      },
+    };
+    yield {
+      type: 'model',
+      event: {
+        type: 'response.output_item.done',
+        item: {
+          type: 'function_call',
+          id: 'fc_abort',
+          call_id: 'call_abort',
+          name: 'slow_tool',
+          arguments: '{}',
+          status: 'completed',
+        },
+      },
+    };
+    const error = new Error('aborted');
+    error.name = 'AbortError';
+    throw error;
+  }
+}
+
 // Test for unhandled rejection when stream loop throws
 
 describe('Runner.run (streaming)', () => {
@@ -212,6 +259,52 @@ describe('Runner.run (streaming)', () => {
     expect(getEventListeners(retainedSignals[0], 'abort').length).toBe(0);
   });
 
+  it('reconciles streamed function calls on abort with conversationId', async () => {
+    const model = new AbortAfterStreamedFunctionCallModel('resp-aborted');
+    const agent = new Agent({ name: 'AbortReconcile', model });
+
+    const result = await run(agent, 'hi', {
+      stream: true,
+      conversationId: 'conv-abort',
+    });
+
+    await result.completed;
+
+    expect(model.requests).toHaveLength(2);
+    expect(model.requests[1].conversationId).toBe('conv-abort');
+    expect(model.requests[1].signal).toBeUndefined();
+    expect(getRequestInputItems(model.requests[1])).toEqual([
+      expect.objectContaining({
+        type: 'function_call_result',
+        callId: 'call_abort',
+        name: 'slow_tool',
+        status: 'incomplete',
+        output: { type: 'text', text: 'aborted' },
+      }),
+    ]);
+  });
+
+  it('uses the streamed response id when reconciling previousResponseId-only aborts', async () => {
+    const model = new AbortAfterStreamedFunctionCallModel('resp-aborted');
+    const agent = new Agent({ name: 'AbortPreviousResponse', model });
+
+    const result = await run(agent, 'hi', {
+      stream: true,
+      previousResponseId: 'resp-before-abort',
+    });
+
+    await result.completed;
+
+    expect(model.requests).toHaveLength(2);
+    expect(model.requests[1].conversationId).toBeUndefined();
+    expect(model.requests[1].previousResponseId).toBe('resp-aborted');
+    expect(getRequestInputItems(model.requests[1])[0]).toMatchObject({
+      type: 'function_call_result',
+      callId: 'call_abort',
+      status: 'incomplete',
+    });
+  });
+
   it('emits agent_updated_stream_event with new agent on handoff', async () => {
     class SimpleStreamingModel implements Model {
       constructor(private resp: ModelResponse) {}