fix(ai): CodexClient consumes SSE stream from chatgpt.com backend

After v1.4.10 fixed the request body, the Codex backend started
returning 400:

  { "detail": "Stream must be set to true" }

The endpoint is streaming-only — `stream: false` is rejected
outright. The official Codex CLI consumes a Server-Sent-Events
stream (`codex-rs/codex-api/src/sse/responses.rs`); we have to do
the same.

CodexClient now requests `stream: true` and reads the response body
incrementally via the standard `ReadableStream` reader. Each SSE
message ends in a blank line; the `data:` payload is JSON. We listen
for three event types from the OpenAI Responses API protocol:

- `response.output_item.done` with an assistant `message` item — the
  final text is read directly from the message's `output_text`
  content. Preferred path because the backend has already stitched
  any deltas together.
- `response.output_text.delta` — incremental text chunks. Fallback
  when no `output_item.done` arrived (e.g. older models or
  reasoning-only responses), all deltas are concatenated.
- `response.completed` — picks up `usage.total_tokens`.
- `response.error` / `error` — bubbled as a thrown Error with the
  upstream message.

Tests rebuilt to drive the client with synthetic SSE bodies, plus
new coverage for the delta-fallback path. All 720 unit tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.7 committed 2026-05-09 14:16:45 +02:00
1 parent 68e5797e48
commit c838d2a37a
4 files changed
+204 -42

No files matched your search

+14
View File
@@ -1,5 +1,19 @@
# Changelog
## [1.4.11] — 2026-05-09
### Fixed — KI Insights via ChatGPT
- **Codex-Backend bekommt jetzt SSE-Streaming.** Nach dem v1.4.10-
Body-Format-Fix kam der nächste 400er: `Stream must be set to true`
— der `chatgpt.com/backend-api/codex/responses`-Endpoint akzeptiert
ausschließlich Server-Sent-Events-Antworten, kein synchrones JSON.
v1.4.11 baut den CodexClient so um, dass er die Antwort als SSE
konsumiert: `output_item.done`-Events liefern den vollständigen
Assistant-Text, `output_text.delta`-Chunks dienen als Fallback,
`response.completed` trägt die Token-Usage. Damit läuft das gesamte
KI-Insights-Feature jetzt durch dein ChatGPT-Abo.
## [1.4.10] — 2026-05-09
### Fixed — KI Insights via ChatGPT
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "healthlog",
"version": "1.4.10",
"version": "1.4.11",
"private": true,
"packageManager": "pnpm@10.31.0",
"scripts": {
+80 -27
View File
@@ -1,25 +1,52 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import { CodexClient } from "../codex-client";
/**
* Build a `Response` whose `.body` streams a list of SSE event strings.
* Mirrors the wire format the Codex backend emits (each event terminated
* with `\n\n`).
*/
function sseResponse(events: string[]): Response {
const encoder = new TextEncoder();
const stream = new ReadableStream<Uint8Array>({
start(controller) {
for (const ev of events) controller.enqueue(encoder.encode(ev));
controller.close();
},
});
return new Response(stream, {
status: 200,
headers: { "content-type": "text/event-stream" },
});
}
const messageDoneEvent = (text: string) =>
`event: response.output_item.done\ndata: ${JSON.stringify({
type: "response.output_item.done",
item: {
type: "message",
role: "assistant",
content: [{ type: "output_text", text }],
},
})}\n\n`;
const completedEvent = (totalTokens = 50) =>
`event: response.completed\ndata: ${JSON.stringify({
type: "response.completed",
response: { id: "resp1", usage: { total_tokens: totalTokens } },
})}\n\n`;
describe("CodexClient", () => {
beforeEach(() => {
vi.restoreAllMocks();
});
it("sends request to codex responses endpoint", async () => {
const mockFetch = vi.fn().mockResolvedValue({
ok: true,
json: () =>
Promise.resolve({
output: [
{
type: "message",
content: [{ type: "output_text", text: '{"summary":"ok"}' }],
},
],
usage: { total_tokens: 50 },
}),
});
it("assembles assistant text from output_item.done events", async () => {
const mockFetch = vi
.fn()
.mockResolvedValue(
sseResponse([messageDoneEvent('{"summary":"ok"}'), completedEvent(50)]),
);
vi.stubGlobal("fetch", mockFetch);
const client = new CodexClient({
@@ -34,13 +61,49 @@ describe("CodexClient", () => {
expect(result.content).toBe('{"summary":"ok"}');
expect(result.providerType).toBe("codex");
expect(result.tokensUsed).toBe(50);
expect(mockFetch.mock.calls[0][0]).toBe(
"https://chatgpt.com/backend-api/codex/responses",
);
const sentBody = JSON.parse(mockFetch.mock.calls[0][1].body);
expect(sentBody.stream).toBe(true);
expect(Array.isArray(sentBody.input)).toBe(true);
expect(sentBody.input[0].content[0]).toEqual({
type: "input_text",
text: "Analyze this.",
});
const headers = mockFetch.mock.calls[0][1].headers;
expect(headers.Authorization).toBe("Bearer test-token");
});
it("falls back to concatenated output_text deltas when no done event arrived", async () => {
const deltaEvent = (delta: string) =>
`event: response.output_text.delta\ndata: ${JSON.stringify({
type: "response.output_text.delta",
delta,
})}\n\n`;
const mockFetch = vi
.fn()
.mockResolvedValue(
sseResponse([
deltaEvent("Hello "),
deltaEvent("world"),
completedEvent(),
]),
);
vi.stubGlobal("fetch", mockFetch);
const client = new CodexClient({
accessToken: "test-token",
onTokenRefresh: vi.fn(),
});
const result = await client.generateCompletion({
systemPrompt: "test",
userPrompt: "test",
});
expect(result.content).toBe("Hello world");
});
it("calls onTokenRefresh on 401 and retries", async () => {
const newToken = "refreshed-token";
const onRefresh = vi.fn().mockResolvedValue(newToken);
@@ -52,19 +115,9 @@ describe("CodexClient", () => {
status: 401,
text: () => Promise.resolve("unauthorized"),
})
.mockResolvedValueOnce({
ok: true,
json: () =>
Promise.resolve({
output: [
{
type: "message",
content: [{ type: "output_text", text: '{"test":true}' }],
},
],
usage: { total_tokens: 10 },
}),
});
.mockResolvedValueOnce(
sseResponse([messageDoneEvent('{"test":true}'), completedEvent()]),
);
vi.stubGlobal("fetch", mockFetch);
const client = new CodexClient({
+109 -14
View File
@@ -93,31 +93,126 @@ export class CodexClient implements AIProvider {
tool_choice: "auto",
parallel_tool_calls: false,
store: false,
stream: false,
// Codex backend rejects sync responses with
// `Stream must be set to true`. We have to consume SSE.
stream: true,
}),
});
}
/**
* Consume the Codex SSE stream and assemble the final assistant
* message. The Codex backend emits OpenAI-Responses-API events:
*
* event: response.output_text.delta
* data: { "delta": "Hello" }
*
* event: response.output_item.done
* data: { "type": "message", "role": "assistant",
* "content": [{ "type": "output_text", "text": "Hello world" }] }
*
* event: response.completed
* data: { "type": "response.completed",
* "response": { "id": "...", "usage": { "total_tokens": 123 } } }
*
* Strategy:
* - Prefer the assembled text in `response.output_item.done` (any
* `message` item with `assistant` role and `output_text` content).
* This avoids re-stitching `delta` chunks ourselves.
* - Fall back to concatenating all `output_text.delta` chunks if no
* output_item.done arrived before the stream ended.
* - Pick up token usage from `response.completed` when present.
* - Throw on `response.error` events with the upstream message.
*/
private async parseResponse(res: Response): Promise<CompletionResult> {
const json = await res.json();
if (!res.body) {
throw new Error("Codex returned no response body");
}
const messageOutput = (
json.output as Array<{
type: string;
content?: Array<{ type: string; text?: string }>;
}>
)?.find((o) => o.type === "message");
const textContent = messageOutput?.content?.find(
(c) => c.type === "output_text",
);
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let assembledText = "";
let deltaText = "";
let tokensUsed: number | null = null;
if (!textContent?.text) {
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// SSE messages are separated by a blank line (\n\n).
let separator: number;
while ((separator = buffer.indexOf("\n\n")) !== -1) {
const rawEvent = buffer.slice(0, separator);
buffer = buffer.slice(separator + 2);
const dataLine = rawEvent
.split("\n")
.find((l) => l.startsWith("data:"));
if (!dataLine) continue;
const payload = dataLine.slice(5).trim();
if (!payload || payload === "[DONE]") continue;
let parsed: {
type?: string;
delta?: string;
item?: {
type?: string;
role?: string;
content?: Array<{ type?: string; text?: string }>;
};
response?: { usage?: { total_tokens?: number } };
error?: { message?: string };
};
try {
parsed = JSON.parse(payload);
} catch {
continue;
}
if (parsed.type === "response.output_text.delta" && parsed.delta) {
deltaText += parsed.delta;
continue;
}
if (parsed.type === "response.output_item.done" && parsed.item) {
const item = parsed.item;
if (item.type === "message" && item.role === "assistant") {
const text = item.content
?.filter((c) => c.type === "output_text" && c.text)
.map((c) => c.text!)
.join("");
if (text) assembledText += text;
}
continue;
}
if (parsed.type === "response.completed") {
tokensUsed = parsed.response?.usage?.total_tokens ?? null;
continue;
}
if (parsed.type === "response.error" || parsed.type === "error") {
const message =
parsed.error?.message ?? "Codex stream returned an error event";
throw new Error(message);
}
}
}
} finally {
reader.releaseLock();
}
const content = assembledText || deltaText;
if (!content) {
throw new Error("Codex returned empty content");
}
return {
content: textContent.text,
tokensUsed: json.usage?.total_tokens ?? null,
content,
tokensUsed,
model: CODEX_MODEL,
providerType: "codex",
};