Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 24 additions & 8 deletions src/adapters/openai-chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -660,6 +660,8 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
const pendingToolCalls: PendingToolCall[] = [];
let toolCallSeq = 0;
const flushToolCalls = function* (): Generator<AdapterEvent> {
// Do not treat flushed tool calls as user-facing output for the finish-less EOF
// fallback — incomplete tool args must stay on the truncation path.
for (const call of pendingToolCalls) {
if (!call.id) call.id = `call_${++toolCallSeq}`;
yield { type: "tool_call_start", id: call.id, name: call.name };
Expand All @@ -674,6 +676,9 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
// explicit `[DONE]` sentinel OR a chunk carrying a non-null `finish_reason` (some
// OpenAI-compatible providers omit `[DONE]` but do send finish_reason).
let finishReason: string | undefined;
// Only answer text enables the finish-less EOF fallback. Reasoning-only streams can be
// suppressed by hideThinkingSummary and must not complete as empty successful turns.
let sawUserFacingOutput = false;

// Single per-line handler shared by the streaming loop and the EOF residual-frame flush, so
// a final frame is parsed identically wherever it lands (no duplicated, drift-prone parsing).
Expand Down Expand Up @@ -743,6 +748,7 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
yield { type: "reasoning_raw_delta", text: delta.reasoning_content };
}
if (typeof delta.content === "string" && delta.content.length > 0) {
sawUserFacingOutput = true;
yield { type: "text_delta", text: delta.content };
}

Expand Down Expand Up @@ -800,21 +806,31 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
if (buffer.length > 0) {
if ((yield* handleDataLine(buffer)) === "terminate") return;
}
yield* flushToolCalls();
// Reader EOF. A graceful close shows at least one terminal signal: `[DONE]` (returns above),
// a non-null finish_reason (sawFinish), or a trailing usage chunk (providers emit usage only
// at end-of-generation). If NONE of those were seen, the stream was cut mid-flight — fail
// closed so the bridge emits a classified response.failed rather than a silent truncation.
// Reader EOF. Prefer failing closed before flushing pending tool calls so the bridge
// never sees a fabricated tool_call_end on a truncated mid-assembly stream.
const sawFinish = finishReason !== undefined;
if (!sawFinish && pendingUsage === undefined) {
const hadPendingToolCallsAtEof = pendingToolCalls.length > 0;
if (!sawFinish && hadPendingToolCallsAtEof) {
debugProviderDiagnostic("openai-chat", "stream-truncated", {
finishReason: finishReason ?? null,
hadUsage: false,
hadUsage: pendingUsage !== undefined,
pendingToolCalls: true,
});
yield { type: "error", message: "upstream stream ended without a terminal signal ([DONE] or finish_reason) - possible truncation" };
return;
}
// Finish-less EOF is only safe when answer text was emitted. Reasoning-only / usage-only
// truncations must stay on the error path (hideThinkingSummary can suppress reasoning).
if (!sawFinish && !sawUserFacingOutput) {
debugProviderDiagnostic("openai-chat", "stream-truncated", {
finishReason: finishReason ?? null,
hadUsage: pendingUsage !== undefined,
});
yield { type: "error", message: "upstream stream ended without a terminal signal ([DONE] or finish_reason) — possible truncation" };
return;
}
// Graceful close that omitted [DONE] but delivered finish_reason and/or final usage.
yield* flushToolCalls();
// Graceful close that omitted [DONE] but delivered finish_reason and/or answer text.
const stopReason = finishReason === "length"
? "max_tokens"
: finishReason === "content_filter"
Expand Down
54 changes: 50 additions & 4 deletions tests/openai-chat-eof.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,18 @@ async function collect(gen: AsyncGenerator<AdapterEvent>): Promise<AdapterEvent[
}

describe("openai-chat stream EOF fail-closed", () => {
test("truncated stream (no [DONE], no finish_reason) yields a terminal error, not a clean done", async () => {
test("truncated stream (no [DONE], no finish_reason) yields done when content was emitted", async () => {
const response = new Response('data: {"choices":[{"delta":{"content":"par"}}]}\n\n');
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
const last = events[events.length - 1];
expect(last.type).toBe("error");
expect(last.type).toBe("done");
expect(events.some(e => e.type === "error")).toBe(false);
});

test("empty EOF without content still errors", async () => {
const response = new Response("");
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("error");
expect(events.some(e => e.type === "done")).toBe(false);
});

Expand Down Expand Up @@ -128,10 +135,49 @@ describe("openai-chat stream EOF fail-closed", () => {
expect(events.some(e => e.type === "error")).toBe(false);
});

test("genuinely truncated stream WITHOUT a trailing newline still fails closed", async () => {
// Mid-content frame, no terminator, no newline — must remain a terminal error.
test("genuinely truncated stream WITHOUT a trailing newline completes when content was emitted", async () => {
// Mid-content frame, no terminator, no newline — content was yielded, so accept done.
const response = new Response('data: {"choices":[{"delta":{"content":"par"}}]}');
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("done");
expect(events.some(e => e.type === "error")).toBe(false);
});

test("EOF with pending tool calls and no finish_reason fails closed", async () => {
const response = new Response(
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{\\"a\\":"}}]}}]}\n\n',
);
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("error");
expect(events.some(e => e.type === "done")).toBe(false);
expect(events.some(e => e.type === "tool_call_end")).toBe(false);
});

test("reasoning-only EOF without finish_reason fails closed", async () => {
const response = new Response(
'data: {"choices":[{"delta":{"reasoning_content":"thinking..."}}]}\n\n',
);
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("error");
expect(events.some(e => e.type === "done")).toBe(false);
});

test("usage-only EOF with pending tool calls fails closed", async () => {
const response = new Response(
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{}"}}]}}]}\n\n' +
'data: {"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}\n\n',
);
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("error");
expect(events.some(e => e.type === "done")).toBe(false);
expect(events.some(e => e.type === "tool_call_end")).toBe(false);
});

test("usage-only EOF without answer text fails closed", async () => {
const response = new Response(
'data: {"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}\n\n',
);
const events = await collect(createOpenAIChatAdapter(provider).parseStream(response));
expect(events.at(-1)?.type).toBe("error");
expect(events.some(e => e.type === "done")).toBe(false);
});
Expand Down
Loading