fix local codex review comment

    - [P0] Delay streaming session persistence until after filter — packages/agents-core/src/run.ts:936-969
    For streaming runs we call ensureStreamInputPersisted?.() on the first turn before #prepareModelCall runs (line 936). ensureStreamInputPersisted immediately persists
  sessionInputItemsToPersist and flips its internal persisted flag. Later, #prepareModelCall invokes applyCallModelInputFilter, which may redact or truncate the turn input and
  updates sessionInputItemsFiltered. However, the second ensureStreamInputPersisted?.() (line 968) is a no-op because the persisted flag is already true, so the session history
  now contains the unfiltered, unredacted input that the filter was supposed to scrub. This leaks exactly the data the caller asked us not to keep. We need to defer the first
  persistence until after the filter (or otherwise allow the post-filter call to write the sanitized items) so that streaming memory mirrors the filtered payload.

Author: seratch

packages/agents-core/src/run.ts

@@ -935,7 +935,6 @@ export class Runner extends RunHooks<any, AgentOutputType<unknown>> {
 
           if (result.state._currentTurn === 1) {
             await this.#runInputGuardrails(result.state);
-            await ensureStreamInputPersisted?.();
           }
 
           const turnInput = serverConversationTracker

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

@@ -810,6 +810,66 @@ describe('Runner.run (streaming)', () => {
     });
   });
 
+  it('persists filtered streaming input instead of the raw turn payload', async () => {
+    const saveInputSpy = vi
+      .spyOn(runImplementation, 'saveStreamInputToSession')
+      .mockResolvedValue();
+
+    const session = createSessionMock();
+
+    const agent = new Agent({
+      name: 'StreamFiltered',
+      model: new ImmediateStreamingModel({
+        output: [fakeModelMessage('done')],
+        usage: new Usage(),
+      }),
+    });
+
+    const runner = new Runner();
+
+    const secretInput = 'super secret';
+    const redactedContent = '[filtered]';
+
+    const result = await runner.run(agent, secretInput, {
+      stream: true,
+      session,
+      callModelInputFilter: ({ modelData }) => {
+        const sanitizedInput = modelData.input.map((item) => {
+          if (
+            item.type === 'message' &&
+            'role' in item &&
+            item.role === 'user'
+          ) {
+            return {
+              ...item,
+              content: redactedContent,
+            };
+          }
+          return item;
+        });
+
+        return {
+          ...modelData,
+          input: sanitizedInput,
+        };
+      },
+    });
+
+    await result.completed;
+
+    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: redactedContent,
+    });
+    expect(JSON.stringify(persistedItems)).not.toContain(secretInput);
+  });
+
   it('skips streaming session persistence when the server manages the conversation', async () => {
     const saveInputSpy = vi
       .spyOn(runImplementation, 'saveStreamInputToSession')