fix(ai): support JSONL responses in Google-compatible streams

packages/ai/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Fixed
 
+- Google-compatible streaming providers now consume newline-delimited JSON responses when the response media type declares NDJSON or JSONL, while preserving SSE parsing and event hooks for standard event streams.
+
 - `todo_write` raw argument rejections now carry bounded, authority-controlled correction codes for each rejected shape: unknown root keys, unknown operation-entry keys, done/drop entries missing a task or phase target, and unknown init list-entry keys. Each code maps to a fixed correction message naming the accepted shape (never echoing the offending input), so invalid calls surface specific guidance while valid payloads keep the existing passthrough/coercion path (#3916).
 - Anthropic Sonnet 5 now exposes Anthropic's real `xhigh` and `max` thinking efforts on the Messages API (`minimal`/`low`/`medium`/`high`/`xhigh`/`max`), matching official support. The previous generic `kind === opus` gate excluded it from the full preset range; the capability predicate is now an explicit version-scoped list (Opus 4.7+, Sonnet 5+), so older Sonnet generations and Bedrock Converse routes stay fail-closed at their previously advertised levels (issue #3913).
 - Alibaba Token Plan now exposes Qwen 3.8 Max under the provider-supported `qwen3.8-max` wire id instead of the rejected `qwen-3.8-max` spelling; catalog regeneration canonicalizes a legacy discovered alias rather than retaining a broken duplicate (#3909).

packages/ai/src/providers/google-shared.ts

@@ -2,7 +2,7 @@
  * Shared utilities for Google Generative AI and Google Cloud Code Assist providers.
  */
 
-import { extractHttpStatusFromError, readSseJson } from "@gajae-code/utils";
+import { extractHttpStatusFromError, readJsonl, readSseJson } from "@gajae-code/utils";
 import { calculateCost } from "../models";
 import type {
 	Api,
@@ -914,13 +914,21 @@ export function streamGoogleGenAI<T extends "google-generative-ai" | "google-ver
 				throw new Error("Google API returned an empty response body");
 			}
 
-			const googleStream = readSseJson<GenerateContentResponse>(response.body, options?.signal, event =>
-				options?.onSseEvent?.(
-					{ event: event.event, data: event.data, raw: [...event.raw] },
-					model,
-					options?.attemptScope,
-				),
-			);
+			const mediaType = (response.headers.get("content-type") ?? "").split(";", 1)[0]?.trim().toLowerCase() ?? "";
+			const isJsonLines =
+				mediaType === "application/x-ndjson" ||
+				mediaType === "application/ndjson" ||
+				mediaType === "application/jsonl" ||
+				mediaType === "application/x-jsonl";
+			const googleStream = isJsonLines
+				? readJsonl<GenerateContentResponse>(response.body, options?.signal)
+				: readSseJson<GenerateContentResponse>(response.body, options?.signal, event =>
+						options?.onSseEvent?.(
+							{ event: event.event, data: event.data, raw: [...event.raw] },
+							model,
+							options?.attemptScope,
+						),
+					);
 
 			stream.push({ type: "start", partial: output });
 			await consumeGoogleStream({

packages/ai/test/google-stream-content-type.test.ts

@@ -0,0 +1,105 @@
+import { describe, expect, it } from "bun:test";
+import { streamGoogleGenAI } from "../src/providers/google-shared";
+import { collectEvents, createBaseModel, createSseResponse } from "./openai-tool-choice-test-helpers";
+
+const chunks = [
+	{ candidates: [{ content: { parts: [{ text: "hello " }] } }] },
+	{
+		candidates: [{ content: { parts: [{ text: "world" }] }, finishReason: "STOP" }],
+		usageMetadata: { promptTokenCount: 2, candidatesTokenCount: 2, totalTokenCount: 4 },
+	},
+];
+
+function streamResponse(response: Response, onSseEvent?: () => void) {
+	const model = createBaseModel("google-generative-ai");
+	return streamGoogleGenAI({
+		model,
+		api: "google-generative-ai",
+		options: onSseEvent ? { onSseEvent } : undefined,
+		prepare: () => ({
+			params: { model: model.id, contents: [] },
+			url: "https://provider.example.test/stream",
+			headers: {},
+			fetch: async () => response,
+		}),
+	});
+}
+
+function createSseFixture(contentType?: string): Response {
+	const response = createSseResponse(chunks);
+	return new Response(response.body, {
+		headers: contentType ? { "content-type": contentType } : {},
+	});
+}
+
+describe("Google stream response framing", () => {
+	it.each([
+		"Application/X-NDJSON; Charset=UTF-8",
+		"application/jsonl",
+	] as const)("reads newline-delimited JSON responses with content type %s", async contentType => {
+		const response = new Response(chunks.map(chunk => JSON.stringify(chunk)).join("\n"), {
+			headers: { "content-type": contentType },
+		});
+		let sseEventCount = 0;
+		const stream = streamResponse(response, () => {
+			sseEventCount++;
+		});
+
+		const events = await collectEvents(stream);
+		const result = await stream.result();
+
+		expect(events.filter(event => event.type === "text_delta").map(event => event.delta)).toEqual([
+			"hello ",
+			"world",
+		]);
+		expect(result.content[0]).toMatchObject({ type: "text", text: "hello world" });
+		expect(result.usage.totalTokens).toBe(4);
+		expect(result.stopReason).toBe("stop");
+		expect(sseEventCount).toBe(0);
+	});
+
+	it("keeps parsing event-stream responses as SSE", async () => {
+		let sseEventCount = 0;
+		const stream = streamResponse(createSseResponse(chunks), () => {
+			sseEventCount++;
+		});
+
+		await collectEvents(stream);
+		const result = await stream.result();
+
+		expect(result.content[0]).toMatchObject({ type: "text", text: "hello world" });
+		expect(result.stopReason).toBe("stop");
+		expect(sseEventCount).toBe(2);
+	});
+
+	it.each([
+		{ label: "missing", contentType: undefined },
+		{ label: "unknown", contentType: 'application/octet-stream; profile="jsonl"' },
+	] as const)("keeps $label content types on the SSE parser", async ({ contentType }) => {
+		let sseEventCount = 0;
+		const stream = streamResponse(createSseFixture(contentType), () => {
+			sseEventCount++;
+		});
+
+		await collectEvents(stream);
+		const result = await stream.result();
+
+		expect(result.content[0]).toMatchObject({ type: "text", text: "hello world" });
+		expect(result.stopReason).toBe("stop");
+		expect(sseEventCount).toBe(2);
+	});
+
+	it("surfaces malformed newline-delimited JSON as a stream error", async () => {
+		const response = new Response('{"candidates":', {
+			headers: { "content-type": "application/x-ndjson" },
+		});
+		const stream = streamResponse(response);
+		const events = await collectEvents(stream);
+		const result = await stream.result();
+
+		expect(result.stopReason).toBe("error");
+		expect(result.errorMessage).toBeTruthy();
+		expect(events.filter(event => event.type === "error")).toHaveLength(1);
+		expect(events.filter(event => event.type === "done")).toHaveLength(0);
+	});
+});