fix local codex review comment:

    - [P1] Persist streaming session input on model errors — packages/agents-core/src/run.ts:1045-1086
    When we stream a run with a session, ensureStreamInputPersisted is only invoked in the final_output and interruption branches. If the model stream rejects before either
  branch is reached (for example a transient provider/network error, an abort signal, or a thrown tool error), we exit the loop via the catch path without ever calling
  ensureStreamInputPersisted, so saveStreamInputToSession never runs. That drops the user’s latest turn from the session, which prevents resuming the conversation precisely when
  we need memory most. We should persist the input as soon as we hand it to the model (or at least in a finally block) so streaming sessions remain consistent even when the stream
  fails early.

Author: seratch

packages/agents-core/src/run.ts

@@ -877,6 +877,16 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
         })
       : undefined;
 
+    let handedInputToModel = false;
+    let streamInputPersisted = false;
+    const persistStreamInputIfNeeded = async () => {
+      if (streamInputPersisted || !ensureStreamInputPersisted) {
+        return;
+      }
+      await ensureStreamInputPersisted();
+      streamInputPersisted = true;
+    };
+
     if (serverConversationTracker && isResumedState) {
       serverConversationTracker.primeFromState({
         originalInput: result.state._originalInput,
@@ -983,7 +993,8 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
             sessionInputUpdate,
           );
 
-          await ensureStreamInputPersisted?.();
+          handedInputToModel = true;
+          await persistStreamInputIfNeeded();
 
           for await (const event of preparedCall.model.getStreamedResponse({
             systemInstructions: preparedCall.modelInput.instructions,
@@ -1081,7 +1092,7 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
             result.state,
             result.state._currentStep.output,
           );
-          await ensureStreamInputPersisted?.();
+          await persistStreamInputIfNeeded();
           // Guardrails must succeed before persisting session memory to avoid storing blocked outputs.
           if (!serverManagesConversation) {
             await saveStreamResultToSession(options.session, result);
@@ -1102,7 +1113,7 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
           result.state._currentStep.type === 'next_step_interruption'
         ) {
           // we are done for now. Don't run any output guardrails
-          await ensureStreamInputPersisted?.();
+          await persistStreamInputIfNeeded();
           if (!serverManagesConversation) {
             await saveStreamResultToSession(options.session, result);
           }
@@ -1129,6 +1140,9 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
         }
       }
     } catch (error) {
+      if (handedInputToModel && !streamInputPersisted) {
+        await persistStreamInputIfNeeded();
+      }
       if (result.state._currentAgentSpan) {
         result.state._currentAgentSpan.setError({
           message: 'Error in agent run',

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

@@ -810,6 +810,40 @@ describe('Runner.run (streaming)', () => {
     });
   });
 
+  it('persists streaming input when the model stream rejects before completion', async () => {
+    const saveInputSpy = vi
+      .spyOn(runImplementation, 'saveStreamInputToSession')
+      .mockResolvedValue();
+
+    const session = createSessionMock();
+    const streamError = new Error('model stream failed');
+
+    const agent = new Agent({
+      name: 'StreamFailurePersistsInput',
+      model: new RejectingStreamingModel(streamError),
+    });
+
+    const runner = new Runner();
+
+    const result = await runner.run(agent, 'save me please', {
+      stream: true,
+      session,
+    });
+
+    await expect(result.completed).rejects.toThrow('model stream failed');
+
+    expect(saveInputSpy).toHaveBeenCalledTimes(1);
+    const [, persistedItems] = saveInputSpy.mock.calls[0];
+    if (!Array.isArray(persistedItems)) {
+      throw new Error('Expected persisted session items to be an array.');
+    }
+    expect(persistedItems).toHaveLength(1);
+    expect(persistedItems[0]).toMatchObject({
+      role: 'user',
+      content: 'save me please',
+    });
+  });
+
   it('persists filtered streaming input instead of the raw turn payload', async () => {
     const saveInputSpy = vi
       .spyOn(runImplementation, 'saveStreamInputToSession')
@@ -971,6 +1005,27 @@ class ImmediateStreamingModel implements Model {
   }
 }
 
+class RejectingStreamingModel implements Model {
+  constructor(private readonly error: Error) {}
+
+  async getResponse(_request: ModelRequest): Promise<ModelResponse> {
+    throw this.error;
+  }
+
+  getStreamedResponse(_request: ModelRequest): AsyncIterable<StreamEvent> {
+    const error = this.error;
+    return {
+      [Symbol.asyncIterator]() {
+        return {
+          async next() {
+            throw error;
+          },
+        } satisfies AsyncIterator<StreamEvent>;
+      },
+    } satisfies AsyncIterable<StreamEvent>;
+  }
+}
+
 function createSessionMock(): Session {
   return {
     getSessionId: vi.fn().mockResolvedValue('session-id'),