refactor(core): centralize model stream attempt state

docs/design/2026-08-21-model-stream-attempt-state.md

@@ -0,0 +1,59 @@
+# Model stream attempt state
+
+## Problem
+
+`GeminiChat.sendMessageStream()` exposes chunks plus retry, compression, and
+model-fallback control events. Its consumers currently keep their own copies of
+the same per-attempt state: text, thoughts, tool calls, usage, response ids,
+finish reasons, and output-truncation markers.
+
+The implementations have drifted. The sub-agent loop ignores model fallback
+and treats continuation retries as fresh restarts. ACP background notifications
+keep failed-attempt text and usage across fresh retries and fallbacks. Other ACP
+loops reset tool calls but can retain stale usage. MessageDisplay also has no
+way to discard text from a restarted attempt.
+
+## Design
+
+Add a small core `ModelStreamAttemptState` that consumes the existing
+`StreamEvent` union and owns only protocol-derived state. It returns a compact
+transition for each event so callers can keep their surface-specific side
+effects.
+
+The state applies these rules:
+
+- chunks append visible text, thought text, and function calls, and replace the
+  latest usage, response id, and finish reason;
+- every retry clears tool calls and per-attempt metadata;
+- continuation retries preserve accumulated text and thought text;
+- fresh retries and model fallback discard accumulated text and thought text;
+- model fallback always starts a fresh attempt;
+- compressed events do not change attempt state.
+
+`Turn`, the sub-agent reasoning loop, forked queries, speculation, and all ACP
+raw-stream loops consume the same transitions. `MessageDisplayDispatcher` gains the same
+`restartAttempt(preserveText)` operation already used by telemetry output
+capture, so hook output follows the stream contract.
+
+The helper deliberately stays on the current Google response type. This change
+centralizes stream lifecycle semantics without attempting the larger protocol
+migration. A future protocol boundary can translate provider-neutral chunks
+into the same transition model.
+
+## Non-goals
+
+- Change provider request or response protocols.
+- Unify UI, ACP, and sub-agent rendering side effects.
+- Change retry or fallback policy inside `GeminiChat`.
+- Rework tool execution scheduling.
+
+## Verification
+
+- Pure state tests cover fresh retry, continuation retry, fallback, metadata,
+  and truncation reset.
+- Sub-agent tests prove continuation text is preserved and fallback state is
+  discarded.
+- Forked-query and speculation tests keep only the active attempt.
+- ACP background-notification tests prove output, usage, and MessageDisplay do
+  not retain a failed attempt.
+- Existing Turn and ACP stream tests remain green.

packages/cli/src/acp-integration/session/Session.test.ts

@@ -369,7 +369,12 @@ function setFakeHome(home: string): () => void {
 
 // Helper to create async generator with chunks (avoids memory leak)
 function createStreamWithChunks(
-  chunks: Array<{ type: unknown; value: unknown }>,
+  chunks: Array<{
+    type: unknown;
+    value?: unknown;
+    info?: unknown;
+    isContinuation?: boolean;
+  }>,
 ) {
   return (async function* () {
     for (const chunk of chunks) {
@@ -6704,6 +6709,118 @@ describe('Session', () => {
       );
     });
 
+    it.each([
+      {
+        name: 'fresh retry',
+        event: { type: core.StreamEventType.RETRY },
+      },
+      {
+        name: 'model fallback',
+        event: {
+          type: core.StreamEventType.MODEL_FALLBACK,
+          info: {
+            fromModel: 'primary',
+            toModel: 'fallback',
+            fallbackIndex: 1,
+          },
+        },
+      },
+    ])(
+      'discards stale background notification state on $name',
+      async ({ event }) => {
+        const messageBus = { request: vi.fn().mockResolvedValue({}) };
+        mockConfig.getMessageBus = vi.fn().mockReturnValue(messageBus);
+        mockConfig.getDisableAllHooks = vi.fn().mockReturnValue(false);
+        mockConfig.hasHooksForEvent = vi
+          .fn()
+          .mockImplementation(
+            (eventName: string) => eventName === 'MessageDisplay',
+          );
+        mockChat.sendMessageStream = vi
+          .fn()
+          .mockResolvedValueOnce(createEmptyStream())
+          .mockResolvedValueOnce(
+            createStreamWithChunks([
+              {
+                type: core.StreamEventType.CHUNK,
+                value: {
+                  candidates: [
+                    { content: { parts: [{ text: 'stale answer' }] } },
+                  ],
+                  usageMetadata: {
+                    promptTokenCount: 111,
+                    candidatesTokenCount: 222,
+                  },
+                },
+              },
+              event,
+              {
+                type: core.StreamEventType.CHUNK,
+                value: {
+                  candidates: [
+                    { content: { parts: [{ text: 'current answer' }] } },
+                  ],
+                },
+              },
+            ]),
+          );
+
+        await session.prompt({
+          sessionId: 'test-session-id',
+          prompt: [{ type: 'text', text: 'start background work' }],
+        });
+        const callback = mockBackgroundTaskRegistry.setNotificationCallback.mock
+          .calls[0][0] as (
+          displayText: string,
+          modelText: string,
+          meta: { agentId: string; status: string },
+        ) => void;
+
+        callback('done', '<task-notification />', {
+          agentId: 'agent-1',
+          status: 'completed',
+        });
+
+        await vi.waitFor(() => {
+          expect(mockClient.sessionUpdate).toHaveBeenCalledWith({
+            sessionId: 'test-session-id',
+            update: expect.objectContaining({
+              content: { type: 'text', text: 'current answer' },
+              _meta: expect.objectContaining({
+                source: 'background_notification_response',
+              }),
+            }),
+          });
+        });
+
+        const responseUpdates = vi
+          .mocked(mockClient.sessionUpdate)
+          .mock.calls.map(([request]) => request.update)
+          .filter(
+            (update) =>
+              update._meta?.['source'] === 'background_notification_response',
+          );
+        expect(responseUpdates).toHaveLength(1);
+        expect(responseUpdates[0]).toEqual(
+          expect.objectContaining({
+            content: { type: 'text', text: 'current answer' },
+          }),
+        );
+        expect(session.cumulativeUsage).toMatchObject({
+          promptTokens: 0,
+          candidateTokens: 0,
+        });
+
+        const finalDisplay = messageBus.request.mock.calls
+          .map(([request]) => request)
+          .find(
+            (request) =>
+              request.eventName === 'MessageDisplay' && request.input.is_final,
+          );
+        expect(finalDisplay?.input.displayed_text).toBe('current answer');
+      },
+    );
+
     it('attaches structured agent metadata built from the canonical entry label', async () => {
       mockChat.sendMessageStream = vi
         .fn()