fix(daemon): centralize mid-turn event constant + recover timed-out drains (#5266)

* fix(daemon): centralize mid-turn event constant + recover timed-out drains

Follow-up to #5175 addressing two post-merge /review suggestions.

Centralize the `mid_turn_message_injected` SSE event `type`: it was a bare
literal in the daemon publisher (acp-bridge), the SDK validator/reducer, and the
browser consumer (webui), so a rename in one could silently break browser-side
dedup. It now lives once in acp-bridge's dependency-free `daemonEventTypes`
module (lightweight like `mcpTimeouts`, so the SDK re-exports it via its
build-time devDep without dragging acp-bridge's type graph into the SDK bundle),
and bridgeClient / the SDK / webui all import the single binding.

Close the drain-timeout message-loss window: the daemon splices + SSE-publishes
(browser dedupes) before the ACP child's response lands, so if the child's 2s
drain timeout fires first, the late response was discarded — losing the messages
from both queues (silent, one-turn loss). The child now recovers that late
response and injects it on the next batch instead of dropping it.

Tests: drain-timeout recovery (Session), and a rename-safety assertion pinning
the shared event constant to the wire literal.

* fix(daemon): address review — log drain recovery + rename buildMidTurnParts

- Emit a `debugLogger.debug` line when a timed-out drain is recovered (session
  id + count), guarded on a non-empty payload, so the recovery path is
  correlatable in production logs.
- Rename `#formatMidTurnParts` → `#buildMidTurnParts`: the method records to the
  chat transcript, so a "format" verb understated its side effect.

packages/acp-bridge/package.json

@@ -47,6 +47,10 @@
       "types": "./dist/bridgeTypes.d.ts",
       "import": "./dist/bridgeTypes.js"
     },
+    "./daemonEventTypes": {
+      "types": "./dist/daemonEventTypes.d.ts",
+      "import": "./dist/daemonEventTypes.js"
+    },
     "./bridgeOptions": {
       "types": "./dist/bridgeOptions.d.ts",
       "import": "./dist/bridgeOptions.js"

packages/acp-bridge/src/bridgeClient.ts

@@ -19,6 +19,10 @@ import type {
 } from '@agentclientprotocol/sdk';
 import { RequestError } from '@agentclientprotocol/sdk';
 import type { BridgeEvent, EventBus } from './eventBus.js';
+// Wire constants shared with the child-side caller (`Session.ts`) and, for the
+// SSE event type, the SDK validator + browser consumer — single sources of truth
+// so a rename can't silently break the protocol.
+import { MID_TURN_MESSAGE_INJECTED_EVENT } from './daemonEventTypes.js';
 import { MID_TURN_QUEUE_DRAIN_METHOD } from './bridgeTypes.js';
 import type { MidTurnQueueEntry } from './bridgeTypes.js';
 import type { BridgeFileSystem } from './bridgeFileSystem.js';
@@ -198,18 +202,6 @@ function sliceLineRange(
  * `SessionEntry` is required to provide them — TS enforces the
  * structural match at the callback signature).
  */
-// `MID_TURN_QUEUE_DRAIN_METHOD` is imported from `./bridgeTypes.js` (the single
-// source of truth shared with the child-side caller in `Session.ts`).
-
-/**
- * SSE frame published when mid-turn messages are actually drained into the
- * running turn. The browser consumes it to move those messages out of its
- * pending queue so they aren't resent as the next turn (a transient dedupe
- * signal — it isn't rendered as a transcript item).
- * `data: { sessionId, messages: string[] }`.
- */
-const MID_TURN_MESSAGE_INJECTED_EVENT = 'mid_turn_message_injected';
-
 export interface BridgeClientSessionEntry {
   sessionId: string;
   events: EventBus;

packages/acp-bridge/src/daemonEventTypes.ts

@@ -0,0 +1,25 @@
+/**
+ * @license
+ * Copyright 2025 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * Daemon SSE event-`type` wire literals shared across the daemon publisher
+ * (`acp-bridge`), the SDK validator/reducer, and the browser consumer.
+ *
+ * Kept in this DEPENDENCY-FREE module (no `import type` from core, unlike
+ * `bridgeTypes.ts`) so the SDK can re-export these from `@qwen-code/sdk/daemon`
+ * via its build-time devDep on acp-bridge WITHOUT pulling acp-bridge's type
+ * graph into the SDK's bundled `.d.ts` — the same lightweight pattern as
+ * `mcpTimeouts.ts`.
+ */
+
+/**
+ * Published when the daemon drains queued mid-turn messages into the running
+ * turn. The browser consumes it to move those messages out of its pending queue
+ * so they aren't resent as the next turn (a transient dedupe signal). Single
+ * source of truth: a rename here propagates to every importer, so it can't
+ * silently break browser-side dedup. `data: { sessionId, messages: string[] }`.
+ */
+export const MID_TURN_MESSAGE_INJECTED_EVENT = 'mid_turn_message_injected';

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

@@ -3587,6 +3587,103 @@ describe('Session', () => {
         expect(drainCalls).toHaveLength(5);
       }, 30_000);
 
+      it('recovers a drain that timed out and injects it on the next batch', async () => {
+        // The daemon answers the drain (splices + SSE-publishes, so the browser
+        // already deduped) but we time out waiting. The late response must not be
+        // discarded — it is recovered and injected on the NEXT batch instead of
+        // being lost from both queues.
+        const tool = {
+          name: 'read_file',
+          kind: core.Kind.Read,
+          build: vi.fn().mockReturnValue({
+            params: { path: '/tmp/test.txt' },
+            getDefaultPermission: vi.fn().mockResolvedValue('allow'),
+            getDescription: vi.fn().mockReturnValue('Read file'),
+            toolLocations: vi.fn().mockReturnValue([]),
+            execute: vi
+              .fn()
+              .mockResolvedValue({ llmContent: 'ok', returnDisplay: 'ok' }),
+          }),
+        };
+        mockToolRegistry.getTool.mockReturnValue(tool);
+        mockConfig.getApprovalMode = vi.fn().mockReturnValue(ApprovalMode.YOLO);
+
+        // Prompt 1's drain: a promise we resolve LATE (after the timeout fires)
+        // with the messages the daemon drained. Prompt 2's drain: empty.
+        let resolveLate: (value: { messages: string[] }) => void = () => {};
+        const latePromise = new Promise<{ messages: string[] }>((res) => {
+          resolveLate = res;
+        });
+        let drainCalls = 0;
+        mockClient.extMethod = vi.fn((method: string) => {
+          if (method !== 'craft/drainMidTurnQueue') return Promise.resolve({});
+          drainCalls += 1;
+          return drainCalls === 1
+            ? latePromise
+            : Promise.resolve({ messages: [] });
+        });
+
+        const toolCallStream = () =>
+          createStreamWithChunks([
+            {
+              type: core.StreamEventType.CHUNK,
+              value: {
+                functionCalls: [
+                  {
+                    id: 'c',
+                    name: 'read_file',
+                    args: { path: '/tmp/test.txt' },
+                  },
+                ],
+              },
+            },
+          ]);
+        const streamMock = vi.fn();
+        for (let i = 0; i < 2; i++) {
+          streamMock
+            .mockResolvedValueOnce(toolCallStream())
+            .mockResolvedValueOnce(createEmptyStream());
+        }
+        mockChat.sendMessageStream = streamMock;
+
+        const prompt = {
+          sessionId: 'test-session-id',
+          prompt: [{ type: 'text' as const, text: 'read file' }],
+        };
+
+        // Prompt 1: the drain times out (latePromise still pending). Nothing is
+        // injected yet.
+        await session.prompt(prompt);
+
+        // The daemon's answer finally arrives. The timeout branch's handler
+        // stashes it for recovery; flush microtasks so the push lands.
+        resolveLate({ messages: ['please also check tests'] });
+        await new Promise((r) => setTimeout(r, 0));
+
+        // Prompt 2: the drain flushes the recovered message into this batch.
+        await session.prompt(prompt);
+
+        const midTurnPart = {
+          text: '\n[User message received during tool execution]: please also check tests',
+        };
+        // Injected into prompt 2's follow-up (4th sendMessageStream call), not
+        // prompt 1's (which timed out with nothing to inject).
+        const calls = vi.mocked(mockChat.sendMessageStream).mock.calls;
+        expect(calls[1]?.[1].message).not.toEqual(
+          expect.arrayContaining([midTurnPart]),
+        );
+        expect(calls[3]?.[1].message).toEqual(
+          expect.arrayContaining([midTurnPart]),
+        );
+        // Recorded exactly once, at injection time.
+        expect(
+          mockChatRecordingService.recordMidTurnUserMessage,
+        ).toHaveBeenCalledTimes(1);
+        expect(
+          mockChatRecordingService.recordMidTurnUserMessage,
+        ).toHaveBeenCalledWith([midTurnPart], 'please also check tests');
+      }, 20_000);
+
       it('keeps mid-turn drain enabled after a transient error', async () => {
         const tool = {
           name: 'read_file',

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

@@ -185,6 +185,11 @@ const ASK_USER_QUESTION_CANCEL_SKIP_MESSAGE =
 // means the client silently drops unknown methods; without a deadline the
 // await would wedge the prompt turn forever.
 const MID_TURN_QUEUE_DRAIN_TIMEOUT_MS = 2_000;
+// Secondary deadline for recovering a drain whose response arrives AFTER the
+// 2s race timeout: within this window the late answer is re-injected on the next
+// batch; beyond it (e.g. degraded transport) it is dropped rather than pushed
+// into an unrelated turn's context.
+const MID_TURN_QUEUE_RECOVERY_TIMEOUT_MS = 30_000;
 const MID_TURN_QUEUE_RESOLVE_TIMEOUT_MS = 10_000;
 const MAX_MID_TURN_DRAIN_ITEMS = 10;
 const MID_TURN_ATTACHMENT_PROCESSING_FAILURE_TEXT =
@@ -644,6 +649,11 @@ export class Session implements SessionContext {
   private lastPromptTokenCountChat: GeminiChat | null = null;
   private midTurnDrainUnavailable = false;
   private midTurnDrainTimeoutStrikes = 0;
+  // Messages from a drain that the daemon answered but we timed out waiting for
+  // (the daemon already spliced + SSE-published them). Re-injected on the next
+  // batch so a transient stall can't silently lose them. See
+  // `#drainMidTurnUserMessages`.
+  private midTurnRecoveredMessages: DrainedMidTurnMessage[] = [];
 
   // Background notification drain state. ACP does not have the TUI's idle
   // hook, so the session serializes registry callbacks through this queue.
@@ -2097,7 +2107,16 @@ export class Session implements SessionContext {
   }
 
   async #drainMidTurnUserMessages(abortSignal: AbortSignal): Promise<Part[]> {
-    if (this.midTurnDrainUnavailable) return [];
+    // Flush anything recovered from a PRIOR timed-out drain first: the daemon
+    // splices + SSE-publishes synchronously, so on a timeout the browser has
+    // already deduped those messages — discarding the late response would lose
+    // them from both queues. We stash them (see the timeout branch) and
+    // re-inject them here on the next batch.
+    const recovered = this.#takeRecoveredMidTurnMessages();
+
+    if (this.midTurnDrainUnavailable) {
+      return this.#buildMidTurnParts(recovered, abortSignal);
+    }
 
     let drainPromise: ReturnType<AgentSideConnection['extMethod']> | undefined;
     try {
@@ -2118,49 +2137,10 @@ export class Session implements SessionContext {
         clearTimeout(timeoutHandle);
       }
       this.midTurnDrainTimeoutStrikes = 0;
-      const drainedMessages = parseMidTurnDrainResponse(response);
-      const drainedParts: Part[] = [];
-      for (const message of drainedMessages) {
-        const displayText =
-          message.kind === 'text' ? message.message : message.displayText;
-        let rawParts: Part[];
-        try {
-          rawParts =
-            message.kind === 'text'
-              ? [{ text: message.message }]
-              : await withTimeoutSignal(
-                  abortSignal,
-                  MID_TURN_QUEUE_RESOLVE_TIMEOUT_MS,
-                  (signal) => this.#resolvePrompt(message.content, signal),
-                );
-        } catch (messageError) {
-          if (abortSignal.aborted) return drainedParts;
-          const errorMessage = this.#formatError(messageError);
-          debugLogger.warn(
-            `Failed to resolve mid-turn message: ${errorMessage}`,
-          );
-          rawParts = [
-            {
-              text: displayText,
-            },
-          ];
-          if (
-            message.kind === 'structured' &&
-            hasInlineMediaContentBlock(message.content)
-          ) {
-            rawParts.push({
-              text: MID_TURN_ATTACHMENT_PROCESSING_FAILURE_TEXT,
-            });
-          }
-        }
-        const parts = prefixMidTurnUserMessageParts(rawParts, displayText);
-        this.config
-          .getChatRecordingService()
-          ?.recordMidTurnUserMessage(parts, displayText);
-        drainedParts.push(...parts);
-      }
-
-      return drainedParts;
+      return this.#buildMidTurnParts(
+        [...recovered, ...parseMidTurnDrainResponse(response)],
+        abortSignal,
+      );
     } catch (error) {
       // The ACP SDK rejects with the raw JSON-RPC error object
       // (`{ code, message, data }`), which is not an `Error` instance, so
@@ -2180,9 +2160,14 @@ export class Session implements SessionContext {
       const isTimeout = error instanceof MidTurnDrainTimeoutError;
       if (isTimeout) {
         this.midTurnDrainTimeoutStrikes += 1;
-        // The lost race leaves the request pending; if the client settles it
-        // later, a rejection must not surface as an unhandled rejection.
-        drainPromise?.catch(() => {});
+        // The lost race leaves the drain request pending. The daemon answers it
+        // by splicing the queue + publishing the SSE echo (so the browser has
+        // already deduped), then returns the messages we just timed out waiting
+        // for. Recover that late response and inject it on the next batch instead
+        // of discarding it (which would lose the messages from both queues —
+        // silent loss). `#recoverLateDrain` bounds the wait and swallows a late
+        // rejection.
+        if (drainPromise) void this.#recoverLateDrain(drainPromise);
       }
       // Repeated timeouts are also permanent: a conforming client answers
       // (or rejects with -32601) immediately, so sustained silence means the
@@ -2203,10 +2188,110 @@ export class Session implements SessionContext {
       debugLogger.warn(
         `Mid-turn queue drain ${isPermanentError ? 'permanently ' : ''}unavailable [session ${this.sessionId}]: ${errorMessage}`,
       );
-      return [];
+      // Even on a failed/timed-out drain, still inject anything recovered from
+      // an EARLIER timeout so a transient stall never strands those messages.
+      return this.#buildMidTurnParts(recovered, abortSignal);
     }
   }
 
+  /** Read and clear the buffer of messages recovered from a timed-out drain. */
+  #takeRecoveredMidTurnMessages(): DrainedMidTurnMessage[] {
+    if (this.midTurnRecoveredMessages.length === 0) return [];
+    const out = this.midTurnRecoveredMessages;
+    this.midTurnRecoveredMessages = [];
+    return out;
+  }
+
+  /**
+   * After a drain times out, the request is still pending; the daemon settles it
+   * shortly after (it splices + SSE-publishes synchronously, so the browser has
+   * already deduped). Recover that late response for the next batch instead of
+   * discarding it, but bound the wait with a secondary deadline so a response
+   * that only arrives long after the turn isn't pushed into an unrelated
+   * context. A late rejection is swallowed (no unhandled rejection).
+   */
+  async #recoverLateDrain(
+    pending: ReturnType<AgentSideConnection['extMethod']>,
+  ): Promise<void> {
+    // Swallow a late rejection regardless of which branch of the race wins.
+    pending.catch(() => {});
+    const expired = Symbol('mid-turn-recovery-expired');
+    let timer: NodeJS.Timeout | undefined;
+    const deadline = new Promise<typeof expired>((resolve) => {
+      timer = setTimeout(
+        () => resolve(expired),
+        MID_TURN_QUEUE_RECOVERY_TIMEOUT_MS,
+      );
+      timer.unref?.();
+    });
+    let late: unknown;
+    try {
+      late = await Promise.race([pending, deadline]);
+    } catch {
+      return; // late rejection — nothing to recover
+    } finally {
+      clearTimeout(timer);
+    }
+    if (late === expired) {
+      debugLogger.warn(
+        `[mid-turn] dropped a drain response that arrived after the ${MID_TURN_QUEUE_RECOVERY_TIMEOUT_MS}ms recovery deadline [session ${this.sessionId}]`,
+      );
+      return;
+    }
+    const lateMessages = parseMidTurnDrainResponse(late);
+    if (lateMessages.length > 0) {
+      debugLogger.debug(
+        `[mid-turn] recovered ${lateMessages.length} message(s) from a timed-out drain [session ${this.sessionId}]`,
+      );
+      this.midTurnRecoveredMessages.push(...lateMessages);
+    }
+  }
+
+  /**
+   * Resolve each drained mid-turn message (text or structured content) into
+   * agent-visible `Part`s and record it once to the chat transcript. Recording
+   * happens on injection (here), so a message recovered from an earlier
+   * timed-out drain is still recorded exactly once.
+   */
+  async #buildMidTurnParts(
+    messages: DrainedMidTurnMessage[],
+    abortSignal: AbortSignal,
+  ): Promise<Part[]> {
+    const parts: Part[] = [];
+    for (const message of messages) {
+      const displayText =
+        message.kind === 'text' ? message.message : message.displayText;
+      let rawParts: Part[];
+      try {
+        rawParts =
+          message.kind === 'text'
+            ? [{ text: message.message }]
+            : await withTimeoutSignal(
+                abortSignal,
+                MID_TURN_QUEUE_RESOLVE_TIMEOUT_MS,
+                (signal) => this.#resolvePrompt(message.content, signal),
+              );
+      } catch (messageError) {
+        if (abortSignal.aborted) return parts;
+        const errorMessage = this.#formatError(messageError);
+        debugLogger.warn(`Failed to resolve mid-turn message: ${errorMessage}`);
+        rawParts = [{ text: displayText }];
+        if (
+          message.kind === 'structured' &&
+          hasInlineMediaContentBlock(message.content)
+        ) {
+          rawParts.push({ text: MID_TURN_ATTACHMENT_PROCESSING_FAILURE_TEXT });
+        }
+      }
+      const built = prefixMidTurnUserMessageParts(rawParts, displayText);
+      this.config
+        .getChatRecordingService()
+        ?.recordMidTurnUserMessage(built, displayText);
+      parts.push(...built);
+    }
+    return parts;
+  }
+
   /**
    * Starts the cron scheduler if cron is enabled and jobs exist.
    * The scheduler runs in the background, pushing fired prompts into

packages/sdk-typescript/src/daemon/events.ts

@@ -10,6 +10,16 @@ import type {
   DaemonMcpTransport,
   PermissionOutcome,
 } from './types.js';
+// Single source of truth: the daemon publisher owns the wire literal in
+// acp-bridge's dependency-free `daemonEventTypes` module. We re-export it so the
+// validator/reducer below, and the browser consumer via `@qwen-code/sdk/daemon`,
+// share the exact same value — a rename can't silently break browser-side dedup.
+// The build-time devDep on acp-bridge inlines the value into the published bundle
+// (same lightweight mechanism as `@qwen-code/acp-bridge/mcpTimeouts`). A `const`
+// keeps its literal type, so it still narrows in `switch (event.type)` and works
+// as a `typeof`-d type argument.
+import { MID_TURN_MESSAGE_INJECTED_EVENT } from '@qwen-code/acp-bridge/daemonEventTypes';
+export { MID_TURN_MESSAGE_INJECTED_EVENT };
 
 export const DAEMON_KNOWN_EVENT_TYPE_VALUES = [
   'session_update',
@@ -21,7 +31,7 @@ export const DAEMON_KNOWN_EVENT_TYPE_VALUES = [
   'session_died',
   'session_closed',
   'session_metadata_updated',
-  'mid_turn_message_injected',
+  MID_TURN_MESSAGE_INJECTED_EVENT,
   'client_evicted',
   'slow_client_warning',
   'stream_error',
@@ -779,7 +789,7 @@ export type DaemonSessionMetadataUpdatedEvent = DaemonEventEnvelope<
   DaemonSessionMetadataUpdatedData
 >;
 export type DaemonMidTurnMessageInjectedEvent = DaemonEventEnvelope<
-  'mid_turn_message_injected',
+  typeof MID_TURN_MESSAGE_INJECTED_EVENT,
   DaemonMidTurnMessageInjectedData
 >;
 export type DaemonClientEvictedEvent = DaemonEventEnvelope<
@@ -1360,7 +1370,7 @@ export function asKnownDaemonEvent(
       return isSessionMetadataUpdatedData(event.data)
         ? (event as DaemonSessionMetadataUpdatedEvent)
         : undefined;
-    case 'mid_turn_message_injected':
+    case MID_TURN_MESSAGE_INJECTED_EVENT:
       return isMidTurnMessageInjectedData(event.data)
         ? (event as DaemonMidTurnMessageInjectedEvent)
         : undefined;
@@ -1850,7 +1860,7 @@ export function reduceDaemonSessionEvent(
     case 'mcp_server_added':
     case 'mcp_server_removed':
     case 'settings_reloaded':
-    case 'mid_turn_message_injected':
+    case MID_TURN_MESSAGE_INJECTED_EVENT:
       return base;
     case 'session_rewound':
       return {

packages/sdk-typescript/src/daemon/index.ts

@@ -56,6 +56,7 @@ export {
 export {
   asKnownDaemonEvent,
   DAEMON_KNOWN_EVENT_TYPE_VALUES,
+  MID_TURN_MESSAGE_INJECTED_EVENT,
   createDaemonAuthState,
   createDaemonSessionViewState,
   isDaemonEventType,

packages/sdk-typescript/test/unit/daemonEvents.test.ts

@@ -9,7 +9,9 @@ import {
   asKnownDaemonEvent,
   createDaemonAuthState,
   createDaemonSessionViewState,
+  DAEMON_KNOWN_EVENT_TYPE_VALUES,
   isDaemonEventType,
+  MID_TURN_MESSAGE_INJECTED_EVENT,
   reduceDaemonAuthEvent,
   reduceDaemonAuthEvents,
   reduceDaemonSessionEvent,
@@ -17,6 +19,18 @@ import {
 } from '../../src/daemon/events.js';
 import type { DaemonEvent } from '../../src/daemon/types.js';
 
+describe('MID_TURN_MESSAGE_INJECTED_EVENT (shared wire constant)', () => {
+  it('is the wire literal and a registered known event type', () => {
+    // The same const is imported by the daemon publisher (acp-bridge) and the
+    // browser consumer (webui), so this also pins THEIR matching. Changing the
+    // wire string is a deliberate protocol change and must update this literal.
+    expect(MID_TURN_MESSAGE_INJECTED_EVENT).toBe('mid_turn_message_injected');
+    expect(DAEMON_KNOWN_EVENT_TYPE_VALUES).toContain(
+      MID_TURN_MESSAGE_INJECTED_EVENT,
+    );
+  });
+});
+
 describe('daemon event schema', () => {
   it('narrows known daemon events by discriminator', () => {
     const event: DaemonEvent = {

packages/webui/src/daemon/midTurnInjectedSidechannel.ts

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import { MID_TURN_MESSAGE_INJECTED_EVENT } from '@qwen-code/sdk/daemon';
 import type { DaemonMidTurnMessageInjectedData } from '@qwen-code/sdk/daemon';
 
 /**
@@ -132,7 +133,7 @@ export function parseSidechannelMidTurnInjected(
 ): DaemonMidTurnMessageInjectedData | undefined {
   if (!event || typeof event !== 'object') return undefined;
   const record = event as Record<string, unknown>;
-  if (record['type'] !== 'mid_turn_message_injected') return undefined;
+  if (record['type'] !== MID_TURN_MESSAGE_INJECTED_EVENT) return undefined;
   const data = record['data'];
   if (!data || typeof data !== 'object') return undefined;
   const dataRecord = data as Record<string, unknown>;