diff --git a/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts b/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts index e12fbc57d6..12d6af832a 100644 --- a/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts +++ b/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts @@ -34,6 +34,7 @@ import { LISTENING_LINE_RE, stopDaemon } from './daemon-process.js'; import { HOSTED_HOME_PREFIX } from '../scratch-dir.js'; import { relayUpstream } from './hosted-relay-headers.js'; import { waitUntil } from './hosted-harness-process.js'; +import { parseWireRows, type WireRow } from './wire-rows.js'; const config = JSON.parse(await readFile(process.argv[2], 'utf8')) as { tenantId: string; @@ -544,25 +545,9 @@ const model = await startFakeOpenAIServer(({ body }) => { const daemon = await new WedgeDaemon().start(model.baseUrl); -interface WireRow { - kind?: string; - requestId?: number; - at?: number; - method?: string; - url?: string; - operation?: string; - status?: number; - requestSnippet?: string; - responseSnippet?: string; - replied?: string; -} - async function readWire(): Promise { try { - return (await readFile(config.wireFile, 'utf8')) - .split('\n') - .filter(Boolean) - .map((line) => JSON.parse(line) as WireRow); + return parseWireRows(await readFile(config.wireFile, 'utf8')); } catch (cause) { if ((cause as NodeJS.ErrnoException)?.code === 'ENOENT') return []; throw cause; diff --git a/integration-tests/helpers/wire-rows.test.ts b/integration-tests/helpers/wire-rows.test.ts new file mode 100644 index 0000000000..6bc889b7b6 --- /dev/null +++ b/integration-tests/helpers/wire-rows.test.ts @@ -0,0 +1,41 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from 'vitest'; +import { parseWireRows } from './wire-rows.js'; + +describe('parseWireRows', () => { + it('returns no rows for an empty buffer', () => { + expect(parseWireRows('')).toEqual([]); + }); + + it('returns no rows when the only row has no trailing newline yet', () => { + expect(parseWireRows('{"kind":"a"}')).toEqual([]); + }); + + it('drops an unterminated tail and keeps the complete rows', () => { + // Two complete rows keep lastIndexOf -> indexOf observable; with a single + // complete row that mutant survives. + expect(parseWireRows('{"kind":"a"}\n{"kind":"b"}\n{"kind":"c"')).toEqual([ + { kind: 'a' }, + { kind: 'b' }, + ]); + }); + + it('returns one row for a single terminated row', () => { + expect(parseWireRows('{"kind":"a"}\n')).toEqual([{ kind: 'a' }]); + }); + + it('returns no rows for a buffer that is only a newline', () => { + expect(parseWireRows('\n')).toEqual([]); + }); + + it('still throws on a malformed complete row', () => { + expect(() => parseWireRows('{"kind":"a"}\n{"kind":\n')).toThrow( + SyntaxError, + ); + }); +}); diff --git a/integration-tests/helpers/wire-rows.ts b/integration-tests/helpers/wire-rows.ts new file mode 100644 index 0000000000..9d8eb4b7df --- /dev/null +++ b/integration-tests/helpers/wire-rows.ts @@ -0,0 +1,31 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +export interface WireRow { + kind?: string; + requestId?: number; + at?: number; + method?: string; + url?: string; + operation?: string; + status?: number; + requestSnippet?: string; + responseSnippet?: string; + replied?: string; +} + +// The blocked-wedge driver appends each row with one appendFile of +// JSON.stringify(row) + '\n', and polls the same file from the same process, +// so a read can land mid-append. Parse only complete rows: an unterminated +// tail is the row still being written, and the next poll reads it whole. A +// complete-but-malformed row still reaches JSON.parse and throws. +export function parseWireRows(text: string): WireRow[] { + return text + .slice(0, text.lastIndexOf('\n') + 1) + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as WireRow); +}