fix(runtime): retry transport interruptions after HTTP 2xx

packages/runtime/src/__tests__/ai-sdk-backend.test.ts

@@ -216,6 +216,182 @@ for (const terminal of ['gateway', 'eof', 'other'] as const) {
   });
 }
 
+describe('AiSdkBackend HTTP 2xx transport recovery', () => {
+  for (const outcome of ['recover', 'exhaust', 'stop', 'step-limit'] as const) {
+    test(`${outcome} after a durable tool result without repeating its effect`, async () => {
+      const durable = durableTurnHarness(`turn-5656-${outcome}`, 'write once and continue');
+      const attempts: ModelCallAttempt[] = [];
+      let calls = 0;
+      let effects = 0;
+      const model = new MockLanguageModelV4({
+        doStream: async () => {
+          calls += 1;
+          if (calls > 1 && (outcome !== 'recover' || calls === 2))
+            throw successfulResponseTransportFailure();
+          const chunks: LanguageModelV4StreamPart[] =
+            calls === 1
+              ? [
+                  { type: 'stream-start', warnings: [] },
+                  { type: 'tool-call', toolCallId: 'write-once', toolName: 'Write', input: '{}' },
+                  {
+                    type: 'finish',
+                    finishReason: { unified: 'tool-calls', raw: 'tool_calls' },
+                    usage: emptyUsage(),
+                  },
+                ]
+              : [
+                  { type: 'stream-start', warnings: [] },
+                  { type: 'text-start', id: 'answer' },
+                  { type: 'text-delta', id: 'answer', delta: 'Done' },
+                  { type: 'text-end', id: 'answer' },
+                  {
+                    type: 'finish',
+                    finishReason: { unified: 'stop', raw: 'stop' },
+                    usage: emptyUsage(),
+                  },
+                ];
+          return {
+            stream: simulateReadableStream({
+              chunks,
+              initialDelayInMs: null,
+              chunkDelayInMs: null,
+            }),
+          };
+        },
+      });
+      const backend = createBackend({
+        connection: connection(),
+        modelId: 'mock-model-id',
+        modelFactory: () => model,
+        tools: [
+          {
+            ...testTool('Write', z.object({})),
+            impl: async () => {
+              effects += 1;
+              return 'committed once';
+            },
+          },
+        ],
+        loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents,
+        ...(outcome === 'step-limit' ? { maxSteps: 1 } : {}),
+        recordModelCallAttempt: ({ attempt }) => {
+          attempts.push(attempt);
+        },
+        providerRetrySleep: async (_delayMs, signal) => {
+          if (outcome !== 'stop') return;
+          await new Promise<void>((_resolve, reject) => {
+            const abort = () => reject(signal.reason ?? new Error('aborted'));
+            if (signal.aborted) abort();
+            else signal.addEventListener('abort', abort, { once: true });
+          });
+        },
+      });
+      const events: SessionEvent[] = [];
+      for await (const event of backend.send(durable.input({ runId: 'run-1' }))) {
+        durable.record(event);
+        events.push(event);
+        if (outcome === 'stop' && event.type === 'provider_retry' && event.phase === 'scheduled') {
+          await backend.stop('user_stop');
+        }
+      }
+      assert.equal(effects, 1);
+      assert.equal(events.filter((event) => event.type === 'tool_result').length, 1);
+      assert.equal(
+        durable.ledger.filter((event) => event.content?.kind === 'function_response').length,
+        1,
+      );
+      if (outcome === 'step-limit') {
+        assert.equal(calls, 1);
+        assert.equal(
+          events.some((event) => event.type === 'provider_retry'),
+          false,
+        );
+        assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'step_limit');
+        return;
+      }
+      assert.ok(
+        model.doStreamCalls
+          .slice(1)
+          .every((call) => JSON.stringify(call.prompt).includes('committed once')),
+      );
+      assert.equal(attempts[1]?.errorClass, 'network');
+      assert.equal(attempts[1]?.httpStatus, 200);
+      assert.equal(attempts[1]?.retryable, true);
+      const error = events.find((event) => event.type === 'error');
+      const retries = events.filter(
+        (event) => event.type === 'provider_retry' && event.phase === 'scheduled',
+      );
+      const completion = events.find((event) => event.type === 'complete');
+      if (outcome === 'recover') {
+        assert.equal(calls, 3);
+        assert.equal(retries.length, 1);
+        assert.equal(error, undefined);
+        assert.equal(completion?.stopReason, 'end_turn');
+        assert.deepEqual(
+          attempts.map(({ status }) => status),
+          ['completed', 'failed', 'completed'],
+        );
+      } else if (outcome === 'exhaust') {
+        assert.equal(calls, 11);
+        assert.equal(retries.length, 9);
+        assert.deepEqual(error?.retry, { decision: 'exhausted', attempts: 10 });
+        assert.equal(completion?.stopReason, 'error');
+      } else if (outcome === 'stop') {
+        assert.equal(calls, 2);
+        assert.equal(retries.length, 1);
+        assert.equal(
+          events.some((event) => event.type === 'provider_retry' && event.phase === 'started'),
+          false,
+        );
+        assert.equal(completion?.stopReason, 'user_stop');
+      }
+    });
+  }
+
+  test('does not replay provider tool activity after a 2xx transport failure', async () => {
+    let calls = 0;
+    const model = new MockLanguageModelV4({
+      doStream: async () => {
+        calls += 1;
+        return {
+          stream: simulateReadableStream({
+            chunks: [
+              { type: 'stream-start', warnings: [] },
+              {
+                type: 'tool-input-start',
+                id: 'search-1',
+                toolName: 'web_search',
+                providerExecuted: true,
+              },
+              { type: 'error', error: successfulResponseTransportFailure() },
+            ] as LanguageModelV4StreamPart[],
+            initialDelayInMs: null,
+            chunkDelayInMs: null,
+          }),
+        };
+      },
+    });
+    const durable = durableTurnHarness('turn-5656-side-effects', 'search once');
+    const backend = createBackend({
+      connection: connection(),
+      modelId: 'mock-model-id',
+      modelFactory: () => model,
+      tools: [],
+      loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents,
+      providerRetrySleep: async () => {},
+    });
+    const events = await drainDurably(backend.send(durable.input()), durable);
+    assert.equal(calls, 1);
+    assert.equal(
+      events.some((event) => event.type === 'provider_retry'),
+      false,
+    );
+    const error = events.find((event) => event.type === 'error');
+    assert.equal(error?.reason, 'network');
+    assert.deepEqual(error?.retry, { decision: 'declined', because: 'side_effects' });
+  });
+});
+
 describe('AiSdkBackend ApplyPatch routing', () => {
   test('advertises apply_patch only to supported native OpenAI models', async () => {
     for (const [providerType, modelId, expected] of [
@@ -10712,31 +10888,42 @@ describe('AiSdkBackend RunTrace', () => {
     assert.notEqual(assistants[0]?.id, assistants[1]?.id);
   });
 
-  for (const { label, providerMetadata, connectionOverride, modelId } of [
-    {
-      label: 'encrypted Responses',
-      providerMetadata: {
-        openai: { itemId: 'reasoning-item-1', reasoningEncryptedContent: 'encrypted-reasoning' },
+  for (const { label, providerMetadata, connectionOverride, modelId, makeFailure } of (
+    [
+      {
+        label: 'encrypted Responses',
+        providerMetadata: {
+          openai: { itemId: 'reasoning-item-1', reasoningEncryptedContent: 'encrypted-reasoning' },
+        },
+        connectionOverride: {
+          slug: 'openai',
+          providerType: 'openai',
+          defaultModel: 'gpt-5.4',
+        } as const,
+        modelId: 'gpt-5.4',
       },
-      connectionOverride: {
-        slug: 'openai',
-        providerType: 'openai',
-        defaultModel: 'gpt-5.4',
-      } as const,
-      modelId: 'gpt-5.4',
-    },
+      {
+        label: 'redacted Anthropic',
+        providerMetadata: { anthropic: { redactedData: 'redacted-reasoning' } },
+        connectionOverride: undefined,
+        modelId: 'mock-model-id',
+      },
+    ] as {
+      label: string;
+      providerMetadata: Record<string, Record<string, string>>;
+      connectionOverride:
+        | { slug: string; providerType: 'openai'; defaultModel: string }
+        | undefined;
+      modelId: string;
+    }[]
+  ).flatMap((scenario) => [
+    { ...scenario, makeFailure: connectionResetFailure },
     {
-      label: 'redacted Anthropic',
-      providerMetadata: { anthropic: { redactedData: 'redacted-reasoning' } },
-      connectionOverride: undefined,
-      modelId: 'mock-model-id',
+      ...scenario,
+      label: `${scenario.label} after HTTP 200`,
+      makeFailure: successfulResponseTransportFailure,
     },
-  ] as {
-    label: string;
-    providerMetadata: Record<string, Record<string, string>>;
-    connectionOverride: { slug: string; providerType: 'openai'; defaultModel: string } | undefined;
-    modelId: string;
-  }[]) {
+  ])) {
     test(`preserves finalized ${label} thinking when the next part fails without retrying`, async () => {
       // Continuation identity (Responses reasoning item ids, encrypted
       // content) cannot be replayed into a fresh request, so thinking that
@@ -10767,7 +10954,7 @@ describe('AiSdkBackend RunTrace', () => {
               { type: 'reasoning-start', id: 'reasoning-2' },
               { type: 'reasoning-delta', id: 'reasoning-2', delta: 'second thought' },
             ],
-            connectionResetFailure(),
+            makeFailure(),
           );
           failCurrentStream = failing.fail;
           return { stream: failing.stream };
@@ -16986,6 +17173,16 @@ function connectionResetFailure(): Error {
   });
 }
 
+function successfulResponseTransportFailure(): APICallError {
+  return new APICallError({
+    message: 'Failed to process successful response',
+    url: 'https://provider.invalid',
+    requestBodyValues: {},
+    statusCode: 200,
+    cause: Object.assign(new Error('connection closed'), { code: 'UND_ERR_SOCKET' }),
+  });
+}
+
 /**
  * Streams `chunks`, then hangs until `fail()` — mirroring a provider that
  * streams part of a step and then drops the connection mid-stream. The chunks