From 0970b84336c6027f3801ccaeedcf8f56f40402a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 13:35:53 -0400 Subject: [PATCH] Q80: turns and replies the provider cut short fail instead of passing as finished Co-Authored-By: Claude Opus 5.5 --- src/main/services/provider-adapters.ts | 7 +- src/main/services/provider-response-runner.ts | 9 +- src/main/services/provider-streaming.ts | 26 +++++- src/main/services/provider-tool-turn.ts | 18 ++++ .../provider-tool-turns/anthropic-format.ts | 3 +- .../provider-tool-turns/bedrock-format.ts | 3 +- .../provider-tool-turns/chat-format.ts | 6 +- .../provider-tool-turns/gemini-format.ts | 6 +- .../provider-tool-turns/responses-format.ts | 21 +++-- tests/unit/incomplete-turns.test.ts | 84 +++++++++++++++++++ tests/unit/provider-service-streaming.test.ts | 28 +++++++ tests/unit/provider-tool-turns.test.ts | 14 ++-- tests/unit/reasoning-replay.test.ts | 18 ++-- tests/unit/stream-usage-request.test.ts | 2 +- 14 files changed, 208 insertions(+), 37 deletions(-) create mode 100644 tests/unit/incomplete-turns.test.ts diff --git a/src/main/services/provider-adapters.ts b/src/main/services/provider-adapters.ts index eed8400b..394ed2c4 100644 --- a/src/main/services/provider-adapters.ts +++ b/src/main/services/provider-adapters.ts @@ -25,6 +25,8 @@ import { authorizeWithParser, refreshUnsupported, revokeLocalProfile } from "./p import { getJson, postEventStream, postJson, type JsonRecord } from "./provider-http"; import { fetchAnthropicModels, fetchOpenAiModels } from "./provider-models-fetch"; import { KIMI_CODE_BASE_URL } from "./oauth/kimi-code-flow"; +import { FINISHED_STOP_REASONS, incompleteTurn, requireFinished } from "./provider-tool-turn"; +import { incompleteReason } from "./provider-tool-turns/responses-format"; const ANTHROPIC_VERSION = "2023-06-01"; const CLAUDE_PRO_BETA = "claude-code-20250219,oauth-2025-04-20,fine-grained-tool-streaming-2025-05-14"; @@ -124,6 +126,7 @@ export const providerAdapters: ProviderAdapterRegistry = { ...(input.instructions ? { instructions: input.instructions } : {}), }); const data = response.data; + if (data.status !== "completed") throw incompleteTurn(incompleteReason(data)); const text = requireText(collectText(data.output_text ?? data.output), "OpenAI"); return { text, rawResponseJson: JSON.stringify(data), rateLimitState: response.rateLimitState }; }, @@ -177,6 +180,7 @@ export const providerAdapters: ProviderAdapterRegistry = { messages: [{ role: "user", content: input.prompt }], }); const data = response.data; + requireFinished(data.stop_reason, FINISHED_STOP_REASONS.anthropic); const text = requireText(collectText(data.content), "Anthropic"); return { text, rawResponseJson: JSON.stringify(data), rateLimitState: response.rateLimitState }; }, @@ -416,7 +420,8 @@ function openAiCompatibleAdapter({ messages, }); const data = response.data; - const text = requireText(collectText(data.choices), label); + requireFinished(Array.isArray(data.choices) && isRecord(data.choices[0]) ? data.choices[0].finish_reason : null, FINISHED_STOP_REASONS.chat); + const text = requireText(collectText(Array.isArray(data.choices) ? data.choices.filter(isRecord).map((choice) => choice.message) : []), label); return { text, rawResponseJson: JSON.stringify(data), rateLimitState: response.rateLimitState }; }, }; diff --git a/src/main/services/provider-response-runner.ts b/src/main/services/provider-response-runner.ts index 150b1f28..bc76e94c 100644 --- a/src/main/services/provider-response-runner.ts +++ b/src/main/services/provider-response-runner.ts @@ -12,7 +12,7 @@ import type { ProviderRateLimitRegistry } from "./provider-rate-limit-registry"; import type { ProviderRequestStore } from "./provider-request-store"; import type { ProviderResponseObserver } from "./provider-response-observer"; import { hashPrompt, readProviderResponseId, requireStreamText } from "./provider-response-utils"; -import { type AgentTurnMessage, type AgentTurnResult, type ToolDefinition, type ToolTurnFn } from "./provider-tool-turn"; +import { incompleteTurn, type AgentTurnMessage, type AgentTurnResult, type ToolDefinition, type ToolTurnFn } from "./provider-tool-turn"; import { selectToolTurn, supportsToolTurn } from "./provider-tool-turns/router"; import { resolveTurnCredential } from "./provider-tool-turns/credential-resolver"; import { currentRunId } from "./run-control"; @@ -201,6 +201,7 @@ export class ProviderResponseRunner { const rawEvents: string[] = []; const textParts: string[] = []; let providerResponseId: string | null = null; + let finished = false; try { const authProfile = execution.adapter.parseAuthProfile(execution.credential.secret); const result = await execution.adapter.streamResponse({ @@ -214,6 +215,7 @@ export class ProviderResponseRunner { if (event.type === "error") throw new Error(event.textDelta ?? "Provider stream failed."); rawEvents.push(event.rawEventJson); if (event.type === "usage") continue; + if (event.type === "message_stop") finished = true; if (event.providerResponseId) providerResponseId = event.providerResponseId; if (event.textDelta) textParts.push(event.textDelta); const responseText = textParts.join(""); @@ -244,8 +246,9 @@ export class ProviderResponseRunner { streamEventCount, }); } + if (!finished) throw incompleteTurn(null); const responseText = requireStreamText(textParts); - const finished = this.deps.requestStore.markSucceeded({ + const succeeded = this.deps.requestStore.markSucceeded({ orgId: input.orgId, requestId, responseText, @@ -254,7 +257,7 @@ export class ProviderResponseRunner { externalResponseId: providerResponseId, completedAt: new Date().toISOString(), }); - if (!finished) return requestId; + if (!succeeded) return requestId; this.deps.responseObserver.recordSuccess({ orgId: input.orgId, provider: execution.provider, diff --git a/src/main/services/provider-streaming.ts b/src/main/services/provider-streaming.ts index 2364a837..6c295914 100644 --- a/src/main/services/provider-streaming.ts +++ b/src/main/services/provider-streaming.ts @@ -1,6 +1,8 @@ import type { ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; import type { JsonRecord, ProviderEventStreamMessage } from "./provider-http"; import { isRecord } from "./provider-http"; +import { FINISHED_STOP_REASONS, incompleteTurn } from "./provider-tool-turn"; +import { incompleteReason } from "./provider-tool-turns/responses-format"; export interface ProviderStreamEvent { /** usage: a chunk that only reports token usage; it is recorded, never shown. */ @@ -41,7 +43,12 @@ export async function* mapOpenAiResponseStream(messages: AsyncIterable typeof choice.finish_reason === "string" && choice.finish_reason.length > 0); +function chatFinishReason(data: JsonRecord): string | null { + const reason = recordArray(data.choices).map((choice) => choice.finish_reason).find((value) => typeof value === "string" && value.length > 0); + return typeof reason === "string" ? reason : null; } function readAnthropicTextDelta(data: JsonRecord): string | null { diff --git a/src/main/services/provider-tool-turn.ts b/src/main/services/provider-tool-turn.ts index c597397d..2580749f 100644 --- a/src/main/services/provider-tool-turn.ts +++ b/src/main/services/provider-tool-turn.ts @@ -79,6 +79,24 @@ export interface ProviderToolTurnInput { export type ToolTurnFn = (input: ProviderToolTurnInput) => Promise; +/** Stop reasons that mean the model finished its turn, per wire format, as Pi maps them. */ +export const FINISHED_STOP_REASONS = { + chat: ["stop", "end", "tool_calls", "function_call"], + anthropic: ["end_turn", "tool_use", "stop_sequence"], + bedrock: ["end_turn", "tool_use", "stop_sequence"], + gemini: ["STOP"], +} as const; + +/** The error for a turn the provider cut short; acting on half a turn's text or arguments is worse than failing. */ +export function incompleteTurn(reason: unknown): Error { + const named = typeof reason === "string" && reason ? reason : "no stop reason"; + return new Error(`The provider stopped before finishing its answer (${named}); the partial answer was not used.`); +} + +export function requireFinished(reason: unknown, finished: readonly string[]): void { + if (typeof reason !== "string" || !finished.includes(reason)) throw incompleteTurn(reason); +} + /** Coerce a tool-call argument payload (string or object) into a JSON object. */ export function parseArguments(value: unknown): JsonRecord { if (isRecord(value)) return value; diff --git a/src/main/services/provider-tool-turns/anthropic-format.ts b/src/main/services/provider-tool-turns/anthropic-format.ts index 0e28f3f3..2b3b1e17 100644 --- a/src/main/services/provider-tool-turns/anthropic-format.ts +++ b/src/main/services/provider-tool-turns/anthropic-format.ts @@ -1,6 +1,6 @@ /** Anthropic Messages API encoding/decoding, shared by anthropic + claude-pro. */ import { isRecord, type JsonRecord } from "../provider-http"; -import { replayIn, type AgentTurnMessage, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; +import { FINISHED_STOP_REASONS, replayIn, requireFinished, type AgentTurnMessage, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; export function anthropicToolDefs(tools: ToolDefinition[]): unknown[] { return tools.map((tool) => ({ name: tool.name, description: tool.description, input_schema: tool.parameters })); @@ -27,6 +27,7 @@ export function toAnthropicMessages(messages: AgentTurnMessage[]): unknown[] { } export function parseAnthropicContent(data: JsonRecord): { text: string; toolCalls: ToolCall[]; replay: AssistantReplay } { + requireFinished(data.stop_reason, FINISHED_STOP_REASONS.anthropic); const content = Array.isArray(data.content) ? data.content : []; const textParts: string[] = []; const toolCalls: ToolCall[] = []; diff --git a/src/main/services/provider-tool-turns/bedrock-format.ts b/src/main/services/provider-tool-turns/bedrock-format.ts index e3bead5d..1b828f35 100644 --- a/src/main/services/provider-tool-turns/bedrock-format.ts +++ b/src/main/services/provider-tool-turns/bedrock-format.ts @@ -1,6 +1,6 @@ /** AWS Bedrock Converse API encoding/decoding (ported from Pi's bedrock handler). */ import { isRecord, type JsonRecord } from "../provider-http"; -import { replayIn, type AgentTurnMessage, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; +import { FINISHED_STOP_REASONS, replayIn, requireFinished, type AgentTurnMessage, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; export function bedrockToolConfig(tools: ToolDefinition[]): JsonRecord { return { @@ -30,6 +30,7 @@ export function toBedrockMessages(messages: AgentTurnMessage[]): JsonRecord[] { } export function parseBedrockOutput(data: JsonRecord): { text: string; toolCalls: ToolCall[]; replay: AssistantReplay } { + requireFinished(data.stopReason, FINISHED_STOP_REASONS.bedrock); const output = isRecord(data.output) ? data.output : null; const message = output && isRecord(output.message) ? output.message : null; const content = message && Array.isArray(message.content) ? message.content : []; diff --git a/src/main/services/provider-tool-turns/chat-format.ts b/src/main/services/provider-tool-turns/chat-format.ts index 22578079..7af27db2 100644 --- a/src/main/services/provider-tool-turns/chat-format.ts +++ b/src/main/services/provider-tool-turns/chat-format.ts @@ -1,6 +1,6 @@ /** OpenAI Chat Completions encoding/decoding (xAI, Groq, Mistral, Cerebras, OpenRouter, Kimi family). */ import { isRecord, ProviderHttpError, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http"; -import { parseArguments, replayIn, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; +import { FINISHED_STOP_REASONS, parseArguments, replayIn, requireFinished, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; /** Fields providers use for reasoning text; the first non-empty one is the provider's, as in Pi. */ const REASONING_FIELDS = ["reasoning_content", "reasoning", "reasoning_text"] as const; @@ -35,6 +35,7 @@ export function toChatMessages(instructions: string | null, messages: AgentTurnM export function parseChatChoice(data: JsonRecord): { text: string; toolCalls: ToolCall[]; replay?: AssistantReplay } { const choices = Array.isArray(data.choices) ? data.choices : []; const first = choices.length > 0 && isRecord(choices[0]) ? choices[0] : null; + requireFinished(first?.finish_reason, FINISHED_STOP_REASONS.chat); const message = first && isRecord(first.message) ? first.message : null; if (!message) return { text: "", toolCalls: [] }; const text = typeof message.content === "string" ? message.content : ""; @@ -78,11 +79,13 @@ export async function accumulateChatStream(messages: AsyncIterable(); const reasoning = { field: "", text: "" }; const details: JsonRecord[] = []; + let finishReason: unknown = null; for await (const message of messages) { raw.push(message.rawData); if (message.data.error !== undefined) throw new ProviderHttpError(streamErrorText(message.data.error), null, streamErrorStatus(message.data.error)); const choices = Array.isArray(message.data.choices) ? message.data.choices : []; const first = choices.length > 0 && isRecord(choices[0]) ? choices[0] : null; + if (first && typeof first.finish_reason === "string") finishReason = first.finish_reason; const delta = first && isRecord(first.delta) ? first.delta : null; if (!delta) continue; if (typeof delta.content === "string") textParts.push(delta.content); @@ -102,6 +105,7 @@ export async function accumulateChatStream(messages: AsyncIterable a - b)) { if (fragment.id && fragment.name) toolCalls.push({ id: fragment.id, name: fragment.name, arguments: parseArguments(fragment.args) }); diff --git a/src/main/services/provider-tool-turns/gemini-format.ts b/src/main/services/provider-tool-turns/gemini-format.ts index 7edfd819..203cef3c 100644 --- a/src/main/services/provider-tool-turns/gemini-format.ts +++ b/src/main/services/provider-tool-turns/gemini-format.ts @@ -4,7 +4,7 @@ * Antigravity request wrapper in google-gemini-cli. */ import { isRecord, ProviderHttpError, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http"; -import { replayIn, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; +import { FINISHED_STOP_REASONS, replayIn, requireFinished, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; const ANTIGRAVITY_SYSTEM_INSTRUCTION = "You are Antigravity, a powerful agentic AI coding assistant designed by the Google Deepmind team working on Advanced Agentic Coding." + @@ -127,6 +127,7 @@ function partsToResult(parts: unknown[]): { text: string[]; toolCalls: ToolCall[ /** Parse a non-streaming Gemini generateContent response (Vertex). */ export function parseGeminiResponse(data: JsonRecord): { text: string; toolCalls: ToolCall[]; replay: AssistantReplay } { const candidates = Array.isArray(data.candidates) ? data.candidates : []; + requireFinished(isRecord(candidates[0]) ? candidates[0].finishReason : null, FINISHED_STOP_REASONS.gemini); const content = candidates.length > 0 && isRecord(candidates[0]) && isRecord(candidates[0].content) ? candidates[0].content : null; const parts = content && Array.isArray(content.parts) ? content.parts : []; const result = partsToResult(parts); @@ -138,6 +139,7 @@ export async function accumulateGeminiStream(messages: AsyncIterable 0 && isRecord(candidates[0]) && isRecord(candidates[0].content) ? candidates[0].content : null; const parts = content && Array.isArray(content.parts) ? content.parts : []; for (const part of parts) { @@ -164,5 +167,6 @@ export async function accumulateGeminiStream(messages: AsyncIterable 0 ? { replay: { format: "gemini", content: replayParts } } : {}) }; } diff --git a/src/main/services/provider-tool-turns/responses-format.ts b/src/main/services/provider-tool-turns/responses-format.ts index 9c812a75..801c91f7 100644 --- a/src/main/services/provider-tool-turns/responses-format.ts +++ b/src/main/services/provider-tool-turns/responses-format.ts @@ -4,7 +4,7 @@ * `convertResponsesTools` and the Responses streaming event shapes. */ import { isRecord, ProviderHttpError, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http"; -import { parseArguments, replayIn, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; +import { incompleteTurn, parseArguments, replayIn, type AgentTurnMessage, type AgentTurnResult, type AssistantReplay, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; export function responsesToolDefs(tools: ToolDefinition[], strict: boolean | null = false): unknown[] { // `strict` must be a boolean or omitted. Passing `strict: null` (the Codex @@ -48,6 +48,7 @@ export function toResponsesInput(messages: AgentTurnMessage[]): unknown[] { /** Parse a non-streaming Responses payload (`data.output[]`). */ export function parseResponsesOutput(data: JsonRecord): { text: string; toolCalls: ToolCall[]; replay: AssistantReplay } { + if (data.status !== "completed") throw incompleteTurn(incompleteReason(data)); const output = Array.isArray(data.output) ? data.output : []; const textParts: string[] = []; const toolCalls: ToolCall[] = []; @@ -73,6 +74,11 @@ function responsesReplay(output: unknown[]): unknown[] { .map(({ id: _id, ...item }) => item); } +/** Why a Responses payload is not complete: its incomplete_details reason, else its status. */ +export function incompleteReason(response: JsonRecord): unknown { + return isRecord(response.incomplete_details) && typeof response.incomplete_details.reason === "string" ? response.incomplete_details.reason : response.status; +} + function eventType(message: ProviderEventStreamMessage): string | null { const type = message.data.type; return typeof type === "string" ? type : message.eventName; @@ -122,14 +128,13 @@ export async function accumulateResponsesStream(messages: AsyncIterable 0 ? completionOutput : streamed; - return { ...parseResponsesOutput({ output }), rawResponseJson: JSON.stringify(raw) }; + return { ...parseResponsesOutput({ status: "completed", output }), rawResponseJson: JSON.stringify(raw) }; } } - if (streamed.length > 0) { - return { ...parseResponsesOutput({ output: streamed }), rawResponseJson: JSON.stringify(raw) }; - } - throw new Error("Responses stream ended before a completion event."); + throw incompleteTurn(null); } diff --git a/tests/unit/incomplete-turns.test.ts b/tests/unit/incomplete-turns.test.ts new file mode 100644 index 00000000..320b08c0 --- /dev/null +++ b/tests/unit/incomplete-turns.test.ts @@ -0,0 +1,84 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import type { JsonRecord, ProviderEventStreamMessage } from "../../src/main/services/provider-http"; +import { accumulateResponsesStream, parseResponsesOutput } from "../../src/main/services/provider-tool-turns/responses-format"; +import { accumulateChatStream, parseChatChoice } from "../../src/main/services/provider-tool-turns/chat-format"; +import { parseAnthropicContent } from "../../src/main/services/provider-tool-turns/anthropic-format"; +import { parseBedrockOutput } from "../../src/main/services/provider-tool-turns/bedrock-format"; +import { accumulateGeminiStream, parseGeminiResponse } from "../../src/main/services/provider-tool-turns/gemini-format"; +import { mapAnthropicMessageStream, mapOpenAiChatCompletionStream, mapOpenAiResponseStream } from "../../src/main/services/provider-streaming"; + +async function* stream(items: JsonRecord[]): AsyncIterable { + for (const data of items) yield { eventName: null, data, rawData: JSON.stringify(data) }; +} + +async function types(events: AsyncIterable<{ type: string; textDelta: string | null }>): Promise { + const seen: string[] = []; + for await (const event of events) seen.push(event.type === "error" ? `error:${event.textDelta}` : event.type); + return seen; +} + +const call = (id: string) => ({ type: "function_call", call_id: id, name: "write_file", arguments: '{"path":"a' }); + +describe("a turn the provider cut short is never treated as finished", () => { + test("Responses: incomplete status, incomplete event, and a stream with no completion all fail", async () => { + expect(() => parseResponsesOutput({ status: "incomplete", incomplete_details: { reason: "max_output_tokens" }, output: [call("c1")] })).toThrow("max_output_tokens"); + await expect(accumulateResponsesStream(stream([ + { type: "response.output_item.done", item: call("c1") }, + { type: "response.incomplete", response: { status: "incomplete", incomplete_details: { reason: "content_filter" }, output: [] } }, + ]))).rejects.toThrow("content_filter"); + await expect(accumulateResponsesStream(stream([{ type: "response.output_item.done", item: call("c1") }]))).rejects.toThrow("no stop reason"); + expect(parseResponsesOutput({ status: "completed", output: [] }).toolCalls).toEqual([]); + }); + + test("Chat: length and content_filter fail, a stream without finish_reason fails, stop and tool_calls pass", async () => { + expect(() => parseChatChoice({ choices: [{ message: { content: "half" }, finish_reason: "length" }] })).toThrow("length"); + await expect(accumulateChatStream(stream([{ choices: [{ delta: { content: "half" } }] }]))).rejects.toThrow("no stop reason"); + await expect(accumulateChatStream(stream([{ choices: [{ delta: { content: "x" }, finish_reason: "content_filter" }] }]))).rejects.toThrow("content_filter"); + expect((await accumulateChatStream(stream([{ choices: [{ delta: { content: "ok" }, finish_reason: "stop" }] }]))).text).toBe("ok"); + expect(parseChatChoice({ choices: [{ message: { content: "ok" }, finish_reason: "tool_calls" }] }).text).toBe("ok"); + }); + + test("Anthropic and Bedrock: max_tokens fails; end_turn and tool_use pass", () => { + expect(() => parseAnthropicContent({ stop_reason: "max_tokens", content: [{ type: "tool_use", id: "t", name: "f", input: {} }] })).toThrow("max_tokens"); + expect(parseAnthropicContent({ stop_reason: "tool_use", content: [] }).toolCalls).toEqual([]); + expect(() => parseBedrockOutput({ stopReason: "max_tokens", output: { message: { content: [] } } })).toThrow("max_tokens"); + expect(parseBedrockOutput({ stopReason: "end_turn", output: { message: { content: [{ text: "ok" }] } } }).text).toBe("ok"); + }); + + test("Gemini: MAX_TOKENS fails, a stream without finishReason fails, STOP passes", async () => { + const parts = [{ text: "ok" }]; + expect(() => parseGeminiResponse({ candidates: [{ content: { parts }, finishReason: "MAX_TOKENS" }] })).toThrow("MAX_TOKENS"); + expect(parseGeminiResponse({ candidates: [{ content: { parts }, finishReason: "STOP" }] }).text).toBe("ok"); + await expect(accumulateGeminiStream(stream([{ response: { candidates: [{ content: { parts } }] } }]))).rejects.toThrow("no stop reason"); + expect((await accumulateGeminiStream(stream([{ response: { candidates: [{ content: { parts }, finishReason: "STOP" }] } }]))).text).toBe("ok"); + }); + + test("streamed replies: a cut-off stop is an error, not a message_stop", async () => { + expect(await types(mapOpenAiResponseStream(stream([{ type: "response.incomplete", response: { incomplete_details: { reason: "max_output_tokens" } } }])))) + .toEqual([expect.stringMatching(/^error:.*max_output_tokens/)]); + expect(await types(mapOpenAiChatCompletionStream(stream([{ choices: [{ delta: { content: "half" }, finish_reason: "length" }] }])))) + .toEqual(["message_start", "text_delta", expect.stringMatching(/^error:.*length/)]); + expect(await types(mapAnthropicMessageStream(stream([{ type: "message_start" }, { type: "message_delta", delta: { stop_reason: "max_tokens" } }, { type: "message_stop" }])))) + .toEqual(["message_start", expect.stringMatching(/^error:.*max_tokens/)]); + }); +}); + +describe("one-shot responses cut short fail instead of saving partial text", () => { + const reply = (body: unknown) => vi.stubGlobal("fetch", async () => new Response(JSON.stringify(body), { status: 200, headers: { "content-type": "application/json" } })); + const input = { authProfile: { type: "api_key" as const, apiKey: "k" }, model: "m", prompt: "p", instructions: null }; + afterEach(() => vi.unstubAllGlobals()); + + test("OpenAI incomplete, Anthropic max_tokens and chat length all reject; finished replies return their text", async () => { + reply({ id: "r", status: "incomplete", incomplete_details: { reason: "max_output_tokens" }, output_text: "half" }); + await expect(providerAdapters.openai.createResponse!(input)).rejects.toThrow("max_output_tokens"); + reply({ stop_reason: "max_tokens", content: [{ type: "text", text: "half" }] }); + await expect(providerAdapters.anthropic.createResponse!(input)).rejects.toThrow("max_tokens"); + reply({ choices: [{ message: { content: "half" }, finish_reason: "length" }] }); + await expect(providerAdapters.groq.createResponse!(input)).rejects.toThrow("length"); + reply({ choices: [{ message: { content: "whole" }, finish_reason: "stop" }] }); + expect((await providerAdapters.groq.createResponse!(input)).text).toBe("whole"); + reply({ stop_reason: "end_turn", content: [{ type: "text", text: "whole" }] }); + expect((await providerAdapters.anthropic.createResponse!(input)).text).toBe("whole"); + }); +}); diff --git a/tests/unit/provider-service-streaming.test.ts b/tests/unit/provider-service-streaming.test.ts index 9ac42e0d..7f9493b3 100644 --- a/tests/unit/provider-service-streaming.test.ts +++ b/tests/unit/provider-service-streaming.test.ts @@ -115,6 +115,34 @@ describe("provider service streaming", () => { connection.close(); }); + test("a stream that ends without the provider's stop fails instead of saving the partial text", async () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const adapters: ProviderAdapterRegistry = { + ...providerAdapters, + openai: { + ...providerAdapters.openai, + parseAuthProfile: (secret) => ({ type: "api_key", apiKey: secret.trim() }), + streamResponse: async () => ({ + events: streamEvents([event("message_start", null, "created", "resp_cut"), event("text_delta", "Hal", "delta-1", "resp_cut")]), + rateLimitState: null, + }), + }, + }; + const providers = new ProviderService(connection.db, adapters, testVault); + let orgId = ""; + let openAiProviderId = ""; + connection.transaction(() => { + orgId = createOrganization(organization); + openAiProviderId = providerId(connection, orgId, "openai"); + providers.saveCredential({ orgId, providerId: openAiProviderId, secret: "sk-live-provider", model: "gpt-5.4" }); + }); + await expect(providers.streamResponse({ orgId, providerId: openAiProviderId, prompt: "Draft", instructions: null, executionMode: PROVIDER_EXECUTION_MODE.FOREGROUND }, () => {})) + .rejects.toThrow("no stop reason"); + expect(connection.db.prepare("select state from provider_requests where org_id = ?").get(orgId)).toEqual({ state: PROVIDER_REQUEST_STATE.FAILED }); + connection.close(); + }); + test("cancels a provider request through the adapter response id", async () => { const connection = createConnection(); const organization = new OrganizationService(connection.db); diff --git a/tests/unit/provider-tool-turns.test.ts b/tests/unit/provider-tool-turns.test.ts index 16f1bd2f..92b2590b 100644 --- a/tests/unit/provider-tool-turns.test.ts +++ b/tests/unit/provider-tool-turns.test.ts @@ -47,7 +47,7 @@ describe("openai responses encoding", () => { }); test("parses output text and function calls", () => { - const data = { output: [ + const data = { status: "completed", output: [ { type: "message", content: [{ type: "output_text", text: "done" }] }, { type: "function_call", call_id: "c2", name: "create_task", arguments: '{"title":"x"}' }, ] }; @@ -88,8 +88,8 @@ describe("openai chat completions encoding", () => { }); test("parses a tool call choice", () => { - const data = { choices: [{ message: { content: "hi", tool_calls: [{ id: "c9", function: { name: "f", arguments: '{"a":1}' } }] } }] }; - expect(parseChatChoice(data)).toEqual({ text: "hi", toolCalls: [{ id: "c9", name: "f", arguments: { a: 1 } }] }); + const data = { choices: [{ message: { content: "hi", tool_calls: [{ id: "c9", function: { name: "f", arguments: '{"a":1}' } }] }, finish_reason: "tool_calls" }] }; + expect(parseChatChoice(data)).toMatchObject({ text: "hi", toolCalls: [{ id: "c9", name: "f", arguments: { a: 1 } }] }); }); test("accumulates streamed content deltas and indexed tool-call fragments", async () => { @@ -130,7 +130,7 @@ describe("anthropic encoding", () => { }); test("parses text and tool_use blocks", () => { - const data = { content: [{ type: "text", text: "ok" }, { type: "tool_use", id: "tu", name: "f", input: { x: 1 } }] }; + const data = { stop_reason: "tool_use", content: [{ type: "text", text: "ok" }, { type: "tool_use", id: "tu", name: "f", input: { x: 1 } }] }; expect(parseAnthropicContent(data)).toMatchObject({ text: "ok", toolCalls: [{ id: "tu", name: "f", arguments: { x: 1 } }] }); }); }); @@ -162,7 +162,7 @@ describe("antigravity (gemini) encoding", () => { test("accumulates streamed candidates", async () => { const result = await accumulateGeminiStream(stream([ - { response: { candidates: [{ content: { parts: [{ text: "hello" }, { functionCall: { name: "f", args: { x: 1 }, id: "g1" } }] } }] } }, + { response: { candidates: [{ content: { parts: [{ text: "hello" }, { functionCall: { name: "f", args: { x: 1 }, id: "g1" } }] }, finishReason: "STOP" }] } }, ])); expect(result.text).toBe("hello"); expect(result.toolCalls).toEqual([{ id: "g1", name: "f", arguments: { x: 1 } }]); @@ -183,7 +183,7 @@ describe("bedrock converse encoding", () => { }); test("parses the converse output", () => { - const data = { output: { message: { content: [{ text: "ok" }, { toolUse: { toolUseId: "t1", name: "f", input: { x: 1 } } }] } } }; + const data = { stopReason: "tool_use", output: { message: { content: [{ text: "ok" }, { toolUse: { toolUseId: "t1", name: "f", input: { x: 1 } } }] } } }; expect(parseBedrockOutput(data)).toMatchObject({ text: "ok", toolCalls: [{ id: "t1", name: "f", arguments: { x: 1 } }] }); }); }); @@ -193,7 +193,7 @@ describe("vertex (gemini) encoding", () => { const body = vertexRequestBody({ model: "gemini-2.5-pro", instructions: "sys", messages: CONVO, tools: [{ name: "f", description: "d", parameters: { type: "object" } }] }); expect((body.systemInstruction as JsonRecord).parts).toEqual([{ text: "sys" }]); expect(body.tools).toEqual([{ functionDeclarations: [{ name: "f", description: "d", parametersJsonSchema: { type: "object" } }] }]); - const parsed = parseGeminiResponse({ candidates: [{ content: { parts: [{ text: "hi" }, { functionCall: { name: "f", args: { x: 1 }, id: "v1" } }] } }] }); + const parsed = parseGeminiResponse({ candidates: [{ content: { parts: [{ text: "hi" }, { functionCall: { name: "f", args: { x: 1 }, id: "v1" } }] }, finishReason: "STOP" }] }); expect(parsed).toMatchObject({ text: "hi", toolCalls: [{ id: "v1", name: "f", arguments: { x: 1 } }] }); }); }); diff --git a/tests/unit/reasoning-replay.test.ts b/tests/unit/reasoning-replay.test.ts index f6b1d463..6f3def4f 100644 --- a/tests/unit/reasoning-replay.test.ts +++ b/tests/unit/reasoning-replay.test.ts @@ -31,7 +31,7 @@ describe("reasoning survives tool turns", () => { { type: "reasoning", summary: [], encrypted_content: "ENC" }, { type: "function_call", call_id: "c1", name: "f", arguments: "{}" }, ]); - const plain = parseResponsesOutput({ output: [{ type: "reasoning", id: "rs_2", summary: [] }, { type: "message", id: "msg_1", role: "assistant", content: [{ type: "output_text", text: "hi" }] }] }); + const plain = parseResponsesOutput({ status: "completed", output: [{ type: "reasoning", id: "rs_2", summary: [] }, { type: "message", id: "msg_1", role: "assistant", content: [{ type: "output_text", text: "hi" }] }] }); expect(plain.replay?.content).toEqual([{ type: "message", role: "assistant", content: [{ type: "output_text", text: "hi" }] }]); }); @@ -41,21 +41,21 @@ describe("reasoning survives tool turns", () => { { type: "redacted_thinking", data: "OPAQUE" }, { type: "tool_use", id: "t1", name: "f", input: { a: 1 } }, ]; - const sent = toAnthropicMessages(after(parseAnthropicContent({ content }))) as JsonRecord[]; + const sent = toAnthropicMessages(after(parseAnthropicContent({ stop_reason: "tool_use", content }))) as JsonRecord[]; expect(sent[1]).toEqual({ role: "assistant", content }); }); test("Bedrock: reasoningContent with its signature goes back unmodified", () => { const content = [{ reasoningContent: { reasoningText: { text: "plan", signature: "SIG" } } }, { toolUse: { toolUseId: "t1", name: "f", input: {} } }]; - const sent = toBedrockMessages(after(parseBedrockOutput({ output: { message: { role: "assistant", content } } }))); + const sent = toBedrockMessages(after(parseBedrockOutput({ stopReason: "tool_use", output: { message: { role: "assistant", content } } }))); expect(sent[1]).toEqual({ role: "assistant", content }); }); test("Gemini: thought signatures on parts go back, streamed or not", async () => { const parts = [{ text: "plan", thought: true }, { functionCall: { name: "f", args: {} }, thoughtSignature: "SIG" }]; - const vertex = vertexRequestBody({ model: "gemini-3-pro", instructions: null, messages: after(parseGeminiResponse({ candidates: [{ content: { parts } }] })), tools: [] }); + const vertex = vertexRequestBody({ model: "gemini-3-pro", instructions: null, messages: after(parseGeminiResponse({ candidates: [{ content: { parts }, finishReason: "STOP" }] })), tools: [] }); expect((vertex.contents as JsonRecord[])[1]).toEqual({ role: "model", parts }); - const streamed = await accumulateGeminiStream(stream(parts.map((part) => ({ response: { candidates: [{ content: { parts: [part] } }] } })))); + const streamed = await accumulateGeminiStream(stream([...parts.map((part) => ({ response: { candidates: [{ content: { parts: [part] } }] } })), { response: { candidates: [{ finishReason: "STOP" }] } }])); expect(streamed.replay).toEqual({ format: "gemini", content: parts }); }); @@ -63,22 +63,22 @@ describe("reasoning survives tool turns", () => { const kimi = await accumulateChatStream(stream([ { choices: [{ delta: { reasoning_content: "think " } }] }, { choices: [{ delta: { reasoning_content: "hard" } }] }, - { choices: [{ delta: { tool_calls: [{ index: 0, id: "c1", function: { name: "f", arguments: "{}" } }] } }] }, + { choices: [{ delta: { tool_calls: [{ index: 0, id: "c1", function: { name: "f", arguments: "{}" } }] }, finish_reason: "tool_calls" }] }, ])); expect((toChatMessages(null, after(kimi)) as JsonRecord[])[1]).toMatchObject({ role: "assistant", reasoning_content: "think hard" }); const openrouter = await accumulateChatStream(stream([ { choices: [{ delta: { reasoning: "a", reasoning_details: [{ type: "reasoning.text", text: "a", index: 0 }] } }] }, { choices: [{ delta: { reasoning: "b", reasoning_details: [{ type: "reasoning.text", text: "b", signature: "SIG", index: 0 }] } }] }, { choices: [{ delta: { reasoning_details: [{ type: "reasoning.encrypted", data: "ENC", index: 1 }] } }] }, - { choices: [{ delta: { content: "done" } }] }, + { choices: [{ delta: { content: "done" }, finish_reason: "stop" }] }, ])); expect(openrouter.replay).toEqual({ format: "chat", content: { reasoning_details: [{ type: "reasoning.text", text: "ab", signature: "SIG", index: 0 }, { type: "reasoning.encrypted", data: "ENC", index: 1 }] } }); - const azure = parseChatChoice({ choices: [{ message: { content: "x", reasoning: "why" } }] }); + const azure = parseChatChoice({ choices: [{ message: { content: "x", reasoning: "why" }, finish_reason: "stop" }] }); expect(azure.replay).toEqual({ format: "chat", content: { reasoning: "why" } }); }); test("a turn from one wire format is never silently replayed in another", () => { - const messages = after(parseAnthropicContent({ content: [{ type: "text", text: "hi" }] })); + const messages = after(parseAnthropicContent({ stop_reason: "end_turn", content: [{ type: "text", text: "hi" }] })); expect(() => toChatMessages(null, messages)).toThrow("anthropic format cannot be sent to a chat provider"); }); }); diff --git a/tests/unit/stream-usage-request.test.ts b/tests/unit/stream-usage-request.test.ts index dae89c88..9c6eb73d 100644 --- a/tests/unit/stream-usage-request.test.ts +++ b/tests/unit/stream-usage-request.test.ts @@ -35,7 +35,7 @@ describe("system instructions", () => { test("the GitHub Copilot turn sends the agent's instructions like every other Responses turn", async () => { vi.stubGlobal("fetch", async (_url: string, init: RequestInit) => { bodies.push(JSON.parse(String(init.body)) as Record); - return new Response(JSON.stringify({ output: [{ type: "message", content: [{ type: "output_text", text: "ok" }] }] }), { status: 200, headers: { "content-type": "application/json" } }); + return new Response(JSON.stringify({ status: "completed", output: [{ type: "message", content: [{ type: "output_text", text: "ok" }] }] }), { status: 200, headers: { "content-type": "application/json" } }); }); await runCopilotTurn({ credential: { accessToken: "tid=1;exp=9999999999" }, model: "gpt-5.4", instructions: "You are Ada, the backend engineer.", messages: [{ role: "user", text: "hi" }], tools: [] }); expect(bodies.at(-1)?.instructions).toBe("You are Ada, the backend engineer.");