Skip to content

Commit b01a623

Browse files
snowyukittysnowyukitty
committed
fix(sse): treat a data frame that parses to a non-record as malformed
`JSON.parse("null")` returns null rather than throwing, so a try/catch around the parse cannot see it. Four SSE frame parsers cast the result straight to `Record<string, unknown>` and dereferenced it on the next line, so a `data: null` frame walked past the malformed-frame guard and crashed the parser mid-stream. Validate the shape at each site instead of asserting it, keeping each parser's existing malformed-frame handling: openai-chat and google emit the established terminal error, anthropic drops the frame so the message_stop check still governs the outcome, and the web-search sidecar warns and skips. This also closes the milder half of the same defect — a scalar or array frame did not crash but was silently accepted as an empty frame. `src/web-search/parse.ts` is not named in the report. It has the identical defect and sits in the web-search path the reporter was actually running. The other eight SSE data-frame parsers were audited and are already correct: they parse into `unknown` and gate on a record predicate before dereferencing. Neither web-search warning copies the frame's content. The new non-record branch reports length and JSON shape only, and the pre-existing malformed-JSON warning one line above stops logging `payload.slice(0, 120)`: an upstream SSE payload can carry model output or credential material, and a frame that failed to parse is the least trustworthy content there is. A regression test asserts that a unique marker in the payload never reaches `console.warn` on either path. Closes #1219
1 parent 9326b85 commit b01a623

5 files changed

Lines changed: 236 additions & 9 deletions

File tree

‎src/adapters/anthropic.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -984,13 +984,21 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
984984
const payload = record.data.trim();
985985
if (!payload) continue;
986986

987-
let data: Record<string, unknown>;
987+
let parsed: unknown;
988988
try {
989-
data = JSON.parse(payload) as Record<string, unknown>;
989+
parsed = JSON.parse(payload);
990990
} catch {
991991
debugDroppedFrame("anthropic", payload);
992992
continue;
993993
}
994+
// `JSON.parse("null")` returns null instead of throwing, so the catch above cannot cover
995+
// it and the `data.type` read below crashed the stream. Drop a non-record frame the same
996+
// way an unparseable one is dropped, so the message_stop check still governs the outcome.
997+
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
998+
debugDroppedFrame("anthropic", payload);
999+
continue;
1000+
}
1001+
const data = parsed as Record<string, unknown>;
9941002

9951003
switch (record.event || data.type) {
9961004
case "message_start": {

‎src/adapters/google.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -497,13 +497,21 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte
497497
}
498498
let emittedContentEvent = false;
499499

500-
let chunk: Record<string, unknown>;
500+
let parsed: unknown;
501501
try {
502-
chunk = JSON.parse(payload);
502+
parsed = JSON.parse(payload);
503503
} catch {
504504
yield { type: "error", message: "malformed upstream SSE data frame" };
505505
return "terminate";
506506
}
507+
// `JSON.parse("null")` returns null rather than throwing, so the catch above cannot cover
508+
// it and the `chunk.error` read below crashed the stream (see openai-chat.ts). Scalars and
509+
// arrays are equally not frames, so the guard is written against the shape.
510+
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
511+
yield { type: "error", message: "malformed upstream SSE data frame" };
512+
return "terminate";
513+
}
514+
const chunk = parsed as Record<string, unknown>;
507515
sawAnyFrame = true;
508516

509517
// Inline provider error inside a 200 stream → terminal error (see openai-chat.ts).

‎src/adapters/openai-chat.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -958,13 +958,22 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
958958
return "terminate";
959959
}
960960

961-
let chunk: Record<string, unknown>;
961+
let parsed: unknown;
962962
try {
963-
chunk = JSON.parse(payload) as Record<string, unknown>;
963+
parsed = JSON.parse(payload);
964964
} catch {
965965
yield { type: "error", message: "malformed upstream SSE data frame" };
966966
return "terminate";
967967
}
968+
// Validate the shape instead of asserting it. `JSON.parse` yields a value, not necessarily
969+
// an object — `JSON.parse("null")` returns null without throwing, so the catch above never
970+
// sees it and the `chunk.error` read below crashed the stream mid-flight. A scalar or array
971+
// frame did not crash but was silently accepted as an empty one; both are malformed.
972+
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
973+
yield { type: "error", message: "malformed upstream SSE data frame" };
974+
return "terminate";
975+
}
976+
const chunk = parsed as Record<string, unknown>;
968977

969978
// A 200/OK chat-completions stream may carry an inline provider error envelope
970979
// instead of a clean [DONE]. Surface it as a terminal error so the bridge emits a

‎src/web-search/parse.ts‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -155,11 +155,23 @@ export async function parseSidecarSSE(response: Response): Promise<WebSearchResu
155155

156156
const handle = (payload: string): void => {
157157
if (!payload || payload === "[DONE]") return;
158-
let data: Record<string, unknown>;
159-
try { data = JSON.parse(payload) as Record<string, unknown>; } catch {
160-
console.warn(`[web-search-parse] malformed SSE JSON (${payload.length} chars): ${payload.slice(0, 120)}`);
158+
// Neither warning below copies the frame's content. An upstream SSE payload can carry model
159+
// output or credential material, and a malformed frame is exactly the case where the content
160+
// is least trustworthy. Length plus a classification separates the two failure modes in a log
161+
// without reproducing anything from the wire.
162+
let parsed: unknown;
163+
try { parsed = JSON.parse(payload); } catch {
164+
console.warn(`[web-search-parse] malformed SSE JSON (${payload.length} chars)`);
161165
return;
162166
}
167+
// `JSON.parse("null")` returns null rather than throwing, so the catch above cannot cover it
168+
// and the `data.type` read below threw out of parseSidecarSSE.
169+
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
170+
const shape = Array.isArray(parsed) ? "array" : parsed === null ? "null" : typeof parsed;
171+
console.warn(`[web-search-parse] non-record SSE JSON frame (${payload.length} chars, ${shape})`);
172+
return;
173+
}
174+
const data = parsed as Record<string, unknown>;
163175
const type = data.type as string | undefined;
164176
if (type === "response.output_text.delta" && typeof data.delta === "string") {
165177
acc.deltaText += data.delta;

‎tests/sse-null-data-frame.test.ts‎

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
import { describe, expect, spyOn, test } from "bun:test";
2+
import { createAnthropicAdapter as createAnthropicAdapterProduction } from "../src/adapters/anthropic";
3+
import { createGoogleAdapter as createGoogleAdapterProduction } from "../src/adapters/google";
4+
import { createOpenAIChatAdapter as createOpenAIChatAdapterProduction } from "../src/adapters/openai-chat";
5+
import type { AdapterEvent, OcxProviderConfig } from "../src/types";
6+
import { parseSidecarSSE } from "../src/web-search/parse";
7+
import { withTestTranslatorBudget } from "./helpers/translator-budget";
8+
9+
// `JSON.parse` returns a value, not necessarily an object: `JSON.parse("null")` is `null` and
10+
// never throws, so a `try/catch` around the parse cannot see it. Every payload below is
11+
// syntactically valid JSON that does NOT deserialize to a record, which is the input class that
12+
// walked past the malformed-frame guard and reached a property access on the parse result.
13+
//
14+
// Only `null` actually crashed — property access on a number, string, boolean or array is legal
15+
// in JS and quietly yields `undefined`, so those frames were silently accepted as empty ones.
16+
// Both are wrong for the same reason, so they are asserted together.
17+
const NON_RECORD_PAYLOADS = ["null", "42", '"text"', "true", "[]"] as const;
18+
19+
// The syntactically-invalid control. It was already handled correctly before this fix; every
20+
// assertion below is written as parity against it so the test states the actual requirement —
21+
// a valid-JSON non-record frame is treated exactly like an unparseable one — rather than
22+
// re-encoding each adapter's terminal message and drifting when those messages change.
23+
const INVALID_JSON_PAYLOAD = "{not json}";
24+
25+
// A token distinctive enough that finding it anywhere in a log line proves the frame's content
26+
// was copied there. Used by the web-search warning-hygiene tests below.
27+
const MARKER = "ocx-marker-4f21b7a9-do-not-log";
28+
29+
const createOpenAIChatAdapter = (...args: Parameters<typeof createOpenAIChatAdapterProduction>) =>
30+
withTestTranslatorBudget(createOpenAIChatAdapterProduction(...args));
31+
const createGoogleAdapter = (...args: Parameters<typeof createGoogleAdapterProduction>) =>
32+
withTestTranslatorBudget(createGoogleAdapterProduction(...args));
33+
const createAnthropicAdapter = (...args: Parameters<typeof createAnthropicAdapterProduction>) =>
34+
withTestTranslatorBudget(createAnthropicAdapterProduction(...args));
35+
36+
function sse(payload: string): Response {
37+
return new Response(`data: ${payload}\n\n`, {
38+
headers: { "content-type": "text/event-stream" },
39+
});
40+
}
41+
42+
async function collect(stream: AsyncGenerator<AdapterEvent>): Promise<AdapterEvent[]> {
43+
const events: AdapterEvent[] = [];
44+
for await (const event of stream) events.push(event);
45+
return events;
46+
}
47+
48+
const openAIChatProvider = {
49+
adapter: "openai-chat",
50+
baseUrl: "https://example.test/v1",
51+
apiKey: "sk-test",
52+
authMode: "key",
53+
} as OcxProviderConfig;
54+
55+
const googleProvider = {
56+
adapter: "google",
57+
baseUrl: "https://generativelanguage.googleapis.com",
58+
apiKey: "google-test-key",
59+
authMode: "key",
60+
} as OcxProviderConfig;
61+
62+
const anthropicProvider = {
63+
adapter: "anthropic",
64+
baseUrl: "https://example.test",
65+
apiKey: "key",
66+
} as OcxProviderConfig;
67+
68+
describe("SSE data frames that parse to a non-record", () => {
69+
describe("openai-chat adapter", () => {
70+
test.each(NON_RECORD_PAYLOADS)("`data: %s` is a terminal malformed frame", async payload => {
71+
const events = await collect(createOpenAIChatAdapter(openAIChatProvider).parseStream(sse(payload)));
72+
73+
expect(events.at(-1)).toEqual({ type: "error", message: "malformed upstream SSE data frame" });
74+
expect(events.some(event => event.type === "done")).toBe(false);
75+
});
76+
77+
test("matches the invalid-JSON control", async () => {
78+
const control = await collect(createOpenAIChatAdapter(openAIChatProvider).parseStream(sse(INVALID_JSON_PAYLOAD)));
79+
80+
for (const payload of NON_RECORD_PAYLOADS) {
81+
expect(await collect(createOpenAIChatAdapter(openAIChatProvider).parseStream(sse(payload)))).toEqual(control);
82+
}
83+
});
84+
85+
test("a non-record frame does not suppress a following [DONE]-less truncation check", async () => {
86+
const response = new Response([
87+
'data: {"choices":[{"delta":{"content":"partial"}}]}\n\n',
88+
"data: null\n\n",
89+
"data: [DONE]\n\n",
90+
].join(""), { headers: { "content-type": "text/event-stream" } });
91+
92+
const events = await collect(createOpenAIChatAdapter(openAIChatProvider).parseStream(response));
93+
94+
// Terminal on the malformed frame, exactly like the existing `{not-json}` case: the trailing
95+
// `[DONE]` must not be reached and must not turn a broken stream into a clean completion.
96+
expect(events.at(-1)).toEqual({ type: "error", message: "malformed upstream SSE data frame" });
97+
expect(events.some(event => event.type === "done")).toBe(false);
98+
});
99+
});
100+
101+
describe("google adapter", () => {
102+
test.each(NON_RECORD_PAYLOADS)("`data: %s` is a terminal malformed frame", async payload => {
103+
const events = await collect(createGoogleAdapter(googleProvider).parseStream(sse(payload)));
104+
105+
expect(events.at(-1)).toEqual({ type: "error", message: "malformed upstream SSE data frame" });
106+
expect(events.some(event => event.type === "done")).toBe(false);
107+
});
108+
109+
test("matches the invalid-JSON control", async () => {
110+
const control = await collect(createGoogleAdapter(googleProvider).parseStream(sse(INVALID_JSON_PAYLOAD)));
111+
112+
for (const payload of NON_RECORD_PAYLOADS) {
113+
expect(await collect(createGoogleAdapter(googleProvider).parseStream(sse(payload)))).toEqual(control);
114+
}
115+
});
116+
});
117+
118+
describe("anthropic adapter", () => {
119+
// The anthropic parser drops an unparseable frame and keeps reading rather than terminating,
120+
// so parity here means "dropped the same way" — the stream still fails closed at EOF because
121+
// no `message_stop` arrived, which is the pre-existing truncation guard doing its job.
122+
test.each(NON_RECORD_PAYLOADS)("`data: %s` is dropped, not thrown", async payload => {
123+
const control = await collect(createAnthropicAdapter(anthropicProvider).parseStream(sse(INVALID_JSON_PAYLOAD)));
124+
const events = await collect(createAnthropicAdapter(anthropicProvider).parseStream(sse(payload)));
125+
126+
expect(events).toEqual(control);
127+
expect(events.some(event => event.type === "done")).toBe(false);
128+
});
129+
130+
test("a non-record frame does not stop a later well-formed frame from being read", async () => {
131+
const response = new Response([
132+
"data: null\n\n",
133+
"event: message_start\n",
134+
'data: {"type":"message_start","message":{"usage":{"input_tokens":3,"output_tokens":1}}}\n\n',
135+
"event: content_block_start\n",
136+
'data: {"type":"content_block_start","content_block":{"type":"text","text":""}}\n\n',
137+
"event: content_block_delta\n",
138+
'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}}\n\n',
139+
"event: message_stop\n",
140+
'data: {"type":"message_stop"}\n\n',
141+
].join(""), { headers: { "content-type": "text/event-stream" } });
142+
143+
const events = await collect(createAnthropicAdapter(anthropicProvider).parseStream(response));
144+
145+
expect(events.some(event => event.type === "text" || event.type === "text_delta")).toBe(true);
146+
expect(events.at(-1)?.type).toBe("done");
147+
});
148+
});
149+
150+
describe("web-search sidecar parser", () => {
151+
test.each(NON_RECORD_PAYLOADS)("`data: %s` is ignored, not thrown", async payload => {
152+
const control = await parseSidecarSSE(sse(INVALID_JSON_PAYLOAD));
153+
const result = await parseSidecarSSE(sse(payload));
154+
155+
expect(result).toEqual(control);
156+
});
157+
158+
// An upstream SSE payload can carry model output or credential material, and a frame that
159+
// failed to parse is the least trustworthy content there is. Both warning paths — unparseable
160+
// and valid-but-not-a-record — must report the frame without reproducing it.
161+
test.each([
162+
["unparseable", `{${MARKER}`],
163+
["non-record", `"${MARKER}"`],
164+
])("the %s warning reports the frame without copying its content", async (_label, payload) => {
165+
const warn = spyOn(console, "warn").mockImplementation(() => {});
166+
try {
167+
await parseSidecarSSE(sse(payload));
168+
169+
// The warning has to actually fire, or the marker assertion below passes vacuously.
170+
expect(warn).toHaveBeenCalled();
171+
const logged = warn.mock.calls.flat().map(String).join("\n");
172+
expect(logged).not.toContain(MARKER);
173+
expect(logged).toContain("[web-search-parse]");
174+
} finally {
175+
warn.mockRestore();
176+
}
177+
});
178+
179+
test("a non-record frame does not discard the answer that follows it", async () => {
180+
const response = new Response([
181+
"data: null\n\n",
182+
'data: {"type":"response.output_text.done","text":"answer"}\n\n',
183+
].join(""), { headers: { "content-type": "text/event-stream" } });
184+
185+
const result = await parseSidecarSSE(response);
186+
187+
expect(result.text).toBe("answer");
188+
});
189+
});
190+
});

0 commit comments

Comments
 (0)