Accumulate Codex stream output items so tool calls aren't lost
Root cause of the GPT-account "empty turn: no text and no tool calls" failures, found by replaying the real stored response: the CEO actually called hire_agent five times, but the ChatGPT/Codex backend delivers each output item in a response.output_item.done event and then sends a response.completed whose response.output is []. accumulateResponsesStream read only the completion event, so every tool call was discarded and the turn parsed as empty — the run failed and the company never staffed. Collect items from output_item.done and fall back to them when the completion carries no output of its own; keep using the completion output when it is populated (non-Codex Responses backends). Verified against the real captured stream (5 hires recovered) and unit-tested with the exact event shape. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
97242ae684
commit
cbb87171e6
2 files changed
+34
-7
No files matched your search
@@ -74,24 +74,37 @@ function errorText(data: JsonRecord): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Accumulate a Responses SSE stream (Codex/Copilot) into the final result. The
|
||||
* result is accepted from the completion event's `response.output`.
|
||||
* Accumulate a Responses SSE stream (Codex/Copilot) into the final result.
|
||||
*
|
||||
* The ChatGPT/Codex backend streams each output item in a
|
||||
* `response.output_item.done` event and then sends a `response.completed`
|
||||
* whose `response.output` is EMPTY. Reading only the completion event there
|
||||
* loses every tool call (the model acted, but the turn parses as empty). So
|
||||
* collect items as they finish and fall back to them when the completion
|
||||
* carries no output of its own.
|
||||
*/
|
||||
export async function accumulateResponsesStream(messages: AsyncIterable<ProviderEventStreamMessage>): Promise<AgentTurnResult> {
|
||||
const raw: string[] = [];
|
||||
const streamed: unknown[] = [];
|
||||
for await (const message of messages) {
|
||||
raw.push(message.rawData);
|
||||
const type = eventType(message);
|
||||
if (!type) continue;
|
||||
if (type === "error" || type === "response.failed") throw new Error(errorText(message.data));
|
||||
if (type === "response.output_item.done" && isRecord(message.data.item)) {
|
||||
streamed.push(message.data.item);
|
||||
}
|
||||
if (type === "response.completed" || type === "response.done" || type === "response.incomplete") {
|
||||
const response = isRecord(message.data.response) ? message.data.response : null;
|
||||
if (response && Array.isArray(response.output)) {
|
||||
const parsed = parseResponsesOutput(response);
|
||||
return { text: parsed.text, toolCalls: parsed.toolCalls, rawResponseJson: JSON.stringify(raw) };
|
||||
}
|
||||
throw new Error("Responses completion event was missing response.output.");
|
||||
const completionOutput = response && Array.isArray(response.output) ? response.output : [];
|
||||
const output = completionOutput.length > 0 ? completionOutput : streamed;
|
||||
const parsed = parseResponsesOutput({ output });
|
||||
return { text: parsed.text, toolCalls: parsed.toolCalls, rawResponseJson: JSON.stringify(raw) };
|
||||
}
|
||||
}
|
||||
if (streamed.length > 0) {
|
||||
const parsed = parseResponsesOutput({ output: streamed });
|
||||
return { text: parsed.text, toolCalls: parsed.toolCalls, rawResponseJson: JSON.stringify(raw) };
|
||||
}
|
||||
throw new Error("Responses stream ended before a completion event.");
|
||||
}
|
||||
@@ -60,6 +60,20 @@ describe("openai responses encoding", () => {
|
||||
]));
|
||||
expect(result.toolCalls).toEqual([{ id: "c3", name: "write_deliverable", arguments: { path: "a" } }]);
|
||||
});
|
||||
|
||||
test("accumulates output_item.done when the completion output is empty (Codex/ChatGPT backend)", async () => {
|
||||
// The Codex backend streams each tool call as an output_item.done event and
|
||||
// then sends a response.completed whose response.output is [].
|
||||
const result = await accumulateResponsesStream(stream([
|
||||
{ type: "response.output_item.added", item: { id: "fc1", type: "function_call", status: "in_progress", arguments: "", call_id: "call_1", name: "hire_agent" } },
|
||||
{ type: "response.function_call_arguments.done", arguments: '{"name":"Dr. Li"}', item_id: "fc1" },
|
||||
{ type: "response.output_item.done", item: { id: "fc1", type: "function_call", status: "completed", arguments: '{"name":"Dr. Li","department":"Engineering"}', call_id: "call_1", name: "hire_agent" } },
|
||||
{ type: "response.output_item.done", item: { id: "m1", type: "message", content: [{ type: "output_text", text: "Staffing the team." }] } },
|
||||
{ type: "response.completed", response: { output: [], status: "completed" } },
|
||||
]));
|
||||
expect(result.toolCalls).toEqual([{ id: "call_1", name: "hire_agent", arguments: { name: "Dr. Li", department: "Engineering" } }]);
|
||||
expect(result.text).toBe("Staffing the team.");
|
||||
});
|
||||
});
|
||||
|
||||
describe("openai chat completions encoding", () => {
|
||||
|
||||
Reference in new issue
Block a user