diff --git a/package.json b/package.json index d6722c3d12..f4b146180a 100644 --- a/package.json +++ b/package.json @@ -96,6 +96,7 @@ "test:tool-reliability": "npx tsx test/continuous-test-suite-tool-reliability.ts", "test:google-native": "npx tsx test/continuous-test-suite-google-native.ts", "test:gemini-abort": "npx tsx test/continuous-test-suite-gemini-abort.ts", + "test:anthropic-cap": "npx tsx test/continuous-test-suite-anthropic-cap.ts", "test:tts": "npx tsx test/continuous-test-suite-tts.ts", "test:voice": "npx tsx test/continuous-test-suite-voice.ts", "test:voice-server": "npx tsx test/continuous-test-suite-voice-server.ts", diff --git a/src/lib/providers/googleVertex.ts b/src/lib/providers/googleVertex.ts index badc0e75e9..6b2311e145 100644 --- a/src/lib/providers/googleVertex.ts +++ b/src/lib/providers/googleVertex.ts @@ -1283,7 +1283,10 @@ export class GoogleVertexProvider extends BaseProvider { } }, (r) => r.stream, - (r, wrapped) => ({ ...r, stream: wrapped }), + // Preserve live getters (finishReason/structuredOutput/…) that the + // spread would otherwise snapshot before the background loop resolves. + (r, wrapped) => + this.preserveStreamResultAccessors(r, { ...r, stream: wrapped }), ); } @@ -3749,6 +3752,13 @@ export class GoogleVertexProvider extends BaseProvider { // pattern in googleAiStudio.ts. const maxSteps = options.maxSteps || DEFAULT_MAX_STEPS; + // Reserve the LAST step of the budget for a forced final_result call when + // structured output is active — see the generate twin for the full + // rationale (tool_choice:"any" makes the text exit unreachable, so an + // un-reserved budget deadlocks into an empty stream). + const agenticStepBudget = useFinalResultTool + ? Math.max(maxSteps - 1, 0) + : maxSteps; const allToolCalls: Array<{ toolName: string; args: Record; @@ -3771,7 +3781,16 @@ export class GoogleVertexProvider extends BaseProvider { cacheReadTokens?: number; cacheCreationTokens?: number; } = { input: 0, output: 0, total: 0 }; - const metadata = { + const metadata: { + streamId: string; + startTime: number; + responseTime: number; + totalToolExecutions: number; + // Mirrors finishReasonRef — set on metadata too because wrapper spreads + // ({ ...result }) snapshot the top-level getter to undefined before the + // background loop resolves, while the metadata object reference survives. + finishReason?: string; + } = { streamId: `native-anthropic-vertex-${Date.now()}`, startTime, responseTime: 0, @@ -3779,6 +3798,10 @@ export class GoogleVertexProvider extends BaseProvider { }; const toolsUsedRef: string[] = []; const structuredOutputRef: { value?: Record } = {}; + // Resolved after the background loop finishes; the StreamResult exposes it + // via a getter (consumers read it after draining the stream), mirroring + // the Gemini stream path's finishReason field. + const finishReasonRef: { value?: string } = {}; // Langfuse/OTel: the native SDK bypasses the Vercel AI SDK's // experimental_telemetry, so emit spans manually — one turn span, one @@ -3818,6 +3841,11 @@ export class GoogleVertexProvider extends BaseProvider { }); const turnContext = otelTrace.setSpan(otelContext.active(), turnSpan); let aggregatedTurnText = ""; + // Count of characters ALREADY delivered to the consumer via live text + // deltas. aggregatedTurnText only reflects completed finalMessage() + // responses, so an aborted mid-answer call would otherwise look "empty" + // to the terminal handling even though the consumer saw partial prose. + let liveTextPushedLength = 0; // Anthropic prompt-cache token accounting, aggregated across loop steps. const turnCacheUsage = { read: 0, @@ -3854,11 +3882,24 @@ export class GoogleVertexProvider extends BaseProvider { const loopPromise = (async () => { let step = 0; const currentMessages = [...messages]; + // Step-cap / abort bookkeeping (mirrors the generate twin): read by the + // terminal recovery block after the loop and by the finishReason + // mapping written into finishReasonRef. + let wasAborted = false; + let hitStepLimit = false; + let synthesizedFinalAnswer = false; + let modelFinished = false; + let lastStopReason: string | null | undefined; try { - while (step < maxSteps) { + while (step < agenticStepBudget) { + // Honor caller aborts BETWEEN steps: break into terminal handling + // (one graceful cap chunk, clean close) instead of throwing — + // channel.error would surface the caller's own abort as a stream + // failure and route consumers into fallback retries. if (options.abortSignal?.aborted) { - throw new Error("Stream aborted by caller"); + wasAborted = true; + break; } step++; @@ -3944,6 +3985,7 @@ export class GoogleVertexProvider extends BaseProvider { ); } channel.push(delta); + liveTextPushedLength += delta.length; } }); @@ -3964,6 +4006,15 @@ export class GoogleVertexProvider extends BaseProvider { generationSpan.recordException(modelCallError); } generationSpan.end(); + // A mid-flight abort (caller signal or the defensive timer + // tripping abortHandler) rejects finalMessage() with an + // abort-shaped error. Break gracefully into the terminal handling + // instead of routing it through channel.error as a failure. + if (options.abortSignal?.aborted || isAbortError(modelCallError)) { + activeStream = undefined; + wasAborted = true; + break; + } throw modelCallError; } activeStream = undefined; @@ -3986,6 +4037,7 @@ export class GoogleVertexProvider extends BaseProvider { usage.input += response.usage?.input_tokens || 0; usage.output += response.usage?.output_tokens || 0; usage.total = usage.input + usage.output; + lastStopReason = response.stop_reason; for (const block of response.content) { if (block.type === "text" && typeof block.text === "string") { @@ -4059,6 +4111,7 @@ export class GoogleVertexProvider extends BaseProvider { if (finalResultCall) { structuredOutputRef.value = finalResultCall.input; channel.push(JSON.stringify(finalResultCall.input)); + modelFinished = true; logger.debug( "[GoogleVertex] Extracted structured output from final_result tool (stream)", { keys: Object.keys(finalResultCall.input) }, @@ -4070,17 +4123,23 @@ export class GoogleVertexProvider extends BaseProvider { // No tools — pure text turn. Listener already pushed all deltas; // loop terminates and channel.close() flushes the consumer. if (toolUseBlocks.length === 0) { + modelFinished = true; break; } // Tool execution loop. tool:start / tool:end events fire from // ToolsManager's wrapped execute (ToolsManager.ts:355) — no inline - // emit needed. - const toolResults: Array<{ - type: "tool_result"; - tool_use_id: string; - content: string; - }> = []; + // emit needed. The array also carries the trailing soft-budget + // nudge text block appended below (tool_result blocks stay first, + // as the Anthropic API requires). + const toolResults: Array< + | { + type: "tool_result"; + tool_use_id: string; + content: string; + } + | { type: "text"; text: string } + > = []; // Per-step bookkeeping for conversation-memory storage. const stepStorageCalls: Array<{ toolCallId: string; @@ -4195,6 +4254,23 @@ export class GoogleVertexProvider extends BaseProvider { output: result, }); } catch (err) { + // An aborted tool call is a cancellation, not a tool failure — + // end the span without recording an error execution/result and + // break the turn. + if (options.abortSignal?.aborted || isAbortError(err)) { + endToolSpan({ aborted: true }); + // Keep persisted tool history paired: the call row was + // already pushed above, so record a neutral cancellation + // result (NOT an error) — an unpaired tool_use replayed on + // the next turn would be rejected by the Anthropic API. + stepStorageResults.push({ + toolCallId: toolUse.id, + toolName: toolUse.name, + output: { aborted: true }, + }); + wasAborted = true; + break; + } const errMsg = `Error executing tool "${toolUse.name}": ${err instanceof Error ? err.message : String(err)}`; const errorPayload = { error: errMsg }; endToolSpan(errorPayload, errMsg); @@ -4238,7 +4314,9 @@ export class GoogleVertexProvider extends BaseProvider { // Persist this step's tool calls/results into conversation memory. // Without this hook, tool rows never land in Redis and the - // chat-history UI loses every tool invocation. + // chat-history UI loses every tool invocation. Runs BEFORE the + // abort break below so tools that DID complete in an aborted step + // (real side effects) still reach the chat history. if (stepStorageCalls.length > 0 || stepStorageResults.length > 0) { withTimeout( this.handleToolExecutionStorage( @@ -4259,6 +4337,29 @@ export class GoogleVertexProvider extends BaseProvider { }); } + // An abort inside the tool-exec loop only breaks that inner + // for-loop. Break the while too so no further model call is issued + // and control reaches the terminal step-cap handling below. + if (wasAborted) { + break; + } + + // Soft budget nudge: with the step cap approaching, tell the model + // to wrap up so the reserved forced-finalization call below stays a + // fallback, not the norm. Rides as a trailing text block on the + // tool_result user turn (cache-safe: it lives in the growing tail). + const stepsRemaining = agenticStepBudget - step; + if (stepsRemaining > 0 && stepsRemaining <= 3) { + toolResults.push({ + type: "text", + text: + `NOTE: Only ${stepsRemaining} tool step(s) remain. Consolidate what you have and ` + + (useFinalResultTool + ? "call final_result with your best answer." + : "provide your final answer."), + }); + } + // Continue the loop: assistant turn + tool_result user turn. // Filter server_tool_use blocks (Anthropic API rejects them in // subsequent message turns). @@ -4275,6 +4376,287 @@ export class GoogleVertexProvider extends BaseProvider { }); } + // Terminal handling — the loop exited without a model-initiated + // finish (step budget exhausted, or the turn was aborted). Never end + // the stream empty: force the reserved final_result step on + // structured-output turns, synthesize a plain-text answer on tool + // turns, or push one graceful cap chunk. Mirrors the generate twin + // and the Gemini paths (ed289b7 / PR #1123). The finalization and + // backstop calls emit no per-call generation span (matching the + // Gemini synthesizeFinalAnswerWithoutTools precedent); their tokens + // still land in `usage` and the turn-span metadata. + if (!modelFinished) { + // A caller abort that landed during the FINAL budgeted step's tool + // execution leaves wasAborted=false (no loop-entry check runs after + // a budget exit) — re-check before issuing any terminal model call. + if (options.abortSignal?.aborted) { + wasAborted = true; + } + const externalToolCallCount = allToolCalls.filter( + (tc) => tc.toolName !== "final_result", + ).length; + if (wasAborted) { + // Budget already blown — never issue another model call. Deliver + // exactly one graceful cap chunk if nothing reached the consumer + // (live deltas already delivered count as "something reached"). + logger.warn( + "[GoogleVertex] Native Anthropic stream loop aborted mid-turn; returning a graceful cap message.", + ); + if ( + aggregatedTurnText.length === 0 && + liveTextPushedLength === 0 && + !structuredOutputRef.value + ) { + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + channel.push(capMessage); + aggregatedTurnText = capMessage; + } + } else if (useFinalResultTool) { + hitStepLimit = true; + logger.warn( + `[GoogleVertex] Native Anthropic stream loop reached maxSteps (${maxSteps}) without final_result; forcing a finalization call.`, + ); + try { + // Reserved finalization step: identical request except + // tool_choice pins final_result — Anthropic guarantees a + // final_result tool_use. Cache treatment is byte-identical to + // loop steps. No text listener: forced tool calls stream no + // text deltas, and the structured JSON is pushed once below. + const cachedFinal = applyVertexAnthropicCacheBreakpoints({ + system: systemPromptWithSchema, + tools, + messages: currentMessages, + }); + const finalizationStream = await client.messages.stream({ + ...requestParams, + tool_choice: { + type: "tool" as const, + name: "final_result", + }, + ...(cachedFinal.system !== undefined && { + system: cachedFinal.system as Parameters< + typeof client.messages.stream + >[0]["system"], + }), + ...(cachedFinal.tools && + cachedFinal.tools.length > 0 && { + tools: cachedFinal.tools as Parameters< + typeof client.messages.stream + >[0]["tools"], + }), + messages: cachedFinal.messages as Parameters< + typeof client.messages.stream + >[0]["messages"], + }); + // Mid-flight aborts are covered by the shared abortHandler via + // activeStream; already-fired aborts were handled by the + // terminal-entry re-check above. + activeStream = finalizationStream; + const response = await finalizationStream.finalMessage(); + activeStream = undefined; + usage.input += response.usage?.input_tokens || 0; + usage.output += response.usage?.output_tokens || 0; + usage.total = usage.input + usage.output; + turnCacheUsage.read += + response.usage?.cache_read_input_tokens ?? 0; + turnCacheUsage.creation += + response.usage?.cache_creation_input_tokens ?? 0; + turnCacheUsage.creation5m += + response.usage?.cache_creation?.ephemeral_5m_input_tokens ?? 0; + turnCacheUsage.creation1h += + response.usage?.cache_creation?.ephemeral_1h_input_tokens ?? 0; + lastStopReason = response.stop_reason; + const forcedFinalResult = ( + response.content as VertexAnthropicContentBlock[] + ).find( + ( + block, + ): block is { + type: "tool_use"; + id: string; + name: string; + input: Record; + } => block.type === "tool_use" && block.name === "final_result", + ); + if (forcedFinalResult) { + structuredOutputRef.value = forcedFinalResult.input; + channel.push(JSON.stringify(forcedFinalResult.input)); + synthesizedFinalAnswer = true; + logger.debug( + "[GoogleVertex] Forced finalization returned structured output (stream)", + { keys: Object.keys(forcedFinalResult.input) }, + ); + } else { + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + channel.push(capMessage); + aggregatedTurnText += capMessage; + } + } catch (error) { + activeStream = undefined; + // An aborted finalization is a cancellation, not a step-cap + // turn — keep finishReason mapping from lastStopReason. + if (options.abortSignal?.aborted || isAbortError(error)) { + hitStepLimit = false; + } + logger.warn( + "[GoogleVertex] Forced finalization call failed; falling back to cap message", + { + error: error instanceof Error ? error.message : String(error), + }, + ); + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + channel.push(capMessage); + aggregatedTurnText += capMessage; + } + } else { + hitStepLimit = true; + if (aggregatedTurnText.length === 0) { + // Pure tool-loop with no streamed text: one tools-disabled call + // so the model answers from the gathered tool results, pushed + // as a single chunk (matching the Gemini stream synth). NOTE: + // the tools array must stay in the request — Anthropic rejects + // histories containing tool_use/tool_result blocks without a + // tools param — so "tools disabled" is tool_choice:"none", + // which also keeps the cached tools prefix byte-identical. + logger.warn( + `[GoogleVertex] Native Anthropic stream loop reached maxSteps (${maxSteps}) with no text; synthesizing a final answer with tools disabled.`, + ); + try { + const backstopSystem = + (systemPromptWithSchema + ? systemPromptWithSchema + "\n\n" + : "") + + "Tool calling is no longer available for this turn. " + + "Provide your final answer directly as plain text now, " + + "using the information gathered so far."; + const cachedBackstop = applyVertexAnthropicCacheBreakpoints({ + system: backstopSystem, + tools, + messages: currentMessages, + }); + const backstopStream = await client.messages.stream({ + ...requestParams, + tool_choice: { type: "none" as const }, + system: cachedBackstop.system as Parameters< + typeof client.messages.stream + >[0]["system"], + ...(cachedBackstop.tools && + cachedBackstop.tools.length > 0 && { + tools: cachedBackstop.tools as Parameters< + typeof client.messages.stream + >[0]["tools"], + }), + messages: cachedBackstop.messages as Parameters< + typeof client.messages.stream + >[0]["messages"], + }); + activeStream = backstopStream; + const response = await backstopStream.finalMessage(); + activeStream = undefined; + usage.input += response.usage?.input_tokens || 0; + usage.output += response.usage?.output_tokens || 0; + usage.total = usage.input + usage.output; + turnCacheUsage.read += + response.usage?.cache_read_input_tokens ?? 0; + turnCacheUsage.creation += + response.usage?.cache_creation_input_tokens ?? 0; + turnCacheUsage.creation5m += + response.usage?.cache_creation?.ephemeral_5m_input_tokens ?? + 0; + turnCacheUsage.creation1h += + response.usage?.cache_creation?.ephemeral_1h_input_tokens ?? + 0; + lastStopReason = response.stop_reason; + const backstopText = ( + response.content as VertexAnthropicContentBlock[] + ) + .filter( + (block): block is { type: "text"; text: string } => + block.type === "text", + ) + .map((b) => b.text) + .join(""); + if (backstopText) { + synthesizedFinalAnswer = true; + channel.push(backstopText); + aggregatedTurnText = backstopText; + } else { + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + channel.push(capMessage); + aggregatedTurnText = capMessage; + } + } catch (error) { + activeStream = undefined; + // An aborted backstop is a cancellation, not a step-cap + // turn — keep finishReason mapping from lastStopReason. + if (options.abortSignal?.aborted || isAbortError(error)) { + hitStepLimit = false; + } + logger.warn( + "[GoogleVertex] Tools-disabled backstop call failed; falling back to cap message", + { + error: + error instanceof Error ? error.message : String(error), + }, + ); + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + channel.push(capMessage); + aggregatedTurnText = capMessage; + } + } + // else: prose already streamed to the consumer across steps — + // add nothing; finishReason "tool-calls" flags the capped turn. + } + } + + // Absolute backstop — never close the stream empty on a turn that + // ran tools (mirrors the generate twin's guard). Catches the + // model-finished-with-empty-content exit, which skips the terminal + // recovery above. + const deliveredNothing = + aggregatedTurnText.length === 0 && + liveTextPushedLength === 0 && + !structuredOutputRef.value; + const externalToolCallTotal = allToolCalls.filter( + (tc) => tc.toolName !== "final_result", + ).length; + if (deliveredNothing && externalToolCallTotal > 0) { + const capMessage = buildToolLoopCapMessage( + maxSteps, + externalToolCallTotal, + ); + channel.push(capMessage); + aggregatedTurnText = capMessage; + } + + // Honest finish reason (same mapping as the generate twin): "length" + // takes precedence, a capped turn without a clean/forced answer is + // "tool-calls", everything else is "stop". Mirrored onto metadata + // (a mutable reference) because wrapper spreads snapshot the + // top-level getter before this line runs. + finishReasonRef.value = + lastStopReason === "max_tokens" + ? "length" + : hitStepLimit && !synthesizedFinalAnswer + ? "tool-calls" + : "stop"; + metadata.finishReason = finishReasonRef.value; + metadata.responseTime = Date.now() - startTime; metadata.totalToolExecutions = allToolCalls.filter( (tc) => tc.toolName !== "final_result", @@ -4385,6 +4767,14 @@ export class GoogleVertexProvider extends BaseProvider { get: () => structuredOutputRef.value, }); + // Resolved by the background loop right before channel.close(); consumers + // read it after draining the stream (same contract as usage/metadata). + Object.defineProperty(result, "finishReason", { + enumerable: true, + configurable: true, + get: () => finishReasonRef.value, + }); + return result; } @@ -4730,8 +5120,21 @@ export class GoogleVertexProvider extends BaseProvider { // Handle tool calling loop with max steps const maxSteps = options.maxSteps || DEFAULT_MAX_STEPS; + // Reserve the LAST step of the budget for a forced final_result call when + // structured output is active. tool_choice:"any" (set above) forces a tool + // call on every assistant turn, so the toolUseBlocks.length===0 text exit + // is structurally unreachable — without a reserved finalization step, a + // turn whose budget is consumed by other tool calls (runaway loop, or all + // tools failing) ends with content:"" and a masking finishReason "stop". + const agenticStepBudget = useFinalResultTool + ? Math.max(maxSteps - 1, 0) + : maxSteps; let step = 0; let finalText = ""; + // Prose produced mid-loop alongside tool calls, accumulated across steps + // via appendStepText — surfaced if the step budget runs out (mirrors the + // Gemini paths' accumulatedText; last-step-only would drop earlier prose). + let accumulatedStepText = ""; let structuredOutput: Record | undefined; const allToolCalls: Array<{ toolName: string; @@ -4749,9 +5152,23 @@ export class GoogleVertexProvider extends BaseProvider { // (notably "length" on token truncation) — the legacy native path always // reported "stop", hiding truncation from callers. let lastStopReason: string | null | undefined; + // Step-cap / abort bookkeeping (mirrors the native Gemini loops): the + // terminal recovery block below and the finishReason mapping both read + // these to distinguish clean finishes, capped turns, and aborted turns. + let wasAborted = false; + let hitStepLimit = false; + let synthesizedFinalAnswer = false; + let modelFinished = false; const currentMessages = [...messages]; - while (step < maxSteps) { + while (step < agenticStepBudget) { + // Honor caller aborts BETWEEN steps: break into terminal handling + // instead of throwing (a throw routes consumers into abortSignal-less + // fallback retries — observed in production as a 600s abort no-op). + if (options.abortSignal?.aborted) { + wasAborted = true; + break; + } step++; try { @@ -4769,24 +5186,30 @@ export class GoogleVertexProvider extends BaseProvider { tools, messages: currentMessages, }); + // The caller's abortSignal rides as an SDK request option so a + // mid-flight abort cancels the HTTP call itself (the SDK rejects with + // an abort-shaped error the per-step catch below turns into a break). const response = await withTimeout( - client.messages.create({ - ...requestParams, - ...(cachedGenerate.system !== undefined && { - system: cachedGenerate.system as Parameters< - typeof client.messages.create - >[0]["system"], - }), - ...(cachedGenerate.tools && - cachedGenerate.tools.length > 0 && { - tools: cachedGenerate.tools as Parameters< + client.messages.create( + { + ...requestParams, + ...(cachedGenerate.system !== undefined && { + system: cachedGenerate.system as Parameters< typeof client.messages.create - >[0]["tools"], + >[0]["system"], }), - messages: cachedGenerate.messages as Parameters< - typeof client.messages.create - >[0]["messages"], - }), + ...(cachedGenerate.tools && + cachedGenerate.tools.length > 0 && { + tools: cachedGenerate.tools as Parameters< + typeof client.messages.create + >[0]["tools"], + }), + messages: cachedGenerate.messages as Parameters< + typeof client.messages.create + >[0]["messages"], + }, + { signal: options.abortSignal }, + ), generateTimeoutMs, "Anthropic generate timed out", ); @@ -4824,6 +5247,7 @@ export class GoogleVertexProvider extends BaseProvider { // Extract structured output and convert to JSON string for finalText structuredOutput = finalResultCall.input; finalText = JSON.stringify(structuredOutput); + modelFinished = true; logger.debug( "[GoogleVertex] Extracted structured output from final_result tool (generate)", { keys: Object.keys(structuredOutput) }, @@ -4843,16 +5267,22 @@ export class GoogleVertexProvider extends BaseProvider { if (toolUseBlocks.length === 0) { // No tool calls, we're done - finalText = responseText || finalText; + finalText = responseText || accumulatedStepText; + modelFinished = true; break; } - // Handle tool calls - const toolResults: Array<{ - type: "tool_result"; - tool_use_id: string; - content: string; - }> = []; + // Handle tool calls. The array also carries the trailing soft-budget + // nudge text block appended below (tool_result blocks stay first, as + // the Anthropic API requires). + const toolResults: Array< + | { + type: "tool_result"; + tool_use_id: string; + content: string; + } + | { type: "text"; text: string } + > = []; // Per-step bookkeeping for conversation-memory storage. Tracks calls // and results for ONLY the tools fired in this step so the storage // hook can tag them with the current stepIndex. @@ -4911,6 +5341,21 @@ export class GoogleVertexProvider extends BaseProvider { output: result, }); } catch (err) { + // An aborted tool call is a cancellation, not a tool failure — + // break the turn without recording an error execution/result. + if (options.abortSignal?.aborted || isAbortError(err)) { + // Keep persisted tool history paired: the call row was + // already pushed above, so record a neutral cancellation + // result (NOT an error) — an unpaired tool_use replayed on + // the next turn would be rejected by the Anthropic API. + stepStorageResults.push({ + toolCallId: toolUse.id, + toolName: toolUse.name, + output: { aborted: true }, + }); + wasAborted = true; + break; + } const errMsg = `Error executing tool "${toolUse.name}": ${err instanceof Error ? err.message : String(err)}`; const errorPayload = { error: errMsg }; toolExecutions.push({ @@ -4954,6 +5399,8 @@ export class GoogleVertexProvider extends BaseProvider { // Without this, tool_call / tool_result rows never reach Redis and // the chat-history UI loses every tool invocation. // Fire-and-forget — storage failures must not break generation. + // Runs BEFORE the abort break below so tools that DID complete in an + // aborted step (real side effects) still reach the chat history. if (stepStorageCalls.length > 0 || stepStorageResults.length > 0) { withTimeout( this.handleToolExecutionStorage( @@ -4974,6 +5421,29 @@ export class GoogleVertexProvider extends BaseProvider { }); } + // An abort inside the tool-exec loop only breaks that inner for-loop. + // Break the while too so no further model call is issued and control + // reaches the terminal step-cap handling below. + if (wasAborted) { + break; + } + + // Soft budget nudge: with the step cap approaching, tell the model to + // wrap up so the reserved forced-finalization call below stays a + // fallback, not the norm. Rides as a trailing text block on the + // tool_result user turn (cache-safe: it lives in the growing tail). + const stepsRemaining = agenticStepBudget - step; + if (stepsRemaining > 0 && stepsRemaining <= 3) { + toolResults.push({ + type: "text", + text: + `NOTE: Only ${stepsRemaining} tool step(s) remain. Consolidate what you have and ` + + (useFinalResultTool + ? "call final_result with your best answer." + : "provide your final answer."), + }); + } + // Add assistant message and tool results to continue the loop // Filter out server_tool_use blocks that the Anthropic API doesn't accept in messages const assistantContent = response.content.filter( @@ -4988,11 +5458,20 @@ export class GoogleVertexProvider extends BaseProvider { content: toolResults, }); - // Store last text in case we hit max steps - if (responseText) { - finalText = responseText; - } + // Accumulate the step's prose so a capped turn can still surface it — + // finalText is reserved for model-initiated finishes. + accumulatedStepText = appendStepText(accumulatedStepText, responseText); } catch (error) { + // A mid-request abort surfaces as an abort-shaped SDK rejection (we + // pass options.abortSignal as the request signal above). Break + // gracefully into the terminal handling instead of re-throwing — a + // re-throw routes the caller's abort into abortSignal-less fallback + // retries. Dual check as in the Gemini loops: the caller's signal OR + // an abort-shaped error either way means "stop", not a real failure. + if (options.abortSignal?.aborted || isAbortError(error)) { + wasAborted = true; + break; + } logger.error( "[GoogleVertex] Native Anthropic SDK generate error", error, @@ -5001,6 +5480,219 @@ export class GoogleVertexProvider extends BaseProvider { } } + // Terminal handling — the loop exited without a model-initiated finish + // (step budget exhausted, or the turn was aborted). Never return "": + // force the reserved final_result step on structured-output turns, + // synthesize a plain-text answer on tool turns, or fall back to a + // graceful cap message. Mirrors the Gemini paths (ed289b7 / PR #1123). + if (!modelFinished && !finalText) { + // A caller abort that landed during the FINAL budgeted step's tool + // execution leaves wasAborted=false (no loop-entry check runs after a + // budget exit) — re-check before issuing any terminal model call. + if (options.abortSignal?.aborted) { + wasAborted = true; + } + const externalToolCallCount = allToolCalls.filter( + (tc) => tc.toolName !== "final_result", + ).length; + if (wasAborted) { + // Budget already blown — never issue another model call; prefer the + // prose the model already produced (mirrors the Gemini abort path), + // else answer with a graceful cap message. finishReason keeps mapping + // from lastStopReason (an aborted turn is not a step-cap turn). + logger.warn( + "[GoogleVertex] Native Anthropic generate loop aborted mid-turn; returning gathered text or a graceful cap message.", + ); + finalText = + accumulatedStepText || + buildToolLoopCapMessage(maxSteps, externalToolCallCount); + } else if (useFinalResultTool) { + hitStepLimit = true; + logger.warn( + `[GoogleVertex] Native Anthropic generate loop reached maxSteps (${maxSteps}) without final_result; forcing a finalization call.`, + ); + try { + // Reserved finalization step: identical request except tool_choice + // pins final_result, so Anthropic guarantees the response is a + // final_result tool_use. Cache treatment is byte-identical to loop + // steps (same applyVertexAnthropicCacheBreakpoints on the same + // stable system/tools prefix), so caching is unaffected. + const cachedFinal = applyVertexAnthropicCacheBreakpoints({ + system: systemPromptWithSchema, + tools, + messages: currentMessages, + }); + const response = await withTimeout( + client.messages.create( + { + ...requestParams, + tool_choice: { + type: "tool" as const, + name: "final_result", + }, + ...(cachedFinal.system !== undefined && { + system: cachedFinal.system as Parameters< + typeof client.messages.create + >[0]["system"], + }), + ...(cachedFinal.tools && + cachedFinal.tools.length > 0 && { + tools: cachedFinal.tools as Parameters< + typeof client.messages.create + >[0]["tools"], + }), + messages: cachedFinal.messages as Parameters< + typeof client.messages.create + >[0]["messages"], + }, + { signal: options.abortSignal }, + ), + generateTimeoutMs, + "Anthropic finalization call timed out", + ); + totalInputTokens += response.usage?.input_tokens || 0; + totalOutputTokens += response.usage?.output_tokens || 0; + totalCacheReadTokens += response.usage?.cache_read_input_tokens || 0; + totalCacheCreationTokens += + response.usage?.cache_creation_input_tokens || 0; + lastStopReason = response.stop_reason; + const forcedFinalResult = ( + response.content as VertexAnthropicContentBlock[] + ).find( + ( + block, + ): block is { + type: "tool_use"; + id: string; + name: string; + input: Record; + } => block.type === "tool_use" && block.name === "final_result", + ); + if (forcedFinalResult) { + structuredOutput = forcedFinalResult.input; + finalText = JSON.stringify(structuredOutput); + synthesizedFinalAnswer = true; + logger.debug( + "[GoogleVertex] Forced finalization returned structured output (generate)", + { keys: Object.keys(structuredOutput) }, + ); + } else { + finalText = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + } + } catch (error) { + // An aborted finalization is a cancellation, not a step-cap turn — + // keep finishReason mapping from lastStopReason, not "tool-calls". + if (options.abortSignal?.aborted || isAbortError(error)) { + hitStepLimit = false; + } + logger.warn( + "[GoogleVertex] Forced finalization call failed; falling back to cap message", + { error: error instanceof Error ? error.message : String(error) }, + ); + finalText = buildToolLoopCapMessage(maxSteps, externalToolCallCount); + } + } else { + hitStepLimit = true; + if (accumulatedStepText) { + // Prefer the prose the model already produced across steps. + logger.warn( + `[GoogleVertex] Native Anthropic generate loop reached maxSteps (${maxSteps}); returning text already gathered from prior steps.`, + ); + finalText = accumulatedStepText; + } else { + // Pure tool-loop with no text: one tools-disabled call so the model + // answers from the gathered tool results (Anthropic twin of the + // Gemini synthesizeFinalAnswerWithoutTools). NOTE: the tools array + // must stay in the request — Anthropic rejects requests whose + // history contains tool_use/tool_result blocks without a tools + // param — so "tools disabled" is expressed as tool_choice:"none", + // which also keeps the cached tools prefix byte-identical. + logger.warn( + `[GoogleVertex] Native Anthropic generate loop reached maxSteps (${maxSteps}) with no text; synthesizing a final answer with tools disabled.`, + ); + try { + const backstopSystem = + (systemPromptWithSchema ? systemPromptWithSchema + "\n\n" : "") + + "Tool calling is no longer available for this turn. Provide " + + "your final answer directly as plain text now, using the " + + "information gathered so far."; + const cachedBackstop = applyVertexAnthropicCacheBreakpoints({ + system: backstopSystem, + tools, + messages: currentMessages, + }); + const response = await withTimeout( + client.messages.create( + { + ...requestParams, + tool_choice: { type: "none" as const }, + system: cachedBackstop.system as Parameters< + typeof client.messages.create + >[0]["system"], + ...(cachedBackstop.tools && + cachedBackstop.tools.length > 0 && { + tools: cachedBackstop.tools as Parameters< + typeof client.messages.create + >[0]["tools"], + }), + messages: cachedBackstop.messages as Parameters< + typeof client.messages.create + >[0]["messages"], + }, + { signal: options.abortSignal }, + ), + generateTimeoutMs, + "Anthropic backstop call timed out", + ); + totalInputTokens += response.usage?.input_tokens || 0; + totalOutputTokens += response.usage?.output_tokens || 0; + totalCacheReadTokens += + response.usage?.cache_read_input_tokens || 0; + totalCacheCreationTokens += + response.usage?.cache_creation_input_tokens || 0; + lastStopReason = response.stop_reason; + const backstopText = ( + response.content as VertexAnthropicContentBlock[] + ) + .filter( + (block): block is { type: "text"; text: string } => + block.type === "text", + ) + .map((b) => b.text) + .join(""); + if (backstopText) { + synthesizedFinalAnswer = true; + finalText = backstopText; + } else { + finalText = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + } + } catch (error) { + // An aborted backstop is a cancellation, not a step-cap turn — + // keep finishReason mapping from lastStopReason, not "tool-calls". + if (options.abortSignal?.aborted || isAbortError(error)) { + hitStepLimit = false; + } + logger.warn( + "[GoogleVertex] Tools-disabled backstop call failed; falling back to cap message", + { + error: error instanceof Error ? error.message : String(error), + }, + ); + finalText = buildToolLoopCapMessage( + maxSteps, + externalToolCallCount, + ); + } + } + } + } + const responseTime = Date.now() - startTime; const externalToolCalls = allToolCalls.filter( (tc) => tc.toolName !== "final_result", @@ -5009,12 +5701,25 @@ export class GoogleVertexProvider extends BaseProvider { (te) => te.name !== "final_result", ); + // Absolute backstop — no code path may surface empty content on a turn + // that ran tools (the consumer-facing "empty response" incident shape). + if (!finalText && externalToolCalls.length > 0) { + finalText = buildToolLoopCapMessage(maxSteps, externalToolCalls.length); + } + const result: EnhancedGenerateResult = { content: finalText, - // Surface truncation: Anthropic "max_tokens" → unified "length" so the - // SDK boundary can flag/observe incomplete structured output. Anything - // else (end_turn / stop_sequence / tool_use) is a normal stop. - finishReason: lastStopReason === "max_tokens" ? "length" : "stop", + // Honest finish reason. "length" (token truncation) takes precedence; a + // step-cap exhaustion without a clean/forced answer surfaces as + // "tool-calls" (aligned with the Gemini paths, so consumers detect + // capped turns uniformly); everything else — including a successful + // forced finalization or backstop synthesis — is a normal "stop". + finishReason: + lastStopReason === "max_tokens" + ? "length" + : hitStepLimit && !synthesizedFinalAnswer + ? "tool-calls" + : "stop", provider: this.providerName, model: modelName, usage: { @@ -5677,10 +6382,39 @@ export class GoogleVertexProvider extends BaseProvider { }, }; - return { + return this.preserveStreamResultAccessors(result, { ...result, stream: wrappedIterable as StreamResult["stream"], - }; + }); + } + + /** + * Re-apply getter-based accessor properties from a source StreamResult onto + * a wrapper copy. Wrapper spreads (`{ ...result }`) invoke and SNAPSHOT + * enumerable getters at wrap time — for background-loop streams (the native + * Anthropic path) that resolve finishReason / structuredOutput / toolCalls + * only as the consumer drains, the snapshot is permanently undefined/empty. + * Copying the accessor descriptors keeps the wrapped result live. Results + * built from plain data properties (the buffered Gemini paths) have no + * getters and pass through untouched. + */ + private preserveStreamResultAccessors( + source: StreamResult, + wrapped: StreamResult, + ): StreamResult { + for (const key of [ + "finishReason", + "structuredOutput", + "toolCalls", + "toolsUsed", + "toolExecutions", + ] as const) { + const descriptor = Object.getOwnPropertyDescriptor(source, key); + if (descriptor?.get) { + Object.defineProperty(wrapped, key, descriptor); + } + } + return wrapped; } /** diff --git a/src/lib/types/stream.ts b/src/lib/types/stream.ts index 008cd19b7e..b0ea1ffaf6 100644 --- a/src/lib/types/stream.ts +++ b/src/lib/types/stream.ts @@ -642,6 +642,10 @@ export type StreamResult = { hasToolErrors?: boolean; guardrailsBlocked?: boolean; error?: string; + // Resolved finish reason for background-loop streams. Lives on metadata + // (a mutable reference the loop fills in) because result-object spreads + // in stream wrappers snapshot top-level getters before the loop resolves. + finishReason?: string; // Thought/reasoning metadata thoughtSignature?: string; thoughts?: Array<{ id?: string; type?: string; content?: string }>; diff --git a/test/continuous-test-suite-anthropic-cap.ts b/test/continuous-test-suite-anthropic-cap.ts new file mode 100644 index 0000000000..4c73734536 --- /dev/null +++ b/test/continuous-test-suite-anthropic-cap.ts @@ -0,0 +1,1299 @@ +#!/usr/bin/env tsx +/** + * Native Anthropic (Claude-on-Vertex) loop: reserved final_result step, + * graceful step-cap recovery, honest finishReason, abortSignal wiring. + * + * Covers the SPEC "UNIT TESTS" list for + * `fix(vertex): reserve final_result step + graceful cap recovery in native + * Anthropic loop`: + * - schema turns that exhaust the budget get ONE forced finalization call + * with tool_choice {type:"tool", name:"final_result"} (total create() + * calls === maxSteps) and never return content "" + * - schema turns that finish within budget are byte-identical to before + * - failed/malformed finalization → buildToolLoopCapMessage + "tool-calls" + * - non-schema pure tool-loops get a tools-disabled (tool_choice:"none") + * backstop call; on failure → cap message + "tool-calls" + * - the soft budget nudge ("Consolidate…") rides the tool_result user turn + * - abortSignal breaks the generate loop without throwing and without a + * finalization call; aborted tool calls are not recorded as failures + * - max_tokens still maps to "length" (precedence over "tool-calls") + * - the stream twin repeats the cap/finalization/backstop cases and yields + * exactly one non-empty chunk on the previously-empty cases + * + * Driven through mock-injected `client.messages.create` / `client.messages + * .stream` (the private `createAnthropicVertexClient` is overridden with a + * cast, matching the precedent in continuous-test-suite-gemini-abort.ts). + * + * Runner: `npx tsx test/continuous-test-suite-anthropic-cap.ts` + * (package.json: `pnpm run test:anthropic-cap`). + */ +import "dotenv/config"; + +import { + defineSuite, + assert, + assertEqual, + assertIncludes, +} from "./helpers/harness.js"; +import { buildToolLoopCapMessage } from "../src/lib/providers/googleNativeGemini3.js"; +import { GoogleVertexProvider } from "../src/lib/providers/googleVertex.js"; + +// --------------------------------------------------------------------------- +// Env: satisfy the GoogleVertexProvider constructor without live network. +// The mock-injected createAnthropicVertexClient means no client is ever +// really built. (Mirrors continuous-test-suite-gemini-abort.ts.) +// --------------------------------------------------------------------------- +/** Run `fn` with `updates` applied to `process.env`, restoring prior values after. */ +async function withTemporaryEnv( + updates: Record, + fn: () => Promise, +): Promise { + const previous = new Map(); + for (const key of Object.keys(updates)) { + previous.set(key, process.env[key]); + const next = updates[key]; + if (next === undefined) { + delete process.env[key]; + } else { + process.env[key] = next; + } + } + try { + return await fn(); + } finally { + for (const [key, value] of previous) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } + } +} + +const VERTEX_ENV = { + GOOGLE_APPLICATION_CREDENTIALS: "/tmp/neurolink-anthropic-cap-creds.json", + GOOGLE_CLOUD_PROJECT_ID: "test-project", + GOOGLE_CLOUD_PROJECT: "test-project", + GOOGLE_CLOUD_LOCATION: "global", +}; + +const MODEL = "claude-sonnet-4-5@20250929"; + +// --------------------------------------------------------------------------- +// Mock @anthropic-ai/vertex-sdk client +// --------------------------------------------------------------------------- + +type AnthropicRequestParams = { + model: string; + max_tokens: number; + messages: Array<{ role: string; content: unknown }>; + tools?: Array<{ name: string }>; + tool_choice?: { type: string; name?: string }; + system?: unknown; +}; + +type AnthropicResponse = { + content: Array>; + usage?: { + input_tokens?: number; + output_tokens?: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + }; + stop_reason?: string | null; +}; + +/** A response whose only content is a tool_use call (the "runaway" step). */ +function toolUseResponse( + name: string, + input: Record, + extraText?: string, +): AnthropicResponse { + return { + content: [ + ...(extraText ? [{ type: "text", text: extraText }] : []), + { type: "tool_use", id: `tu_${name}_${Math.random()}`, name, input }, + ], + usage: { input_tokens: 5, output_tokens: 3 }, + stop_reason: "tool_use", + }; +} + +/** A final_result tool_use response (what a forced finalization returns). */ +function finalResultResponse( + payload: Record, + stopReason = "tool_use", +): AnthropicResponse { + return { + content: [ + { + type: "tool_use", + id: "tu_final", + name: "final_result", + input: payload, + }, + ], + usage: { input_tokens: 5, output_tokens: 7 }, + stop_reason: stopReason, + }; +} + +/** A plain-text terminal response (model stopped on its own). */ +function textResponse( + text: string, + stopReason = "end_turn", +): AnthropicResponse { + return { + content: text ? [{ type: "text", text }] : [], + usage: { input_tokens: 5, output_tokens: 7 }, + stop_reason: stopReason, + }; +} + +function abortError(message = "Request was aborted."): Error { + const e = new Error(message); + e.name = "AbortError"; + return e; +} + +/** + * Behavior for one mocked model call: a response to return, an error to + * throw, and (stream mock only) text deltas fired at the "text" listener + * before finalMessage resolves. + */ +type CallBehavior = { + response?: AnthropicResponse; + reject?: Error; + textDeltas?: string[]; + /** Side effect executed as the call is issued (e.g. trip an AbortController). */ + onCall?: () => void; +}; + +type GenerateRecorder = { + calls: AnthropicRequestParams[]; + /** The request-options AbortSignal forwarded to each create() call. */ + signals: Array; + client: { + messages: { + create: ( + p: AnthropicRequestParams, + opts?: { signal?: AbortSignal }, + ) => Promise; + }; + }; +}; + +/** Recording mock for the generate path (client.messages.create). */ +function makeGenerateMock( + plan: (callIndex: number, params: AnthropicRequestParams) => CallBehavior, +): GenerateRecorder { + const calls: AnthropicRequestParams[] = []; + const signals: Array = []; + return { + calls, + signals, + client: { + messages: { + create: async ( + p: AnthropicRequestParams, + opts?: { signal?: AbortSignal }, + ) => { + const index = calls.length; + calls.push(p); + signals.push(opts?.signal); + const behavior = plan(index, p); + behavior.onCall?.(); + if (behavior.reject) { + throw behavior.reject; + } + return behavior.response as AnthropicResponse; + }, + }, + }, + }; +} + +type MockMessageStream = { + on: (event: string, cb: (delta: string) => void) => MockMessageStream; + controller: { abort: () => void }; + finalMessage: () => Promise; +}; + +type StreamRecorder = { + calls: AnthropicRequestParams[]; + client: { + messages: { + stream: (p: AnthropicRequestParams) => MockMessageStream; + }; + }; +}; + +/** Recording mock for the stream path (client.messages.stream). */ +function makeStreamMock( + plan: (callIndex: number, params: AnthropicRequestParams) => CallBehavior, +): StreamRecorder { + const calls: AnthropicRequestParams[] = []; + return { + calls, + client: { + messages: { + stream: (p: AnthropicRequestParams) => { + const index = calls.length; + calls.push(p); + const behavior = plan(index, p); + behavior.onCall?.(); + const textListeners: Array<(delta: string) => void> = []; + const stream: MockMessageStream = { + on(event: string, cb: (delta: string) => void) { + if (event === "text") { + textListeners.push(cb); + } + return stream; + }, + controller: { + abort: () => { + /* mid-flight cancel is simulated via behavior.reject */ + }, + }, + async finalMessage() { + if (behavior.reject) { + throw behavior.reject; + } + for (const delta of behavior.textDeltas ?? []) { + for (const cb of textListeners) { + cb(delta); + } + } + return behavior.response as AnthropicResponse; + }, + }; + return stream; + }, + }, + }, + }; +} + +/** Construct a provider with the mock Anthropic client injected. */ +async function makeProvider(client: unknown): Promise { + const provider = new GoogleVertexProvider( + MODEL, + undefined, + undefined, + "global", + ); + ( + provider as unknown as { + createAnthropicVertexClient: () => Promise; + } + ).createAnthropicVertexClient = async () => client; + return provider; +} + +type GenResult = { + content: string; + finishReason: string; + usage: { input: number; output: number; total: number }; + toolsUsed?: string[]; + toolExecutions?: unknown[]; + enhancedWithTools?: boolean; +}; + +/** Invoke the private native-Anthropic generate loop against the mock client. */ +async function runGenerate( + provider: GoogleVertexProvider, + options: Record, +): Promise { + return ( + provider as unknown as { + executeNativeAnthropicGenerate(o: unknown): Promise; + } + ).executeNativeAnthropicGenerate(options); +} + +type StreamResultShape = { + finishReason?: string; + usage: { input: number; output: number; total: number }; + structuredOutput?: unknown; + stream: AsyncIterable<{ content: string }>; +}; + +/** Invoke the private native-Anthropic stream loop and drain its chunks plus result. */ +async function runStream( + provider: GoogleVertexProvider, + options: Record, +): Promise<{ chunks: string[]; result: StreamResultShape }> { + const result = (await ( + provider as unknown as { + executeNativeAnthropicStream(o: unknown): Promise; + } + ).executeNativeAnthropicStream(options)) as StreamResultShape; + const collected: string[] = []; + for await (const c of result.stream) { + collected.push(c.content); + } + return { chunks: collected, result }; +} + +/** A simple echo tool the loop can execute. */ +function makeTool( + execute: ( + args: unknown, + opts: { abortSignal?: AbortSignal }, + ) => Promise = async () => ({ ok: true }), +) { + return { + myTool: { + description: "A test tool", + parameters: { + type: "object", + properties: { q: { type: "string" } }, + }, + execute, + }, + }; +} + +/** Extract every "Consolidate" nudge text from a request's trailing user turn. */ +function nudgeTextsInLastUserTurn(params: AnthropicRequestParams): string[] { + const last = params.messages[params.messages.length - 1]; + if (!last || last.role !== "user" || !Array.isArray(last.content)) { + return []; + } + return (last.content as Array>) + .filter( + (block) => + block.type === "text" && + typeof block.text === "string" && + (block.text as string).includes("Consolidate"), + ) + .map((block) => block.text as string); +} + +// =========================================================================== +// Suite +// =========================================================================== + +const { test, runSuite } = defineSuite("Anthropic-on-Vertex Cap + Abort"); + +/** Registers and runs every case in the Anthropic cap/finalization/abort suite. */ +async function main(): Promise { + const { z } = await import("zod"); + const SCHEMA = z.object({ answer: z.number() }); + const PAYLOAD = { answer: 42 }; + + // ------------------------------------------------------------------------- + // 1. Generate + schema: budget exhausted -> reserved forced finalization + // ------------------------------------------------------------------------- + await test("generate schema: never calls final_result -> forced finalization call, content=JSON, finishReason 'stop', create() calls === maxSteps", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 3; + // Agentic steps (indices 0..maxSteps-2) return runaway tool calls; the + // reserved finalization call (index maxSteps-1) returns final_result. + const mock = makeGenerateMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual( + mock.calls.length, + maxSteps, + "agentic steps + 1 reserved finalization call", + ); + // Loop steps force a tool call via tool_choice:"any"; the reserved + // finalization call pins final_result. + assertEqual(mock.calls[0].tool_choice?.type, "any"); + assertEqual(mock.calls[maxSteps - 1].tool_choice?.type, "tool"); + assertEqual(mock.calls[maxSteps - 1].tool_choice?.name, "final_result"); + assertEqual(result.content, JSON.stringify(PAYLOAD)); + assertEqual( + JSON.stringify(JSON.parse(result.content)), + JSON.stringify(PAYLOAD), + "content must be valid structured JSON", + ); + assertEqual(result.finishReason, "stop"); + assert( + !(result.toolsUsed ?? []).includes("final_result"), + "final_result must stay filtered from toolsUsed", + ); + assert( + result.usage.output >= 3 * 2 + 7 - 3, + "finalization usage counted", + ); + }); + }); + + // ------------------------------------------------------------------------- + // 2. Generate + schema: final_result within budget -> unchanged behavior + // ------------------------------------------------------------------------- + await test("generate schema: final_result at step 2 -> no finalization call, create() calls === 2", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeGenerateMock((i) => + i === 0 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps: 5, + }); + assertEqual( + mock.calls.length, + 2, + "breaks at final_result, no extra call", + ); + assert( + mock.calls.every((c) => c.tool_choice?.type !== "tool"), + "no forced-finalization call on a clean finish", + ); + assertEqual(result.content, JSON.stringify(PAYLOAD)); + assertEqual(result.finishReason, "stop"); + }); + }); + + // ------------------------------------------------------------------------- + // 3. Generate + schema: maxSteps=1 -> finalization IS the only call + // ------------------------------------------------------------------------- + await test("generate schema: maxSteps=1 -> the finalization call is the only call", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeGenerateMock(() => ({ + response: finalResultResponse(PAYLOAD), + })); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps: 1, + }); + assertEqual(mock.calls.length, 1); + assertEqual(mock.calls[0].tool_choice?.type, "tool"); + assertEqual(mock.calls[0].tool_choice?.name, "final_result"); + assertEqual(result.content, JSON.stringify(PAYLOAD)); + assertEqual(result.finishReason, "stop"); + }); + }); + + // ------------------------------------------------------------------------- + // 4. Generate + schema: finalization throws -> cap message, "tool-calls" + // ------------------------------------------------------------------------- + await test("generate schema: finalization call throws -> cap message, finishReason 'tool-calls', never ''", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { reject: new Error("boom from finalization") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(result.content, buildToolLoopCapMessage(maxSteps, 1)); + assertEqual(result.finishReason, "tool-calls"); + assert(result.content.length > 0, "content must never be empty"); + }); + }); + + await test("generate schema: finalization returns malformed (no tool_use) -> cap message, 'tool-calls'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: textResponse("not a tool call", "end_turn") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(result.content, buildToolLoopCapMessage(maxSteps, 1)); + assertEqual(result.finishReason, "tool-calls"); + }); + }); + + // ------------------------------------------------------------------------- + // 5. Generate non-schema: pure tool-loop to cap -> tool_choice:"none" backstop + // ------------------------------------------------------------------------- + await test("generate non-schema: pure tool-loop to cap -> tools-disabled backstop call, its text returned, finishReason 'stop'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => + i < maxSteps + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: textResponse("SYNTHESIZED ANSWER", "end_turn") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps + 1, "loop + 1 backstop call"); + const backstop = mock.calls[maxSteps]; + // Anthropic rejects tool_use/tool_result history without a tools param, + // so "tools disabled" must be tool_choice:"none" with tools retained. + assertEqual(backstop.tool_choice?.type, "none"); + assert( + Array.isArray(backstop.tools) && backstop.tools.length > 0, + "backstop must keep the tools param (API validity + cache prefix)", + ); + assertIncludes( + JSON.stringify(backstop.system ?? ""), + "no longer available", + ); + assertEqual(result.content, "SYNTHESIZED ANSWER"); + // synthesizedFinalAnswer -> "stop", parity with the Gemini synth path. + assertEqual(result.finishReason, "stop"); + assert(result.usage.output > 3 * maxSteps, "backstop tokens added"); + }); + }); + + await test("generate non-schema: backstop fails -> cap message, finishReason 'tool-calls'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => + i < maxSteps + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { reject: new Error("backstop boom") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps, + }); + assertEqual(result.content, buildToolLoopCapMessage(maxSteps, maxSteps)); + assertEqual(result.finishReason, "tool-calls"); + }); + }); + + await test("generate non-schema: capped with mid-loop prose -> prose returned, 'tool-calls', no backstop call", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => { + if (i < maxSteps) { + return { + response: toolUseResponse("myTool", { q: "x" }, `step${i} prose`), + }; + } + return { response: textResponse("SHOULD NOT BE CALLED") }; + }); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps, "no backstop when prose exists"); + // Prose is ACCUMULATED across steps (appendStepText), not last-step-only. + assertEqual(result.content, "step0 prose\nstep1 prose"); + assertEqual(result.finishReason, "tool-calls"); + }); + }); + + // ------------------------------------------------------------------------- + // 6. Budget nudge + // ------------------------------------------------------------------------- + await test("generate schema: budget nudge appended to tool_result turns when steps-remaining <= 3", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 5; // agentic budget 4, finalization reserved + const mock = makeGenerateMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps); + // Request 0 carries no tool results yet -> no nudge anywhere. + assertEqual(nudgeTextsInLastUserTurn(mock.calls[0]).length, 0); + // After step 1 (3 agentic steps remain) the tool_result turn seen by + // request 1 carries the nudge; same for 2 and 1 remaining. + for (const [callIndex, remaining] of [ + [1, 3], + [2, 2], + [3, 1], + ] as const) { + const nudges = nudgeTextsInLastUserTurn(mock.calls[callIndex]); + assert( + nudges.length === 1, + `request ${callIndex} must carry exactly one trailing nudge`, + ); + assertIncludes(nudges[0], `Only ${remaining} tool step(s) remain`); + assertIncludes(nudges[0], "call final_result with your best answer"); + } + }); + }); + + // ------------------------------------------------------------------------- + // 7. Abort wiring (generate) + // ------------------------------------------------------------------------- + await test("generate: abort mid-turn -> loop breaks (no throw), no finalization call, cap message", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 10; + const controller = new AbortController(); + const mock = makeGenerateMock((i) => { + if (i === 0) { + return { response: toolUseResponse("myTool", { q: "x" }) }; + } + // Caller aborts right as the 2nd request starts; SDK rejects abort-shaped. + return { + onCall: () => controller.abort(), + reject: abortError(), + }; + }); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + abortSignal: controller.signal, + }); + // No throw escaped (we got a result); calls stopped immediately. + assertEqual(mock.calls.length, 2, "create() calls stop on abort"); + assert( + mock.calls.every((c) => c.tool_choice?.type !== "tool"), + "no finalization call after abort (budget already blown)", + ); + // The caller's signal must actually be forwarded as the SDK request + // option — not just simulated by the mock's injected rejection. + assert( + mock.signals.every((s) => s === controller.signal), + "options.abortSignal must ride every create() as the request signal", + ); + assertIncludes(result.content, "step limit for a single turn"); + // Aborted turns map finishReason from lastStopReason, not "tool-calls". + assertEqual(result.finishReason, "stop"); + }); + }); + + await test("generate: pre-aborted signal -> zero model calls, cap message", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const controller = new AbortController(); + controller.abort(); + const mock = makeGenerateMock(() => ({ + response: toolUseResponse("myTool", { q: "x" }), + })); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps: 5, + abortSignal: controller.signal, + }); + assertEqual(mock.calls.length, 0, "no request when already aborted"); + assertIncludes(result.content, "step limit for a single turn"); + }); + }); + + await test("generate: aborted tool call not recorded as a failure; turn breaks", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + // No caller signal here: the tool throws an AbortError-named error, so + // isAbortError is the SOLE discriminator that must break the turn. + let executeCount = 0; + const mock = makeGenerateMock(() => ({ + response: toolUseResponse("myTool", { q: "x" }), + })); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(async () => { + executeCount++; + throw abortError("aborted mid-tool"); + }), + maxSteps: 6, + }); + assertEqual(executeCount, 1, "aborted tool must not be retried"); + assertEqual( + mock.calls.length, + 1, + "aborted tool must break the turn — no further model request", + ); + assert( + !JSON.stringify(result.toolExecutions ?? []).includes( + "Error executing tool", + ), + "aborted tool must not be recorded as a failed execution", + ); + assertIncludes(result.content, "step limit for a single turn"); + }); + }); + + // ------------------------------------------------------------------------- + // 8. max_tokens precedence + // ------------------------------------------------------------------------- + await test("generate: max_tokens stop_reason maps to 'length' even when the cap was hit", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeGenerateMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD, "max_tokens") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(result.finishReason, "length"); + assertEqual(result.content, JSON.stringify(PAYLOAD)); + }); + }); + + // ------------------------------------------------------------------------- + // 9. Regression: no empty content while tools ran + // ------------------------------------------------------------------------- + await test("generate: clean text-exit with EMPTY text after tool steps -> cap message, never ''", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeGenerateMock((i) => + i === 0 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: textResponse("", "end_turn") }, + ); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps: 5, + }); + assert( + result.content.length > 0, + "content must never be empty when enhancedWithTools is true", + ); + assertEqual(result.enhancedWithTools, true); + }); + }); + + // ------------------------------------------------------------------------- + // 10. Stream twin: cap -> forced finalization, exactly one JSON chunk + // ------------------------------------------------------------------------- + await test("stream schema: cap -> forced finalization, exactly ONE JSON chunk, structuredOutput set, finishReason 'stop'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 3; + const mock = makeStreamMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps); + assertEqual(mock.calls[maxSteps - 1].tool_choice?.type, "tool"); + assertEqual(mock.calls[maxSteps - 1].tool_choice?.name, "final_result"); + assertEqual(chunks.length, 1, "exactly one non-empty chunk"); + assertEqual(chunks[0], JSON.stringify(PAYLOAD)); + assertEqual( + JSON.stringify(result.structuredOutput), + JSON.stringify(PAYLOAD), + ); + assertEqual(result.finishReason, "stop"); + }); + }); + + await test("stream schema: finalization fails -> exactly ONE cap chunk, finishReason 'tool-calls'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeStreamMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { reject: new Error("finalization boom") }, + ); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(chunks.length, 1, "exactly one graceful chunk"); + assertEqual(chunks[0], buildToolLoopCapMessage(maxSteps, 1)); + assert(chunks[0].length > 0, "chunk must be non-empty"); + assertEqual(result.finishReason, "tool-calls"); + }); + }); + + await test("stream schema: final_result within budget -> unchanged (one JSON chunk, no finalization call)", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeStreamMock((i) => + i === 0 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps: 5, + }); + assertEqual(mock.calls.length, 2); + assert( + mock.calls.every((c) => c.tool_choice?.type !== "tool"), + "no forced-finalization call on a clean finish", + ); + assertEqual(chunks.length, 1); + assertEqual(chunks[0], JSON.stringify(PAYLOAD)); + assertEqual(result.finishReason, "stop"); + }); + }); + + // ------------------------------------------------------------------------- + // 11. Stream twin: non-schema backstop + // ------------------------------------------------------------------------- + await test("stream non-schema: pure tool-loop to cap -> tool_choice:'none' backstop, ONE text chunk, finishReason 'stop'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeStreamMock((i) => + i < maxSteps + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: textResponse("SYNTHESIZED ANSWER", "end_turn") }, + ); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps + 1, "loop + 1 backstop call"); + assertEqual(mock.calls[maxSteps].tool_choice?.type, "none"); + assert( + Array.isArray(mock.calls[maxSteps].tools) && + (mock.calls[maxSteps].tools as unknown[]).length > 0, + "backstop must keep the tools param", + ); + assertEqual(chunks.length, 1, "backstop text pushed as one chunk"); + assertEqual(chunks[0], "SYNTHESIZED ANSWER"); + assertEqual(result.finishReason, "stop"); + }); + }); + + await test("stream non-schema: backstop fails -> ONE cap chunk, finishReason 'tool-calls'", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeStreamMock((i) => + i < maxSteps + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { reject: new Error("backstop boom") }, + ); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps, + }); + assertEqual(chunks.length, 1); + assertEqual(chunks[0], buildToolLoopCapMessage(maxSteps, maxSteps)); + assertEqual(result.finishReason, "tool-calls"); + }); + }); + + // ------------------------------------------------------------------------- + // 12. Stream twin: abort -> graceful single chunk, no stream error + // ------------------------------------------------------------------------- + await test("stream: abort mid-turn -> ONE cap chunk, stream closes without error, no finalization call", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 10; + const controller = new AbortController(); + const mock = makeStreamMock((i) => { + if (i === 0) { + return { response: toolUseResponse("myTool", { q: "x" }) }; + } + return { + onCall: () => controller.abort(), + reject: abortError(), + }; + }); + const provider = await makeProvider(mock.client); + const { chunks, result } = await runStream(provider, { + input: { text: "run tools" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + abortSignal: controller.signal, + }); + assertEqual(mock.calls.length, 2, "calls stop on abort"); + assert( + mock.calls.every((c) => c.tool_choice?.type !== "tool"), + "no finalization call after abort", + ); + assertEqual(chunks.length, 1, "exactly one graceful chunk on abort"); + assertIncludes(chunks[0], "step limit for a single turn"); + assertEqual(result.finishReason, "stop"); + }); + }); + + // ------------------------------------------------------------------------- + // 13. Stream twin: nudge parity + // ------------------------------------------------------------------------- + await test("stream schema: budget nudge appended to tool_result turns when steps-remaining <= 3", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 5; // agentic budget 4 + const mock = makeStreamMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + await runStream(provider, { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + }); + assertEqual(mock.calls.length, maxSteps); + const nudges = nudgeTextsInLastUserTurn(mock.calls[1]); + assertEqual(nudges.length, 1, "request 1 carries the first nudge"); + assertIncludes(nudges[0], "Only 3 tool step(s) remain"); + }); + }); + + // ------------------------------------------------------------------------- + // 14. Review fixes: finishReason survives wrapper spreads via metadata + // ------------------------------------------------------------------------- + await test("stream: metadata.finishReason survives a pre-drain result spread (wrapper-spread snapshot)", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeStreamMock(() => ({ + response: textResponse("plain answer", "end_turn"), + })); + const provider = await makeProvider(mock.client); + const result = (await ( + provider as unknown as { + executeNativeAnthropicStream(o: unknown): Promise; + } + ).executeNativeAnthropicStream({ + input: { text: "hi" }, + maxSteps: 3, + })) as StreamResultShape & { metadata?: { finishReason?: string } }; + // Simulate the pre-existing stream wrappers, which spread the result + // BEFORE the background loop resolves. + const dto = { ...result }; + for await (const _chunk of result.stream) { + // drain + } + assertEqual( + (dto as { metadata?: { finishReason?: string } }).metadata + ?.finishReason, + "stop", + "metadata.finishReason must survive the spread (mutable reference)", + ); + assertEqual( + result.finishReason, + "stop", + "top-level getter resolves on the original result", + ); + }); + }); + + // ------------------------------------------------------------------------- + // 15. Review fixes: aborted-step completed tools still persisted + // ------------------------------------------------------------------------- + await test("generate: tools that completed in an aborted step are still persisted to conversation memory", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const stored: Array<{ calls: unknown[]; results: unknown[] }> = []; + const mock = makeGenerateMock(() => ({ + response: { + content: [ + { type: "tool_use", id: "tu_ok", name: "okTool", input: {} }, + { type: "tool_use", id: "tu_ab", name: "abortTool", input: {} }, + ], + usage: { input_tokens: 5, output_tokens: 3 }, + stop_reason: "tool_use", + }, + })); + const provider = await makeProvider(mock.client); + ( + provider as unknown as { + handleToolExecutionStorage: ( + calls: unknown[], + results: unknown[], + ) => Promise; + } + ).handleToolExecutionStorage = async (calls, results) => { + stored.push({ calls, results }); + }; + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: { + okTool: { + description: "completes", + parameters: { type: "object", properties: {} }, + execute: async () => ({ ok: true }), + }, + abortTool: { + description: "throws AbortError", + parameters: { type: "object", properties: {} }, + execute: async () => { + throw abortError("aborted mid-tool"); + }, + }, + }, + maxSteps: 6, + }); + assertEqual(mock.calls.length, 1, "turn breaks after the aborted tool"); + assertEqual(stored.length, 1, "storage hook must still run"); + const storedJson = JSON.stringify(stored[0]); + assertIncludes(storedJson, "okTool"); + assert( + JSON.stringify(stored[0].results).includes("okTool"), + "the completed tool's result must be persisted", + ); + // Pairing: the aborted call's row must carry a neutral cancellation + // result — an unpaired tool_use replayed next turn would 400. + assertEqual( + stored[0].calls.length, + stored[0].results.length, + "every persisted tool call must have a paired result", + ); + const abortedResult = JSON.stringify( + (stored[0].results as Array<{ toolName?: string }>).find( + (r) => r.toolName === "abortTool", + ), + ); + assertIncludes(abortedResult, '"aborted":true'); + assert( + !abortedResult.includes("Error executing tool"), + "the aborted call must not be persisted as a tool failure", + ); + assert(result.content.length > 0, "graceful content after abort"); + }); + }); + + // ------------------------------------------------------------------------- + // 16. Review fixes: abort prefers already-produced prose (generate) + // ------------------------------------------------------------------------- + await test("generate: abort mid-turn with prior prose -> prose returned, not the cap message", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const controller = new AbortController(); + const mock = makeGenerateMock((i) => { + if (i === 0) { + return { + response: toolUseResponse("myTool", { q: "x" }, "partial analysis"), + }; + } + return { onCall: () => controller.abort(), reject: abortError() }; + }); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps: 10, + abortSignal: controller.signal, + }); + assertEqual( + result.content, + "partial analysis", + "abort must not discard prose the model already produced", + ); + assert( + !result.content.includes("step limit"), + "no fabricated step-limit claim when prose exists", + ); + }); + }); + + // ------------------------------------------------------------------------- + // 17. Review fixes: abort landing in the FINAL budgeted step's tool exec + // ------------------------------------------------------------------------- + await test("generate schema: abort during last budgeted step's tool exec -> terminal re-check skips the finalization call", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const controller = new AbortController(); + const mock = makeGenerateMock(() => ({ + response: toolUseResponse("myTool", { q: "x" }), + })); + const provider = await makeProvider(mock.client); + const result = await runGenerate(provider, { + input: { text: "give me the answer" }, + tools: makeTool(async () => { + // Tool ignores the signal but the caller aborts while it runs; it + // still returns normally, so the loop exits by budget, not abort. + controller.abort(); + return { ok: true }; + }), + schema: SCHEMA, + maxSteps: 2, // agentic budget 1 — this tool step is the last one + abortSignal: controller.signal, + }); + assertEqual( + mock.calls.length, + 1, + "no terminal finalization call after a late-landing abort", + ); + assert( + mock.calls.every((c) => c.tool_choice?.type !== "tool"), + "tool_choice:'tool' must never be issued post-abort", + ); + assertIncludes(result.content, "step limit for a single turn"); + }); + }); + + await test("stream schema: abort during last budgeted step's tool exec -> no terminal call, one cap chunk", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const controller = new AbortController(); + const mock = makeStreamMock(() => ({ + response: toolUseResponse("myTool", { q: "x" }), + })); + const provider = await makeProvider(mock.client); + const { chunks } = await runStream(provider, { + input: { text: "give me the answer" }, + tools: makeTool(async () => { + controller.abort(); + return { ok: true }; + }), + schema: SCHEMA, + maxSteps: 2, + abortSignal: controller.signal, + }); + assertEqual(mock.calls.length, 1, "no terminal model call post-abort"); + assertEqual(chunks.length, 1, "exactly one graceful chunk"); + assertIncludes(chunks[0], "step limit for a single turn"); + }); + }); + + // ------------------------------------------------------------------------- + // 18. Review fixes: live deltas already delivered suppress the abort cap + // ------------------------------------------------------------------------- + await test("stream: abort after live text deltas were delivered -> no contradictory cap chunk appended", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const controller = new AbortController(); + const mock = makeStreamMock((i) => { + if (i === 0) { + // Deltas reach the consumer live even though the response carries + // only a tool_use block (aggregatedTurnText stays empty). + return { + textDeltas: ["partial answer already delivered"], + response: toolUseResponse("myTool", { q: "x" }), + }; + } + return { onCall: () => controller.abort(), reject: abortError() }; + }); + const provider = await makeProvider(mock.client); + const { chunks } = await runStream(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps: 10, + abortSignal: controller.signal, + }); + assertEqual(chunks.length, 1, "only the live delta reaches the consumer"); + assertEqual(chunks[0], "partial answer already delivered"); + assert( + chunks.every((c) => !c.includes("step limit")), + "no cap chunk after delivered prose", + ); + }); + }); + + // ------------------------------------------------------------------------- + // 19. Review fixes: stream absolute empty backstop + // ------------------------------------------------------------------------- + await test("stream non-schema: clean text-exit with EMPTY content after tool steps -> one cap chunk, never zero chunks", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const mock = makeStreamMock((i) => + i === 0 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: textResponse("", "end_turn") }, + ); + const provider = await makeProvider(mock.client); + const { chunks } = await runStream(provider, { + input: { text: "run tools" }, + tools: makeTool(), + maxSteps: 5, + }); + assertEqual( + chunks.length, + 1, + "stream must never close empty on a turn that ran tools", + ); + assert(chunks[0].length > 0, "backstop chunk must be non-empty"); + }); + }); + + // ------------------------------------------------------------------------- + // 20. Review fixes: lifecycle wrapping preserves live result accessors + // ------------------------------------------------------------------------- + await test("stream: wrapStreamResultWithLifecycle preserves finishReason/structuredOutput getters through its spread", async () => { + await withTemporaryEnv(VERTEX_ENV, async () => { + const maxSteps = 2; + const mock = makeStreamMock((i) => + i < maxSteps - 1 + ? { response: toolUseResponse("myTool", { q: "x" }) } + : { response: finalResultResponse(PAYLOAD) }, + ); + const provider = await makeProvider(mock.client); + const options = { + input: { text: "give me the answer" }, + tools: makeTool(), + schema: SCHEMA, + maxSteps, + // A lifecycle callback forces the wrapper to take its spread path. + onFinish: async () => undefined, + }; + const inner = provider as unknown as { + executeNativeAnthropicStream(o: unknown): Promise; + } as { + executeNativeAnthropicStream(o: unknown): Promise; + wrapStreamResultWithLifecycle?: unknown; + }; + const result = await inner.executeNativeAnthropicStream(options); + const wrapped = ( + provider as unknown as { + wrapStreamResultWithLifecycle( + o: unknown, + r: StreamResultShape, + t: number, + ): StreamResultShape & { structuredOutput?: unknown }; + } + ).wrapStreamResultWithLifecycle(options, result, Date.now()); + assert( + wrapped !== result, + "wrapper must have taken the spread path (onFinish present)", + ); + for await (const _chunk of wrapped.stream) { + // drain through the WRAPPED iterable + } + assertEqual( + wrapped.finishReason, + "stop", + "finishReason getter must survive the lifecycle wrapper spread", + ); + assertEqual( + JSON.stringify(wrapped.structuredOutput), + JSON.stringify(PAYLOAD), + "structuredOutput getter must survive the lifecycle wrapper spread", + ); + }); + }); +} + +await runSuite(main);