ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-28 05:16:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai-responses): reconcile terminal tool calls (#108461)

Browse Source 
* fix(ai): reconcile terminal Responses tool calls

Recover validated terminal tool calls through one completion owner when item-done events are missing. Reuse positional completion state, preserve exact arguments, and reject ambiguous or conflicting batches before tool execution. Also remove the shared empty identity alias for anonymous calls.

Fixes #108460. Builds on #122560 and preserves the original contribution from #108461.

Co-authored-by: snotty <snotty@users.noreply.github.com>

* refactor(ai): type terminal tool scratch state directly

* fix(ai): retain anonymous tool completion ownership

* chore(ai): prune resolved assertion baseline

* test(ai): deduplicate generated tool identity coverage

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
Co-authored-by: snotty <snotty@users.noreply.github.com>
This commit is contained in:
Mike
2026-08-27 11:39:37 +08:00
committed by GitHub
parent e593842b8a
commit 8c1a55d446
8 changed files with 389 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
+1 -1
View File
@@ -1542,7 +1542,7 @@ packages/ai/src/transports/openai-responses-payload-policy.ts 3
packages/ai/src/transports/openai-responses-prompt-observer-internal.ts 2
packages/ai/src/transports/openai-responses-replay-internal.ts 11
packages/ai/src/transports/openai-responses-replay-messages-internal.ts 15
packages/ai/src/transports/openai-responses-stream-internal.ts 3
packages/ai/src/transports/openai-responses-stream-internal.ts 2
packages/ai/src/transports/openai-responses-stream-observer-internal.ts 1
packages/ai/src/transports/openai-responses-stream-slots-internal.ts 2
packages/ai/src/transports/openai-responses-stream-terminal-internal.ts 2
packages/ai/src/providers/openai-responses-shared.test.ts
+3 -1
View File
@@ -2065,10 +2065,12 @@ describe("processResponsesStream", () => {
responseEvents([
{
type: "response.output_item.added",
output_index: 0,
item: { type: "function_call", name: "computer", arguments: "" },
},
{
type: "response.output_item.done",
output_index: 0,
item: { type: "function_call", name: "computer", arguments: "{}" },
},
{ type: "response.completed", response: { id: "resp_idless", status: "completed" } },
@@ -2473,7 +2475,7 @@ describe("processResponsesStream", () => {
stream,
nativeOpenAIModel,
),
).rejects.toThrow("Responses stream completed with unresolved tool calls");
).rejects.toThrow("Responses stream changed output item identity");
expect(events.map((event) => event.type)).toEqual(["toolcall_start", "toolcall_delta"]);
});
packages/ai/src/providers/openai-responses-tool-call-tracker.ts
+23 -4
View File
@@ -4,6 +4,7 @@ export type ResponsesToolCallIdentity = { itemId?: string; callId?: string };
export type ResponsesToolCallState = ResponsesToolCallIdentity & {
argumentStreamReliable: boolean;
outputIndex?: number;
};
type ResponsesToolCallEvent = {
@@ -58,10 +59,13 @@ export function createResponsesToolCallTracker<TState extends ResponsesToolCallS
const resolveCompatible = (
candidates: Iterable<TState>,
identity: ResponsesToolCallIdentity,
allowUnmatchedIdentity: boolean,
): TState | undefined => {
const uniqueCandidates = [...new Set(candidates)];
if (!identity.itemId && !identity.callId) {
return uniqueCandidates.length === 1 ? uniqueCandidates.at(0) : undefined;
return allowUnmatchedIdentity && uniqueCandidates.length === 1
? uniqueCandidates.at(0)
: undefined;
}
const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity));
const matches = compatible.filter((state) => sharesIdentity(state, identity));
@@ -73,7 +77,10 @@ export function createResponsesToolCallTracker<TState extends ResponsesToolCallS
// Only a sole active call may adopt an identity it did not already know.
// Parallel calls require a positive match so missing indices stay fail-closed.
const soleCompatible =
uniqueCandidates.length === 1 && compatible.length === 1 && matches.length === 0
allowUnmatchedIdentity &&
uniqueCandidates.length === 1 &&
compatible.length === 1 &&
matches.length === 0
? compatible.at(0)
: undefined;
return soleCompatible ? adoptIdentity(soleCompatible, identity) : undefined;
@@ -89,12 +96,14 @@ export function createResponsesToolCallTracker<TState extends ResponsesToolCallS
if (indexedCalls.has(outputIndex)) {
throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`);
}
state.outputIndex = outputIndex;
indexedCalls.set(outputIndex, state);
},
resolve(
event: ResponsesToolCallEvent,
identity: ResponsesToolCallIdentity = readEventIdentity(event),
allowUnmatchedIdentity = true,
): TState | undefined {
const outputIndex = readOutputIndex(event);
if (outputIndex !== undefined) {
@@ -110,15 +119,20 @@ export function createResponsesToolCallTracker<TState extends ResponsesToolCallS
// A compatibility stream may add calls without indices, then start
// including them. Bind only the one identity-matched (or sole) candidate.
const unindexed = resolveCompatible(unindexedCalls, identity);
const unindexed = resolveCompatible(unindexedCalls, identity, allowUnmatchedIdentity);
if (unindexed) {
unindexedCalls.delete(unindexed);
unindexed.outputIndex = outputIndex;
indexedCalls.set(outputIndex, unindexed);
}
return unindexed;
}
return resolveCompatible([...indexedCalls.values(), ...unindexedCalls], identity);
return resolveCompatible(
[...indexedCalls.values(), ...unindexedCalls],
identity,
allowUnmatchedIdentity,
);
},
forget(toolCall: TState): void {
@@ -141,5 +155,10 @@ export function createResponsesToolCallTracker<TState extends ResponsesToolCallS
hasActive(): boolean {
return indexedCalls.size > 0 || unindexedCalls.size > 0;
},
hasExactlyActive(expected: readonly TState[]): boolean {
const active = new Set([...indexedCalls.values(), ...unindexedCalls]);
return active.size === expected.length && expected.every((state) => active.has(state));
},
};
}
packages/ai/src/transports/openai-responses-stream-internal.ts
+106 -68
View File
@@ -56,6 +56,7 @@ export async function processResponsesStream<TApi extends Api>(
model: Model<TApi>,
options?: ResponsesStreamOptions,
) {
type CompletedToolCall = Extract<ResponseOutputItem, { type: "function_call" }>;
type StreamingToolCallBlock = ToolCall & { partialJson: string };
type StreamingToolCallState = ResponsesToolCallState & {
block: StreamingToolCallBlock;
@@ -176,19 +177,99 @@ export async function processResponsesStream<TApi extends Api>(
}
}
};
const { finalizeResponse, finalizeFailedResponse, recoverTerminalOutput } =
createResponsesTerminalController({
output,
stream,
model,
options,
outputs,
getLastTextBlock: () => lastTextBlock,
setLastTextBlock: (block) => {
lastTextBlock = block;
},
markFinalized: () => undefined,
});
const terminal = createResponsesTerminalController({
output,
stream,
model,
options,
outputs,
getLastTextBlock: () => lastTextBlock,
setLastTextBlock: (block) => {
lastTextBlock = block;
},
});
const finalizeToolCall = (
item: CompletedToolCall,
outputIndex: number | undefined,
streamingToolCall: StreamingToolCallState | undefined,
validated: Pick<ToolCall, "name" | "arguments">,
): void => {
const identity = {
type: item.type,
id: item.id || streamingToolCall?.itemId,
call_id: item.call_id || streamingToolCall?.callId,
};
const finalOutputIndex = outputIndex ?? streamingToolCall?.outputIndex;
// A wholly anonymous, unindexed done event cannot be deduplicated. Keep
// its active owner until the terminal snapshot supplies an output position.
if (finalOutputIndex === undefined && !identity.id && !identity.call_id) {
if (!streamingToolCall) {
throw new Error("Responses stream completed tool call without an output identity");
}
return;
}
if (streamingToolCall) {
streamingToolCalls.forget(streamingToolCall);
for (const slot of outputSlots.values()) {
if (slot.type === "toolCall" && slot.toolCall === streamingToolCall) {
outputSlots.forget(slot);
}
}
}
terminal.emitToolCallCompletion(identity, finalOutputIndex, streamingToolCall, validated);
};
const prepareTerminalToolCalls = (items: ResponseOutputItem[]) => {
const prepared = new Map<number, () => void>();
const recovered: StreamingToolCallState[] = [];
const callIds = new Set<string>();
const allowUnmatchedIdentity =
items.filter(
(item, index) => item.type === "function_call" && !outputs.get(item, index)?.completed,
).length === 1;
for (const [outputIndex, item] of items.entries()) {
const tracked = outputs.get(item, outputIndex);
if (item.type !== "function_call") {
continue;
}
if (item.call_id && callIds.has(item.call_id)) {
throw new Error("Responses stream repeated a terminal tool-call identity");
}
if (item.call_id) {
callIds.add(item.call_id);
}
// Completed positions must be skipped before resolve can adopt an
// unindexed active call. The positional tracker still checks identity.
if (tracked?.completed) {
continue;
}
const state = streamingToolCalls.resolve(
{ output_index: outputIndex },
readResponsesToolCallItemIdentity(item),
allowUnmatchedIdentity,
);
if (tracked && !state) {
throw new Error("Responses stream completed with unresolved tool calls");
}
const validated = resolveCompletedResponsesToolCall(item, { name: state?.block.name });
if (state) {
recovered.push(state);
}
prepared.set(outputIndex, () => finalizeToolCall(item, outputIndex, state, validated));
}
if (!streamingToolCalls.hasExactlyActive(recovered)) {
throw new Error("Responses stream completed with unresolved tool calls");
}
// All terminal calls and active-call coverage are validated before any
// toolcall_end can authorize execution; terminal ordering is checked next.
return (outputIndex: number) => {
const complete = prepared.get(outputIndex);
if (!complete) {
throw new Error("Responses stream completed with unresolved tool calls");
}
complete();
};
};
const guardedStream = adaptResponsesStream(
withFirstStreamEventTimeout(openaiStream, {
@@ -548,6 +629,9 @@ export async function processResponsesStream<TApi extends Api>(
}
outputSlots.forget(outputSlot);
} else if (item.type === "function_call") {
if (outputs.get(item, readResponsesOutputIndex(event))?.completed) {
continue;
}
const streamingToolCall = streamingToolCalls.resolve(
event,
readResponsesToolCallItemIdentity(item),
@@ -557,7 +641,6 @@ export async function processResponsesStream<TApi extends Api>(
if (!streamingToolCall && streamingToolCalls.hasActive()) {
continue;
}
const streamedArguments = streamingToolCall?.block.partialJson ?? "";
const completedArguments =
typeof item.arguments === "string" ? item.arguments : undefined;
if (
@@ -567,68 +650,23 @@ export async function processResponsesStream<TApi extends Api>(
) {
continue;
}
const finalArguments =
completedArguments !== undefined &&
(completedArguments.length > 0 || !streamedArguments)
? completedArguments
: streamedArguments;
const validated = resolveCompletedResponsesToolCall(item, {
name: streamingToolCall?.block.name,
arguments: finalArguments,
arguments: completedArguments || streamingToolCall?.block.partialJson || "",
});
let toolCall: ToolCall;
let contentIndex: number;
if (streamingToolCall) {
const block = streamingToolCall.block;
// The SDK permits the added item to omit its item id, then supplies
// the canonical id on completion. Upgrade the same public block so
// replay and its function_call_output retain both identities.
block.id = resolveResponsesToolCallId(item, block.id);
block.name = validated.name;
// Finalize in-place and strip the scratch buffer so replay only
// carries parsed arguments.
block.arguments = validated.arguments;
delete (block as { partialJson?: string }).partialJson;
toolCall = block;
contentIndex = streamingToolCall.contentIndex;
} else {
toolCall = {
type: "toolCall",
id: resolveResponsesToolCallId(item),
name: validated.name,
arguments: validated.arguments,
};
// Some compatible streams only send the completed item. Preserve
// the normal balanced lifecycle and persist the call for replay.
blocks.push(toolCall);
contentIndex = blocks.length - 1;
stream.push({ type: "toolcall_start", contentIndex, partial: output });
}
if (streamingToolCall) {
streamingToolCalls.forget(streamingToolCall);
for (const slot of outputSlots.values()) {
if (slot.type === "toolCall" && slot.toolCall === streamingToolCall) {
outputSlots.forget(slot);
}
}
}
stream.push({
type: "toolcall_end",
contentIndex,
toolCall,
partial: output,
});
outputs.set(item, contentIndex, readResponsesOutputIndex(event), true);
finalizeToolCall(item, readResponsesOutputIndex(event), streamingToolCall, validated);
}
} else if (event.type === "response.completed" || event.type === "response.incomplete") {
if (streamingToolCalls.hasActive()) {
if (event.type === "response.incomplete" && streamingToolCalls.hasActive()) {
throw new Error("Responses stream completed with unresolved tool calls");
}
finalizeResponse(event.response, event.type);
terminal.finalizeResponse(event.response, event.type);
if (event.type === "response.completed" || output.stopReason === "length") {
recoverTerminalOutput(event.response.output ?? [], event.type === "response.completed");
const items = event.response.output ?? [];
const completeToolCall =
event.type === "response.completed" ? prepareTerminalToolCalls(items) : undefined;
terminal.recoverTerminalOutput(items, completeToolCall);
}
terminalResponse = event.type === "response.completed" ? event.response : null;
if (
@@ -644,7 +682,7 @@ export async function processResponsesStream<TApi extends Api>(
);
} else if (event.type === "response.failed") {
const failure = normalizeResponsesFailedEvent(isRecord(event) ? event : {}, model);
finalizeFailedResponse(event.response, failure.responseId);
terminal.finalizeFailedResponse(event.response, failure.responseId);
throw new ResponsesStreamFailure(failure, event.response);
}
}
packages/ai/src/transports/openai-responses-stream-slots-internal.ts
+2 -3
View File
@@ -51,9 +51,8 @@ export function createResponsesOutputTracker() {
if ((item.type === "reasoning" || item.type === "message") && item.id) {
return `${item.type}:${item.id}`;
}
return item.type === "function_call"
? `function_call:${item.call_id ?? item.id ?? ""}`
: undefined;
const callId = item.call_id ?? item.id;
return item.type === "function_call" && callId ? `function_call:${callId}` : undefined;
};
const get = (item: ResponsesOutputIdentityItem, outputIndex?: number) => {
const key = identity(item);
packages/ai/src/transports/openai-responses-stream-terminal-internal.ts
+33 -23
View File
@@ -113,7 +113,6 @@ export function createResponsesTerminalController(params: {
outputs: ResponsesOutputTracker;
getLastTextBlock: () => TextBlockReference | null;
setLastTextBlock: (block: TextBlockReference | null) => void;
markFinalized: () => void;
}) {
const { output, stream, model, options } = params;
const blocks = output.content;
@@ -204,21 +203,31 @@ export function createResponsesTerminalController(params: {
stream.push({ type: "text_end", contentIndex: index, content: text, partial: output });
return index;
};
const appendToolCall = (item: Extract<ResponseOutputItem, { type: "function_call" }>): number => {
const validated = resolveCompletedResponsesToolCall(item);
const toolCall: ToolCall = {
type: "toolCall",
id: resolveResponsesToolCallId(item),
name: validated.name,
arguments: validated.arguments,
};
blocks.push(toolCall);
const contentIndex = blocks.length - 1;
stream.push({ type: "toolcall_start", contentIndex, partial: output });
const emitToolCallCompletion = (
item: { type: "function_call"; id?: string; call_id?: string },
outputIndex: number | undefined,
started: { block: ToolCall; contentIndex: number } | undefined,
validated: Pick<ToolCall, "name" | "arguments">,
): void => {
// Complete the same public block with authoritative identities and arguments;
// scratch JSON must never survive into transcript replay.
const completed = { id: resolveResponsesToolCallId(item, started?.block.id), ...validated };
const toolCall: ToolCall & { partialJson?: string } = started
? Object.assign(started.block, completed)
: { type: "toolCall", ...completed };
delete toolCall.partialJson;
const contentIndex = started?.contentIndex ?? blocks.length;
if (!started) {
blocks.push(toolCall);
stream.push({ type: "toolcall_start", contentIndex, partial: output });
}
params.outputs.set(item, contentIndex, outputIndex, true);
stream.push({ type: "toolcall_end", contentIndex, toolCall, partial: output });
return contentIndex;
};
const recoverTerminalOutput = (items: ResponseOutputItem[], includeToolCalls: boolean) => {
const recoverTerminalOutput = (
items: ResponseOutputItem[],
completeToolCall?: (outputIndex: number) => void,
) => {
let hasCompletedLaterOutput = false;
for (const [outputIndex, item] of [...items.entries()].toReversed()) {
const tracked = params.outputs.get(item, outputIndex);
@@ -234,16 +243,13 @@ export function createResponsesTerminalController(params: {
hasCompletedLaterOutput = true;
continue;
}
if (item.type === "function_call" && !includeToolCalls) {
if (item.type === "function_call" && !completeToolCall) {
continue;
}
// Previously emitted content indexes cannot be reordered after a missing earlier item.
if (hasCompletedLaterOutput) {
throw new Error("Responses stream omitted an output item before completed output");
}
if (item.type === "function_call") {
resolveCompletedResponsesToolCall(item);
}
}
for (const [terminalIndex, item] of items.entries()) {
if (item.type === "message") {
@@ -281,11 +287,11 @@ export function createResponsesTerminalController(params: {
model,
options?.reasoningReplayMetadata,
);
} else if (includeToolCalls && item.type === "function_call") {
if (params.outputs.get(item, terminalIndex)) {
} else if (completeToolCall && item.type === "function_call") {
if (params.outputs.get(item, terminalIndex)?.completed) {
continue;
}
params.outputs.set(item, appendToolCall(item), terminalIndex, true);
completeToolCall(terminalIndex);
}
}
}
@@ -323,7 +329,6 @@ export function createResponsesTerminalController(params: {
>["response"],
terminalEventType: "response.completed" | "response.incomplete",
) => {
params.markFinalized();
backfillReasoning(response.output ?? []);
finalizeTerminalFacts(response);
const terminal = resolveResponsesTerminalStopReason({
@@ -335,5 +340,10 @@ export function createResponsesTerminalController(params: {
output.stopReason = terminal.stopReason;
output.errorMessage = terminal.errorMessage;
};
return { finalizeResponse, finalizeFailedResponse: finalizeTerminalFacts, recoverTerminalOutput };
return {
finalizeResponse,
finalizeFailedResponse: finalizeTerminalFacts,
recoverTerminalOutput,
emitToolCallCompletion,
};
}
packages/ai/src/transports/openai-responses-stream-terminal-tools.test.ts
+221
View File
@@ -0,0 +1,221 @@
import { describe, expect, it } from "vitest";
import { completed, runFixture } from "./openai-responses-stream-parity.test-helpers.js";
const tool = (slot: number, overrides: Record<string, unknown> = {}) => ({
type: "function_call",
id: `fc_${slot}`,
call_id: `call_${slot}`,
name: "lookup",
arguments: JSON.stringify({ slot }),
status: "completed",
...overrides,
});
const added = (slot: number, overrides: Record<string, unknown> = {}) => ({
type: "response.output_item.added",
output_index: slot,
item: tool(slot, { arguments: "", status: "in_progress", ...overrides }),
});
describe("Responses terminal tool completion", () => {
it("does not repeat an anonymous unindexed call after its item-done event", async () => {
const anonymous = { id: undefined, call_id: undefined };
const result = await runFixture([
{ ...added(0, anonymous), output_index: undefined },
{ type: "response.output_item.done", item: tool(0, anonymous) },
completed("resp_anonymous_done", [tool(0, anonymous)]),
]);
expect(result.error).toBeNull();
expect(result.content).toHaveLength(1);
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([
{ type: "toolcall_end", contentIndex: 0 },
]);
});
it("rejects an anonymous call with no stream or terminal position without completing it", async () => {
const anonymous = { id: undefined, call_id: undefined };
const result = await runFixture([
{ ...added(0, anonymous), output_index: undefined },
{ type: "response.output_item.done", item: tool(0, anonymous) },
completed("resp_no_position", []),
]);
expect(result.error).toBe("Responses stream completed with unresolved tool calls");
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
});
it.each(["indexed", "identified"])(
"retains a known %s owner when its done event omits identity",
async (identity) => {
const anonymous = { id: undefined, call_id: undefined };
const result = await runFixture([
identity === "indexed" ? added(0, anonymous) : { ...added(0), output_index: undefined },
{ type: "response.output_item.done", item: tool(0, anonymous) },
completed("resp_known_owner", []),
]);
expect(result.error).toBeNull();
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([
{ type: "toolcall_end", contentIndex: 0 },
]);
},
);
it("rejects an anonymous unindexed done-only call without silently dropping it", async () => {
const item = tool(0, { id: undefined, call_id: undefined });
const result = await runFixture([
{ type: "response.output_item.done", item },
completed("resp_done_only", [item]),
]);
expect(result.error).toBe("Responses stream completed tool call without an output identity");
expect(result.content).toHaveLength(0);
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
});
it("does not publish anonymous unindexed completions before ambiguous terminal matching", async () => {
const anonymous = { id: undefined, call_id: undefined };
const result = await runFixture([
{ ...added(0, anonymous), output_index: undefined },
{ type: "response.output_item.done", item: tool(0, anonymous) },
{ ...added(1, anonymous), output_index: undefined },
{ type: "response.output_item.done", item: tool(1, anonymous) },
completed("resp_ambiguous_done", [tool(0), tool(1)]),
]);
expect(result.error).not.toBeNull();
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
});
it.each(["indexed", "rotated", "unindexed", "anonymous"])(
"completes a %s call exactly once when its item-done event is missing",
async (identity) => {
const anonymous = identity === "anonymous" ? { id: undefined, call_id: undefined } : {};
const item = tool(0, {
...anonymous,
...(identity === "rotated" ? { id: "fc_terminal_rotated" } : {}),
arguments: '{"slot":0,"id":9007199254740993}',
});
const result = await runFixture([
{
...added(0, anonymous),
...(identity === "unindexed" || identity === "anonymous"
? { output_index: undefined }
: {}),
},
completed("resp_missing_done", [item]),
]);
expect(result.error).toBeNull();
expect(result.stopReason).toBe("toolUse");
expect(result.events).toEqual([
{ type: "toolcall_start", contentIndex: 0 },
{ type: "toolcall_end", contentIndex: 0 },
]);
expect(result.content).toEqual([
{
type: "toolCall",
id:
identity === "anonymous"
? "call_<generated>"
: `call_0|${identity === "rotated" ? "fc_terminal_rotated" : "fc_0"}`,
name: "lookup",
arguments: { slot: 0, id: "9007199254740993" },
partialJson: false,
},
]);
},
);
it("does not adopt an unindexed anonymous call into an already completed position", async () => {
const anonymous = { id: undefined, call_id: undefined };
const result = await runFixture([
{ type: "response.output_item.done", output_index: 0, item: tool(0, anonymous) },
{ ...added(1, anonymous), output_index: undefined },
completed("resp_anonymous", [tool(0, anonymous), tool(1, anonymous)]),
]);
expect(result.error).toBeNull();
expect(
result.content.map((block) => (block.type === "toolCall" ? block.arguments : null)),
).toEqual([{ slot: 0 }, { slot: 1 }]);
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([
{ type: "toolcall_end", contentIndex: 0 },
{ type: "toolcall_end", contentIndex: 1 },
]);
});
it.each([
["malformed arguments", { arguments: '{"slot":' }],
["non-object arguments", { arguments: "[]" }],
["incomplete status", { status: "incomplete" }],
["changed name", { name: "delete_record" }],
["changed call identity", { call_id: "call_conflicting" }],
])(
"rejects a terminal batch with later %s before any tool completes",
async (_name, override) => {
const result = await runFixture([
added(0),
added(1),
completed("resp_invalid_batch", [tool(0), tool(1, override)]),
]);
expect(result.error).not.toBeNull();
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
},
);
it("rejects a changed completed call identity before another active call completes", async () => {
const result = await runFixture([
{ type: "response.output_item.done", output_index: 0, item: tool(0) },
added(1),
completed("resp_completed_conflict", [tool(0, { call_id: "call_conflicting" }), tool(1)]),
]);
expect(result.error).toBe("Responses stream changed output item identity");
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([
{ type: "toolcall_end", contentIndex: 0 },
]);
});
it.each([
["missing call", []],
["duplicate call", [tool(0), tool(0)]],
["unmatched call", [tool(0, { call_id: "call_other" })]],
])("rejects a terminal %s without completing the active call", async (_name, items) => {
const result = await runFixture([added(0), completed("resp_unresolved", items)]);
expect(result.error).not.toBeNull();
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
});
it("never completes active tools from an incomplete response", async () => {
const result = await runFixture([
added(0),
{
type: "response.incomplete",
response: { id: "resp_incomplete", status: "incomplete", output: [tool(0)] },
},
]);
expect(result.error).toBe("Responses stream completed with unresolved tool calls");
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
});
it.each([1, 2])(
"rejects %i anonymous unindexed calls with ambiguous terminal positions",
async (count) => {
const result = await runFixture([
...Array.from({ length: count }, (_, slot) => ({
...added(slot, { id: undefined, call_id: undefined }),
output_index: undefined,
})),
completed("resp_ambiguous", [tool(0), tool(1)]),
]);
expect(result.error).not.toBeNull();
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([]);
},
);
it("completes indexed anonymous and terminal-only calls without duplicate callbacks", async () => {
const result = await runFixture([
added(0, { id: undefined, call_id: undefined }),
completed("resp_indexed_and_new", [tool(0), tool(1)]),
]);
expect(result.error).toBeNull();
expect(result.content).toHaveLength(2);
expect(result.events.filter((event) => event.type === "toolcall_end")).toEqual([
{ type: "toolcall_end", contentIndex: 0 },
{ type: "toolcall_end", contentIndex: 1 },
]);
});
});
src/agents/openai-transport-stream.streaming.test.ts
-41
View File
@@ -378,47 +378,6 @@ describe("openai transport stream", () => {
]);
});
it("keeps idless Responses tool-call ids stable and response-unique", async () => {
const runOnce = async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
const events: CapturedStreamEvent[] = [];
await testing.processResponsesStream(
streamChunks([
{
type: "response.output_item.added",
item: { type: "function_call", name: "computer", arguments: "" },
},
{
type: "response.output_item.done",
item: { type: "function_call", name: "computer", arguments: "{}" },
},
{ type: "response.completed", response: { id: "resp_idless", status: "completed" } },
]),
output,
{ push: (event) => events.push(event as CapturedStreamEvent) },
model,
);
const block = output.content.find((entry) => entry.type === "toolCall") as
| { id?: string }
| undefined;
const end = events.find((event) => event.type === "toolcall_end") as
| { toolCall?: { id?: string } }
| undefined;
if (!block?.id || !end?.toolCall?.id) {
throw new Error("missing tool-call lifecycle");
}
return { blockId: block.id, endId: end.toolCall.id };
};
const first = await runOnce();
const second = await runOnce();
expect(first.blockId).toMatch(/^call_[0-9a-f]{24}$/);
expect(first.endId).toBe(first.blockId);
expect(second.endId).toBe(second.blockId);
expect(second.blockId).not.toBe(first.blockId);
});
it("materializes one stable tool call for a done-only idless Responses item", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);