From 9a508773db4e48e1e411673b1ce58dff27c9117c Mon Sep 17 00:00:00 2001 From: Shaojin Wen Date: Thu, 18 Jun 2026 14:09:17 +0800 Subject: [PATCH] fix(daemon): centralize mid-turn event constant + recover timed-out drains (#5266) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 4 + packages/acp-bridge/src/bridgeClient.ts | 16 +- packages/acp-bridge/src/daemonEventTypes.ts | 25 +++ .../acp-integration/session/Session.test.ts | 97 ++++++++++ .../src/acp-integration/session/Session.ts | 181 +++++++++++++----- packages/sdk-typescript/src/daemon/events.ts | 18 +- packages/sdk-typescript/src/daemon/index.ts | 1 + .../test/unit/daemonEvents.test.ts | 14 ++ .../src/daemon/midTurnInjectedSidechannel.ts | 3 +- 9 files changed, 294 insertions(+), 65 deletions(-) create mode 100644 packages/acp-bridge/src/daemonEventTypes.ts diff --git a/packages/acp-bridge/package.json b/packages/acp-bridge/package.json index e6e9136495..07a7290448 100644 --- a/packages/acp-bridge/package.json +++ b/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" diff --git a/packages/acp-bridge/src/bridgeClient.ts b/packages/acp-bridge/src/bridgeClient.ts index a64a8826ae..3bb3d146cb 100644 --- a/packages/acp-bridge/src/bridgeClient.ts +++ b/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; diff --git a/packages/acp-bridge/src/daemonEventTypes.ts b/packages/acp-bridge/src/daemonEventTypes.ts new file mode 100644 index 0000000000..f1dbe297a4 --- /dev/null +++ b/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'; diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index c093c05f12..affe3a4443 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/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', diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 8927ee6c10..c9918ee090 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/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 { - 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 | 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, + ): Promise { + // 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((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 { + 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 diff --git a/packages/sdk-typescript/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index ea6a9cd6b4..4535b6fe28 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/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 { diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 77a6139b4b..9d62e742bf 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/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, diff --git a/packages/sdk-typescript/test/unit/daemonEvents.test.ts b/packages/sdk-typescript/test/unit/daemonEvents.test.ts index 48d4d142bf..e51027044d 100644 --- a/packages/sdk-typescript/test/unit/daemonEvents.test.ts +++ b/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 = { diff --git a/packages/webui/src/daemon/midTurnInjectedSidechannel.ts b/packages/webui/src/daemon/midTurnInjectedSidechannel.ts index ea6b956fcd..20c25b51f9 100644 --- a/packages/webui/src/daemon/midTurnInjectedSidechannel.ts +++ b/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; - 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;