diff --git a/CHANGELOG.md b/CHANGELOG.md index fa1783f046..22e9eb8ab5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,37 @@ versions are listed at ## Unreleased +### Changed: a response cut at the output token limit reports `PROVIDER_OUTPUT_TRUNCATED` + +An Anthropic response that stops at the output token limit part way through a +tool call now fails with the new curated code `PROVIDER_OUTPUT_TRUNCATED` and +the message "The model stopped at its output token limit before it finished +the response." It previously surfaced as a malformed provider stream, and on +agents that use provider replay checkpoints it surfaced as "Provider replay +turn failed before its boundary", neither of which named the real cause. + +The failure is terminal, not retryable: the same request and the same output +token budget truncate again. Raise the model output token limit, or ask for a +shorter response. This is a deliberate retry-semantics change and needs your +decision if you depend on the old behaviour: `PROVIDER_OUTPUT_TRUNCATED` joins +the curated provider failure codes, so it is classified as a known terminal +error and hosted child runs -- including durable child forks -- stop retrying +it, where a truncation previously landed in the unknown, retryable +`PROVIDER_STREAM_ERROR` bucket. A retry above temperature 0 could occasionally have produced a shorter +tool input and succeeded; that accidental recovery is gone, in exchange for a +named failure instead of a retry loop against a budget that cannot fit the +response. The incomplete tool call is also dropped rather than replayed, so no +partial tool input reaches a tool, and no later tool call from the same +truncated turn is dispatched. + +A replay checkpoint boundary that fails now reports the provider failure the +stream reports instead of a fixed message. Anything that matched on the literal +string "Provider replay turn failed before its boundary" must match on the run +error code instead. That string remains only as the neutral fallback for a turn +that ends with no reported cause, such as a client cancellation, and a failure +to persist a durable run event now reports +`DURABLE_RUN_EVENT_PERSISTENCE_FAILED` rather than a provider failure. + ### Changed: `veryfront up` pushes committed work again `veryfront up` now pushes the local source to main whenever the checkout no diff --git a/docs/api-reference/veryfront/provider.md b/docs/api-reference/veryfront/provider.md index 1a8e25c36b..63724f2bac 100644 --- a/docs/api-reference/veryfront/provider.md +++ b/docs/api-reference/veryfront/provider.md @@ -205,13 +205,14 @@ import { #### Classes -| Name | Description | Source | -| ------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | -| `ProviderError` | Base class for typed provider errors. The `retryable` flag is the primary signal for callers (or a retry wrapper) to decide whether to re-issue the request. `retryAfterMs` is set when the provider gave an explicit delay hint (Retry-After header, Retry-Info trailer). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | -| `ProviderOverloadedError` | Provider reports it is overloaded (Anthropic 529, OpenAI/Google 503). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | -| `ProviderQuotaError` | Provider account quota is exhausted - non-retryable. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | -| `ProviderRateLimitError` | Provider is rate limiting this API key (OpenAI/Google 429 with Retry-After). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | -| `ProviderRequestError` | Non-retryable 4xx/5xx that doesn't fit another bucket. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| Name | Description | Source | +| ------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | +| `ProviderError` | Base class for typed provider errors. The `retryable` flag is the primary signal for callers (or a retry wrapper) to decide whether to re-issue the request. `retryAfterMs` is set when the provider gave an explicit delay hint (Retry-After header, Retry-Info trailer). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| `ProviderOutputTruncatedError` | Provider stopped generating at the output token limit, leaving the response incomplete (for example a `tool_use` block whose input JSON never closed). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| `ProviderOverloadedError` | Provider reports it is overloaded (Anthropic 529, OpenAI/Google 503). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| `ProviderQuotaError` | Provider account quota is exhausted - non-retryable. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| `ProviderRateLimitError` | Provider is rate limiting this API key (OpenAI/Google 429 with Retry-After). | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | +| `ProviderRequestError` | Non-retryable 4xx/5xx that doesn't fit another bucket. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/provider/runtime-loader/provider-http.ts) | #### Types diff --git a/extensions/ext-llm-anthropic/src/anthropic-stream.test.ts b/extensions/ext-llm-anthropic/src/anthropic-stream.test.ts index 559d582dcd..4f76289747 100644 --- a/extensions/ext-llm-anthropic/src/anthropic-stream.test.ts +++ b/extensions/ext-llm-anthropic/src/anthropic-stream.test.ts @@ -1,6 +1,10 @@ -import { assertEquals, assertRejects } from "#veryfront/testing/assert.ts"; +import { assertEquals, assertInstanceOf, assertRejects } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; -import { ProviderOverloadedError, ProviderRequestError } from "veryfront/provider/shared"; +import { + ProviderOutputTruncatedError, + ProviderOverloadedError, + ProviderRequestError, +} from "veryfront/provider/shared"; import { addAnthropicUsage, extractAnthropicUsage, @@ -906,6 +910,228 @@ describe("ext-llm-anthropic/anthropic-stream", () => { ); }); + it("classifies a max_tokens-truncated tool_use as a provider output truncation", async () => { + const truncatedToolStream = [ + data({ type: "message_start", message: { usage: { input_tokens: 1 } } }), + data({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_x", name: "create_file", input: {} }, + }), + data({ + type: "content_block_delta", + index: 0, + delta: { + type: "input_json_delta", + partial_json: '{"path":"/inbox/mail-1.json","content":"truncated', + }, + }), + data({ type: "content_block_stop", index: 0 }), + data({ + type: "message_delta", + delta: { stop_reason: "max_tokens" }, + usage: { output_tokens: 4096 }, + }), + data({ type: "message_stop" }), + ].join(""); + + const error = await assertRejects( + () => collectParts(streamFromText(truncatedToolStream)), + ProviderOutputTruncatedError, + "provider output truncated at the max output token limit", + ); + assertInstanceOf(error, ProviderOutputTruncatedError); + assertEquals(error.retryable, false); + assertEquals( + error.message.includes("tool call arguments were not valid JSON object text"), + false, + ); + }); + + // Codex P2 on veryfront-code#4516: a usage-only message_delta carries no + // stop_reason, and deciding there would classify the truncation as a + // malformed stream before the delta that actually says max_tokens. + it("waits for a stop reason when a usage-only delta arrives first", async () => { + const lateStopReasonStream = [ + data({ type: "message_start", message: { usage: { input_tokens: 1 } } }), + data({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_z", name: "create_file", input: {} }, + }), + data({ + type: "content_block_delta", + index: 0, + delta: { + type: "input_json_delta", + partial_json: '{"path":"/inbox/mail-2.json","content":"trunc', + }, + }), + data({ type: "content_block_stop", index: 0 }), + // Usage only -- no stop_reason yet. + data({ type: "message_delta", delta: {}, usage: { output_tokens: 2048 } }), + data({ + type: "message_delta", + delta: { stop_reason: "max_tokens" }, + usage: { output_tokens: 4096 }, + }), + data({ type: "message_stop" }), + ].join(""); + + const error = await assertRejects( + () => collectParts(streamFromText(lateStopReasonStream)), + ProviderOutputTruncatedError, + "provider output truncated at the max output token limit", + ); + assertInstanceOf(error, ProviderOutputTruncatedError); + }); + + it("still reports a malformed tool stream when the stop reason is not max_tokens", async () => { + const malformedToolStream = [ + data({ type: "message_start", message: { usage: { input_tokens: 1 } } }), + data({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_y", name: "create_file", input: {} }, + }), + data({ + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":' }, + }), + data({ type: "content_block_stop", index: 0 }), + data({ type: "message_delta", delta: { stop_reason: "end_turn" } }), + data({ type: "message_stop" }), + ].join(""); + + await assertRejects( + () => collectParts(streamFromText(malformedToolStream)), + ProviderRequestError, + "tool call arguments were not valid JSON object text", + ); + }); + + it("yields no tool call after a deferred tool input failure", async () => { + const mixedToolStream = [ + data({ type: "message_start", message: { usage: { input_tokens: 1 } } }), + data({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_truncated", name: "create_file", input: {} }, + }), + data({ + type: "content_block_delta", + index: 0, + delta: { + type: "input_json_delta", + partial_json: '{"path":"/inbox/mail-1.json","content":', + }, + }), + data({ type: "content_block_stop", index: 0 }), + data({ + type: "content_block_start", + index: 1, + content_block: { type: "tool_use", id: "toolu_complete", name: "bash", input: {} }, + }), + data({ + type: "content_block_delta", + index: 1, + delta: { type: "input_json_delta", partial_json: '{"command":"pwd"}' }, + }), + data({ type: "content_block_stop", index: 1 }), + data({ type: "message_delta", delta: { stop_reason: "end_turn" } }), + data({ type: "message_stop" }), + ].join(""); + + const parts: unknown[] = []; + await assertRejects( + async () => { + for await (const part of streamAnthropicCompatibleParts(streamFromText(mixedToolStream))) { + parts.push(part); + } + }, + ProviderRequestError, + "tool call arguments were not valid JSON object text", + ); + + // Deferring the failure must not let a later, well-formed tool call reach + // the caller: the turn throws either way, and a dispatched tool call from + // a failed turn would be a side effect the pre-deferral parser never had. + assertEquals( + parts.some((part) => (part as { type?: string }).type === "tool-call"), + false, + ); + // Streaming progress parts (`tool-input-start` / `tool-input-delta`) still + // flow, as they do for any tool block; only the dispatchable `tool-call` + // part is withheld. + assertEquals( + parts.every((part) => + (part as { type?: string }).type === "tool-input-start" || + (part as { type?: string }).type === "tool-input-delta" + ), + true, + ); + }); + + it("classifies a truncation resolved from a buffered trailing message_delta", async () => { + // The `message_delta` handler refuses to run while a content block is open, + // so the only way a deferred failure survives to `validateCompletion()` is + // the client tool-use read timeout: the trailing event is still in the SSE + // buffer and `mergeTrailingBufferUsage()` applies it first. + let cancelCount = 0; + const stream = streamFromChunksWithCancelSpy([ + [ + data({ type: "message_start", message: { usage: { input_tokens: 1 } } }), + data({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_done", name: "bash", input: {} }, + }), + data({ + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"command":"pwd"}' }, + }), + data({ type: "content_block_stop", index: 0 }), + data({ + type: "content_block_start", + index: 1, + content_block: { type: "tool_use", id: "toolu_cut", name: "create_file", input: {} }, + }), + data({ + type: "content_block_delta", + index: 1, + delta: { type: "input_json_delta", partial_json: '{"path":"/inbox/mail-1.json","c' }, + }), + data({ type: "content_block_stop", index: 1 }), + // Deliberately unterminated: only the trailing-buffer flush sees it. + `event: message_delta\r\ndata: ${ + JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "max_tokens" }, + usage: { output_tokens: 4096 }, + }) + }`, + ].join(""), + ], { + closeDelayMs: 600, + onCancel: () => cancelCount++, + }); + + await assertRejects( + () => + collectParts(stream, { + clientToolUseTrailingUsageGraceMs: 5, + allowPostTerminalUsage: true, + }), + ProviderOutputTruncatedError, + "provider output truncated at the max output token limit", + ); + + await waitForCondition(() => cancelCount === 1 && !stream.locked, 500); + assertEquals(stream.locked, false); + }); + it("accepts a complete empty assistant stream", async () => { for ( const terminal of [ diff --git a/extensions/ext-llm-anthropic/src/anthropic-stream.ts b/extensions/ext-llm-anthropic/src/anthropic-stream.ts index f207e33d24..c454975992 100644 --- a/extensions/ext-llm-anthropic/src/anthropic-stream.ts +++ b/extensions/ext-llm-anthropic/src/anthropic-stream.ts @@ -1,6 +1,7 @@ import { mergeUsage, parseSseChunk, + ProviderOutputTruncatedError, ProviderOverloadedError, ProviderRateLimitError, ProviderRequestError, @@ -105,6 +106,24 @@ function invalidAnthropicStream( }); } +const UNPARSABLE_TOOL_INPUT_ISSUE = "tool call arguments were not valid JSON object text"; + +/** + * Build the terminal error for a response the provider cut at the output token + * limit. Fine-grained tool streaming does not guarantee valid `partial_json` + * when `stop_reason` is `max_tokens`, so an unparsable tool input plus that + * stop reason is a truncation, not a malformed stream. + */ +function truncatedAnthropicOutput(providerLabel: string): ProviderOutputTruncatedError { + return new ProviderOutputTruncatedError({ + provider: "anthropic", + status: 200, + message: + `${providerLabel} request failed: provider output truncated at the max output token limit (incomplete tool_use input)`, + retryable: false, + }); +} + function readAnthropicStreamIndex( record: Record, eventType: string, @@ -768,6 +787,19 @@ export async function* streamAnthropicCompatibleParts( let sawDoneMarker = false; let sawStopReason = false; let completedSupportedContentBlocks = 0; + // A tool input that failed to parse is only classifiable once `stop_reason` + // arrives in `message_delta`, which is always later than `content_block_stop`. + let sawUnparsableToolInput = false; + + const resolveDeferredToolInputFailure = (): + | ProviderOutputTruncatedError + | ProviderRequestError + | undefined => { + if (!sawUnparsableToolInput) return undefined; + return rawStopReason === "max_tokens" + ? truncatedAnthropicOutput(providerLabel) + : invalidAnthropicStream(providerLabel, UNPARSABLE_TOOL_INPUT_ISSUE); + }; const mergeRecordUsage = (record: Record) => { usage = mergeUsage(usage, extractAnthropicUsage(record)); @@ -894,6 +926,10 @@ export async function* streamAnthropicCompatibleParts( if (!sawMessageStart) { throw invalidAnthropicStream(providerLabel, "stream contained no provider envelope"); } + const deferredToolInputFailure = resolveDeferredToolInputFailure(); + if (deferredToolInputFailure) { + throw deferredToolInputFailure; + } if ( openContentBlocks.size > 0 || reasoningBlocks.size > 0 || @@ -1434,6 +1470,15 @@ export async function* streamAnthropicCompatibleParts( if (!current) { continue; } + if (sawUnparsableToolInput) { + // The stream is already going to throw once `stop_reason` arrives. + // Before the deferral the first unparsable input threw right here, + // so no later tool call was ever yielded; keep that guarantee so a + // consumer can never dispatch a tool call from a doomed turn. + toolCalls.delete(index); + rawContentBlocks.delete(index); + continue; + } const input = joinAnthropicToolInput(current) || "{}"; let parsedInput: Record; try { @@ -1444,10 +1489,15 @@ export async function* streamAnthropicCompatibleParts( } parsedInput = parsedRecord; } catch { - throw invalidAnthropicStream( - providerLabel, - "tool call arguments were not valid JSON object text", - ); + // Defer: `stop_reason` decides whether this is a truncated response + // or a malformed stream, and it only arrives in `message_delta`. + // Drop the block so no partial tool call reaches the caller or a + // replay checkpoint, where a `tool_use` without a `tool_result` + // would invalidate the next provider request. + sawUnparsableToolInput = true; + toolCalls.delete(index); + rawContentBlocks.delete(index); + continue; } if (rawBlock) rawBlock.input = parsedInput; @@ -1498,6 +1548,17 @@ export async function* streamAnthropicCompatibleParts( if (normalizedFinishReason) { finishReason = normalizedFinishReason; } + // Only decide once a stop reason has actually arrived. A usage-only + // message_delta carries no stop_reason, and resolving here would + // classify the deferred failure as a malformed stream before the + // later delta that says max_tokens -- turning the truncation this + // change exists to identify back into the generic error. + if (sawStopReason) { + const deferredToolInputFailure = resolveDeferredToolInputFailure(); + if (deferredToolInputFailure) { + throw deferredToolInputFailure; + } + } continue; } diff --git a/src/agent/hosted/executor-agent-schema.ts b/src/agent/hosted/executor-agent-schema.ts index 83214d7266..ef5e1d8122 100644 --- a/src/agent/hosted/executor-agent-schema.ts +++ b/src/agent/hosted/executor-agent-schema.ts @@ -34,6 +34,7 @@ const failureStatus = { AI_PROVIDER_WORKSPACE_LIMIT_EXCEEDED: 502, AI_PROVIDER_BILLING_ERROR: 502, GATEWAY_PROJECT_REQUIRED: 400, + PROVIDER_OUTPUT_TRUNCATED: 502, EXTERNAL_SERVICE_ERROR: 502, PERMISSION_DENIED: 403, DURABLE_RUN_EVENT_PERSISTENCE_FAILED: 500, @@ -61,6 +62,7 @@ export const EXECUTOR_AGENT_FAILURE_CODES = Object.freeze( "AI_PROVIDER_WORKSPACE_LIMIT_EXCEEDED", "AI_PROVIDER_BILLING_ERROR", "GATEWAY_PROJECT_REQUIRED", + "PROVIDER_OUTPUT_TRUNCATED", "EXTERNAL_SERVICE_ERROR", "PERMISSION_DENIED", "DURABLE_RUN_EVENT_PERSISTENCE_FAILED", diff --git a/src/agent/runtime/chat-stream-handler.test.ts b/src/agent/runtime/chat-stream-handler.test.ts index ec09e8114f..76a8f06c6f 100644 --- a/src/agent/runtime/chat-stream-handler.test.ts +++ b/src/agent/runtime/chat-stream-handler.test.ts @@ -10,6 +10,12 @@ import { SpanKind, } from "#veryfront/observability/tracing/api-shim.ts"; import { createMockResult, createSSECollector } from "./chat-stream-handler.test-helpers.ts"; +import { + resolveRelayableExecutionFailure, + resolveRuntimeExecutionErrorEvent, +} from "./chat-stream-handler.ts"; +import { createRuntimeProviderStreamFailure } from "#veryfront/runtime/provider-stream-error-provenance.ts"; +import { ProviderOutputTruncatedError } from "#veryfront/provider/runtime-loader/provider-http.ts"; import { announceStreamedToolCallInput, createRuntimeStreamSource, @@ -3663,3 +3669,35 @@ describe("chat-stream-handler provider-executed tool finalization", () => { ); }); }); + +// Codex P1 on veryfront-code#4516: the replay relay writes a PUBLIC RunError, so +// a non-provider failure must not carry its own message across. A persistence +// error can contain a database URL or an internal path, which AGENTS.md forbids +// in user-facing output. +describe("resolveRelayableExecutionFailure", () => { + it("withholds a non-provider failure's message from the relay", () => { + const persistenceFailure = new Error( + "finalize failed: postgres://user:pw@db.internal:5432/veryfront timed out", + ); + + // The SSE fallback path may still show it; the relay may not. + assertEquals( + resolveRuntimeExecutionErrorEvent(persistenceFailure).error, + "finalize failed: postgres://user:pw@db.internal:5432/veryfront timed out", + ); + assertStrictEquals(resolveRelayableExecutionFailure(persistenceFailure), undefined); + }); + + it("relays a curated provider terminal error", () => { + const truncated = new ProviderOutputTruncatedError({ + provider: "anthropic", + status: 200, + message: "anthropic request failed: provider output truncated at the max output token limit", + retryable: false, + }); + const relayed = resolveRelayableExecutionFailure(createRuntimeProviderStreamFailure(truncated)); + + // The whole point of #1467: the real classified cause reaches the run error. + assertEquals(relayed?.code, "PROVIDER_OUTPUT_TRUNCATED"); + }); +}); diff --git a/src/agent/runtime/chat-stream-handler.ts b/src/agent/runtime/chat-stream-handler.ts index 52830cc898..51d78113e0 100644 --- a/src/agent/runtime/chat-stream-handler.ts +++ b/src/agent/runtime/chat-stream-handler.ts @@ -206,6 +206,38 @@ function resolveRuntimeFallbackErrorEvent(error: unknown): RuntimeStreamErrorEve } } +/** + * The subset of an execution failure that may cross into a PUBLIC RunError. + * + * The SSE stream may carry a fallback `Error.message`, but the relay writes into + * a durable, client-visible RunError. A non-provider failure there -- a + * persistence error from `turnPersistence.finalize()`, say -- can carry database + * URLs or internal paths, which AGENTS.md forbids in user-facing output. Only + * curated provider diagnostics and explicitly public lifecycle messages are + * relayed; anything else returns undefined so the relay keeps its neutral + * boundary message. + */ +export function resolveRelayableExecutionFailure( + error: unknown, +): { message: string; code?: string } | undefined { + const providerFailure = readRuntimeProviderStreamFailureCause(error); + if (providerFailure.found) { + const knownProviderError = resolveKnownProviderTerminalError(providerFailure.cause); + if (!knownProviderError) return undefined; + return { + message: knownProviderError.message, + ...(knownProviderError.code ? { code: knownProviderError.code } : {}), + }; + } + + if (isStreamLifecycleFailure(error)) { + const event = resolveRuntimeStreamErrorEvent(error); + return { message: event.error, ...(event.code ? { code: event.code } : {}) }; + } + + return undefined; +} + /** Serialize an outer runtime failure without inferring provider provenance. */ export function resolveRuntimeExecutionErrorEvent(error: unknown): RuntimeStreamErrorEvent { const providerFailure = readRuntimeProviderStreamFailureCause(error); diff --git a/src/agent/runtime/index.ts b/src/agent/runtime/index.ts index b8008a3a87..daec3727e5 100644 --- a/src/agent/runtime/index.ts +++ b/src/agent/runtime/index.ts @@ -58,7 +58,7 @@ import { type ToolResultPart, } from "../types.ts"; import { ensureModelReady, type ModelRuntime, resolveModel } from "#veryfront/provider"; -import { DURABLE_RUN_EVENT_PERSISTENCE_FAILED } from "#veryfront/errors"; +import { DURABLE_RUN_EVENT_PERSISTENCE_FAILED, isVeryfrontError } from "#veryfront/errors"; import { generateId } from "#veryfront/utils/id.ts"; import { detectPlatform, getPlatformCapabilities } from "#veryfront/platform/core-platform.ts"; import { @@ -96,6 +96,7 @@ import { createRuntimeStreamSource, createStreamState, processStream, + resolveRelayableExecutionFailure, resolveRuntimeExecutionErrorEvent, type StreamingToolCall, type StreamingToolResult, @@ -167,6 +168,7 @@ import { getRuntimeToolExposureCheckpointPersister, isRuntimeProviderReplayCheckpointPersistenceRequired, isRuntimeToolExposureCheckpointPersistenceRequired, + type ProviderReplayTurnFailure, resolveRuntimeToolLoading, type RuntimeToolFilterConfig, } from "./runtime-tool-config.ts"; @@ -1188,7 +1190,7 @@ type RuntimeProviderReplayCheckpointEmission = { state: ProviderReplayCheckpointEmissionState | undefined; persist: ((checkpoint: ProviderReplayCheckpoint) => void | Promise) | undefined; complete: (() => void | Promise) | undefined; - fail: (() => void | Promise) | undefined; + fail: ((failure?: ProviderReplayTurnFailure) => void | Promise) | undefined; failed: boolean; required: boolean; }; @@ -1223,10 +1225,11 @@ function resolveRuntimeProviderReplayCheckpointEmission( async function failProviderReplayCheckpointTurn( emission: RuntimeProviderReplayCheckpointEmission, + failure?: ProviderReplayTurnFailure, ): Promise { if (emission.failed) return; emission.failed = true; - await emission.fail?.(); + await emission.fail?.(failure); } async function persistProviderReplayCheckpointAfterTurn(input: { @@ -1236,11 +1239,35 @@ async function persistProviderReplayCheckpointAfterTurn(input: { try { await persistProviderReplayCheckpointAfterTurnUnsafe(input); } catch (error) { - await failProviderReplayCheckpointTurn(input.emission); + await failProviderReplayCheckpointTurn( + input.emission, + resolveProviderReplayPersistenceFailure(error), + ); throw error; } } +/** + * Attribute a checkpoint persistence failure to Veryfront, not to the provider. + * + * Only the curated title and code of our own durable-run-event error cross the + * boundary; anything else falls back to the relay's neutral default. + */ +function resolveProviderReplayPersistenceFailure( + error: unknown, +): ProviderReplayTurnFailure | undefined { + if ( + !isVeryfrontError(error) || + error.slug !== DURABLE_RUN_EVENT_PERSISTENCE_FAILED.slug + ) { + return undefined; + } + return { + message: error.title, + code: "DURABLE_RUN_EVENT_PERSISTENCE_FAILED", + }; +} + async function persistProviderReplayCheckpointAfterTurnUnsafe(input: { emission: RuntimeProviderReplayCheckpointEmission; providerMetadata: Record | undefined; @@ -2426,7 +2453,15 @@ export class AgentRuntime { await turnPersistence.commit(); return response; }).catch(async (error) => { - await failProviderReplayCheckpointTurn(providerReplayCheckpointEmission); + // A cancellation keeps the relay's neutral default: only a real + // failure hands the relay the sanitized provider cause. + // Same rule as the stream path: the relay writes a public RunError, so + // only curated diagnostics cross it. A persistence failure keeps the + // neutral boundary message rather than exposing its own text. + const relayFailure = isAbortError(error, abortSignal) + ? undefined + : resolveRelayableExecutionFailure(error); + await failProviderReplayCheckpointTurn(providerReplayCheckpointEmission, relayFailure); throw error; }); } finally { @@ -2676,21 +2711,35 @@ export class AgentRuntime { } catch (finalizationError) { error = finalizationError; } + // Resolve the sanitized event first so the replay relay fails with + // the same cause the stream reports, instead of a manufactured one. + // A cancellation is not a provider failure: it keeps the relay's + // neutral default rather than surfacing the raw abort reason. + const aborted = isAbortError(error, streamAbortSignal); + const errorEvent = aborted ? undefined : resolveRuntimeExecutionErrorEvent(error); + // The relay writes a PUBLIC RunError, so it takes only curated + // diagnostics -- a persistence failure's raw message can carry + // internal detail the SSE fallback path is allowed to show but a + // durable client-visible error is not. + const relayFailure = aborted ? undefined : resolveRelayableExecutionFailure(error); try { - await failProviderReplayCheckpointTurn(providerReplayCheckpointEmission); + await failProviderReplayCheckpointTurn( + providerReplayCheckpointEmission, + relayFailure, + ); } catch (failureHookError) { logger.debug("Provider replay failure hook rejected", { error: failureHookError, }); } - if (isAbortError(error, streamAbortSignal)) { + if (!errorEvent) { closeSSEStream(controller); return; } this.status = "error"; logger.error("Agent stream error", { error }); - sendSSE(controller, encoder, resolveRuntimeExecutionErrorEvent(error)); + sendSSE(controller, encoder, errorEvent); closeSSEStream(controller); } finally { try { diff --git a/src/agent/runtime/provider-replay-emission.test.ts b/src/agent/runtime/provider-replay-emission.test.ts index 3234d6c9e6..ac12e63480 100644 --- a/src/agent/runtime/provider-replay-emission.test.ts +++ b/src/agent/runtime/provider-replay-emission.test.ts @@ -6,6 +6,7 @@ import { assertThrows, } from "#veryfront/testing/assert.ts"; import { describe, it } from "#veryfront/testing/bdd.ts"; +import { waitFor } from "#veryfront/testing/deno-compat.ts"; import { defineSchema } from "#veryfront/schemas"; import { tool } from "#veryfront/tool"; import { agent, type AgentConfig } from "#veryfront/agent"; @@ -17,7 +18,8 @@ import { createProviderReplayCheckpointEmissionState, type ProviderReplayCheckpoint, } from "./provider-replay.ts"; -import type { RuntimeToolFilterConfig } from "./runtime-tool-config.ts"; +import type { ProviderReplayTurnFailure, RuntimeToolFilterConfig } from "./runtime-tool-config.ts"; +import { ProviderOutputTruncatedError } from "veryfront/provider/shared"; const MESSAGE_ID = "assistant-message-1"; const SIGNATURE = "test-signature"; @@ -436,4 +438,125 @@ describe("provider replay checkpoint emission", () => { assertInstanceOf(error, VeryfrontError); assertEquals(error.slug, "durable-run-event-persistence-failed"); }); + /** + * Regression coverage for #1467: this is the junction between the typed + * truncation the Anthropic parser now raises and the run error an internal + * agent reports. The runtime is the only place that carries the sanitized + * `{message, code}` pair into the replay failure hook, so the hook argument + * is asserted against a real provider failure rather than a hand-built one. + */ + it("hands the classified provider truncation to the replay failure hook", async () => { + const failures: (ProviderReplayTurnFailure | undefined)[] = []; + const privateProviderDetail = "incomplete tool_use input "; + const model = scriptedModel([() => { + throw new ProviderOutputTruncatedError({ + provider: "anthropic", + status: 200, + message: + `Anthropic request failed: provider output truncated at the max output token limit (${privateProviderDetail})`, + retryable: false, + }); + }], { + modelId: "anthropic/truncated-provider-replay-stream", + provider: "anthropic", + only: "stream", + }); + const config = { + id: "truncated-provider-replay-stream", + model: "anthropic/truncated-provider-replay-stream", + system: "Answer.", + skills: false, + maxSteps: 1, + resolveModelTransport: () => ({ model }), + __vfProviderReplayCheckpointMessageId: MESSAGE_ID, + __vfProviderReplayCheckpointTurnFailed: (failure?: ProviderReplayTurnFailure) => { + failures.push(failure); + }, + } as AgentConfig & RuntimeToolFilterConfig; + + const stream = await agent(config).stream({ input: "Answer" }); + const body = await stream.toDataStreamResponse().text(); + + assertEquals(failures.length, 1); + assertEquals(failures[0]?.code, "PROVIDER_OUTPUT_TRUNCATED"); + assertEquals( + failures[0]?.message, + "The model stopped at its output token limit before it finished the response. " + + "Raise the model output token limit, or ask for a shorter response.", + ); + assertEquals(failures[0]?.message.includes(privateProviderDetail), false); + assertEquals(body.includes("PROVIDER_OUTPUT_TRUNCATED"), true); + assertEquals(body.includes(privateProviderDetail), false); + }); + + it("keeps a cancelled turn's replay failure free of the cancellation reason", async () => { + const failures: (ProviderReplayTurnFailure | undefined)[] = []; + const cancelReasonMarker = "client disconnected "; + const model = scriptedModel([{ hangUntilAbort: true }], { + modelId: "anthropic/cancelled-provider-replay-stream", + provider: "anthropic", + only: "stream", + }); + const config = { + id: "cancelled-provider-replay-stream", + model: "anthropic/cancelled-provider-replay-stream", + system: "Answer.", + skills: false, + maxSteps: 1, + resolveModelTransport: () => ({ model }), + __vfProviderReplayCheckpointMessageId: MESSAGE_ID, + __vfProviderReplayCheckpointTurnFailed: (failure?: ProviderReplayTurnFailure) => { + failures.push(failure); + }, + } as AgentConfig & RuntimeToolFilterConfig; + + const abortController = new AbortController(); + const stream = await agent(config).stream({ + input: "Answer", + abortSignal: abortController.signal, + }); + const bodyPromise = stream.toDataStreamResponse().text(); + await waitFor(() => model.callCount > 0, { + message: "the model call must start before the run is cancelled", + }); + abortController.abort(new DOMException(cancelReasonMarker, "AbortError")); + const body = await bodyPromise; + + assertEquals(failures.length, 1); + // A cancellation reports no cause at all, so the relay keeps its neutral + // default instead of surfacing the client's raw abort reason. + assertEquals(failures[0], undefined); + assertEquals(body.includes(cancelReasonMarker), false); + }); + + it("attributes a checkpoint persistence failure to Veryfront, not the provider", async () => { + const failures: (ProviderReplayTurnFailure | undefined)[] = []; + const model = scriptedModel([{ text: "done" }], { + modelId: "anthropic/required-provider-replay-persistence", + provider: "anthropic", + only: "generate", + }); + const config = { + id: "required-provider-replay-persistence", + model: "anthropic/required-provider-replay-persistence", + system: "Answer.", + skills: false, + maxSteps: 1, + resolveModelTransport: () => ({ model }), + __vfProviderReplayCheckpointPersistenceRequired: true, + __vfProviderReplayCheckpointTurnFailed: (failure?: ProviderReplayTurnFailure) => { + failures.push(failure); + }, + } as AgentConfig & RuntimeToolFilterConfig; + + await assertRejects( + () => agent(config).generate({ input: "Answer" }), + VeryfrontError, + "provider replay checkpoint message identity is required", + ); + + assertEquals(failures.length, 1); + assertEquals(failures[0]?.code, "DURABLE_RUN_EVENT_PERSISTENCE_FAILED"); + assertEquals(failures[0]?.message, "Durable run event persistence failed"); + }); }); diff --git a/src/agent/runtime/runtime-tool-config.ts b/src/agent/runtime/runtime-tool-config.ts index 748427d169..01635a3ee4 100644 --- a/src/agent/runtime/runtime-tool-config.ts +++ b/src/agent/runtime/runtime-tool-config.ts @@ -28,6 +28,17 @@ export type RuntimeToolLoadingMode = "eager" | "deferred"; export const SOURCE_INTEGRATION_POLICY_CONTEXT_KEY = "__vfSourceIntegrationPolicy"; +/** + * Sanitized cause handed to the replay turn failure hook. + * + * Only the already-sanitized `{message, code}` pair that the runtime sends on + * the stream crosses this boundary. The raw provider error never does. + */ +export type ProviderReplayTurnFailure = { + message: string; + code?: string; +}; + export type RuntimeToolFilterConfig = AgentConfig & { __vfForwardedIntegrationToolDefs?: Array< { name: string; description: string; parameters: Record } @@ -39,7 +50,9 @@ export type RuntimeToolFilterConfig = AgentConfig & { checkpoint: ProviderReplayCheckpoint, ) => void | Promise; __vfProviderReplayCheckpointTurnComplete?: () => void | Promise; - __vfProviderReplayCheckpointTurnFailed?: () => void | Promise; + __vfProviderReplayCheckpointTurnFailed?: ( + failure?: ProviderReplayTurnFailure, + ) => void | Promise; __vfProviderReplayCheckpointPersistenceRequired?: boolean; __vfPersistToolExposureCheckpoint?: ( checkpoint: ToolExposureCheckpoint, @@ -166,7 +179,7 @@ export function getRuntimeProviderReplayCheckpointTurnComplete( /** Return the trusted hook that aborts one provider response boundary. */ export function getRuntimeProviderReplayCheckpointTurnFailed( config: AgentConfig, -): (() => void | Promise) | undefined { +): ((failure?: ProviderReplayTurnFailure) => void | Promise) | undefined { const value = (config as RuntimeToolFilterConfig).__vfProviderReplayCheckpointTurnFailed; return typeof value === "function" ? value : undefined; } diff --git a/src/chat/provider-error-registry.ts b/src/chat/provider-error-registry.ts index 4fb04fbf26..ab2faeacee 100644 --- a/src/chat/provider-error-registry.ts +++ b/src/chat/provider-error-registry.ts @@ -41,6 +41,13 @@ export const AI_PROVIDER_BILLING_ERROR = { status: 502, } as const; +export const PROVIDER_OUTPUT_TRUNCATED_ERROR = { + code: "PROVIDER_OUTPUT_TRUNCATED", + message: + "The model stopped at its output token limit before it finished the response. Raise the model output token limit, or ask for a shorter response.", + status: 502, +} as const; + export const GATEWAY_PROJECT_REQUIRED_ERROR = { code: "GATEWAY_PROJECT_REQUIRED", message: "A project is required to use Veryfront-managed AI inference", @@ -61,6 +68,7 @@ export const CURATED_PROVIDER_FAILURE_CODES = [ "AI_PROVIDER_WORKSPACE_LIMIT_EXCEEDED", "AI_PROVIDER_BILLING_ERROR", "GATEWAY_PROJECT_REQUIRED", + "PROVIDER_OUTPUT_TRUNCATED", ] as const; export type CuratedProviderFailureCode = typeof CURATED_PROVIDER_FAILURE_CODES[number]; @@ -100,6 +108,7 @@ const failures = { AI_PROVIDER_WORKSPACE_LIMIT_EXCEEDED: AI_PROVIDER_WORKSPACE_LIMIT_ERROR, AI_PROVIDER_BILLING_ERROR: AI_PROVIDER_BILLING_ERROR, GATEWAY_PROJECT_REQUIRED: GATEWAY_PROJECT_REQUIRED_ERROR, + PROVIDER_OUTPUT_TRUNCATED: PROVIDER_OUTPUT_TRUNCATED_ERROR, } as const; /** Return fixed local diagnostics; provider payload/status values are never forwarded. */ diff --git a/src/chat/provider-errors.ts b/src/chat/provider-errors.ts index d9447541f3..f1fea35f43 100644 --- a/src/chat/provider-errors.ts +++ b/src/chat/provider-errors.ts @@ -1,5 +1,6 @@ import { safeJsonParse } from "#veryfront/utils/json.ts"; import { + ProviderOutputTruncatedError, ProviderOverloadedError, ProviderQuotaError, } from "#veryfront/provider/runtime-loader/provider-http.ts"; @@ -11,6 +12,7 @@ import { MODEL_UNSUPPORTED_ASSISTANT_PREFILL_ERROR, OUTPUT_SCHEMA_NOT_CLOSED_ERROR, PROJECT_SCHEMA_ERROR, + PROVIDER_OUTPUT_TRUNCATED_ERROR, registeredProviderFailure, } from "./provider-error-registry.ts"; export { safeJsonParse }; @@ -464,6 +466,9 @@ function parseProviderErrorInner( if (error instanceof ProviderQuotaError) { return AI_PROVIDER_BILLING_ERROR; } + if (error instanceof ProviderOutputTruncatedError) { + return PROVIDER_OUTPUT_TRUNCATED_ERROR; + } if (error instanceof ProviderOverloadedError) { return { code: "OVERLOADED_ERROR", diff --git a/src/internal-agents/run-stream.test.ts b/src/internal-agents/run-stream.test.ts index d5cd0e2c02..b42deca90b 100644 --- a/src/internal-agents/run-stream.test.ts +++ b/src/internal-agents/run-stream.test.ts @@ -28,7 +28,10 @@ import type { } from "#veryfront/sandbox"; import { registerSkill } from "#veryfront/skill/registry.ts"; import { type ModelRuntime, registerModelProvider } from "#veryfront/provider"; +import { ProviderOutputTruncatedError } from "veryfront/provider/shared"; +import { tool } from "#veryfront/tool"; import type { RemoteToolSource, Tool } from "#veryfront/tool"; +import { defineSchema } from "#veryfront/schemas"; import { __resetLoggerConfigForTests, type LogEntry } from "#veryfront/utils/logger/logger.ts"; import type { AgentRunEventSink } from "#veryfront/runtime/model-call-context.ts"; import { getActiveRunEventSinks } from "#veryfront/runtime/run-event-sink-context.ts"; @@ -978,9 +981,164 @@ describe("internal-agents/run-stream", () => { }); } + it("surfaces a truncated provider response as a classified run error", async () => { + const sessionManager = new AgentRunSessionManager(); + const truncationError = new ProviderOutputTruncatedError({ + provider: "anthropic", + status: 200, + message: + "Anthropic request failed: provider output truncated at the max output token limit (incomplete tool_use input)", + retryable: false, + }); + const unregister = registerModelProvider("issue-1467", () => ({ + provider: "issue-1467", + modelId: "issue-1467/truncating", + doGenerate: () => Promise.reject(new Error("generate must not be called")), + doStream: () => + Promise.resolve({ + stream: new ReadableStream({ + start(controller) { + controller.error(truncationError); + }, + }), + }), + })); + + try { + const runtimeAgent = createAgent({ + id: "issue-1467-truncation", + model: "issue-1467/truncating", + system: "Reply to the user.", + skills: false, + }); + const response = await createRuntimeAgentStreamResponse( + { + threadId: crypto.randomUUID(), + runId: "run_issue_1467", + messages: [{ id: "message-1", role: "user", content: "Hello" }], + tools: [], + context: [], + }, + runtimeAgent, + { sessionManager }, + ); + const frames = parseSseFrames(await response.text()); + const runError = frames.find((frame) => frame.event === "RunError")?.data as + | Record + | undefined; + + assertEquals(runError?.code, "PROVIDER_OUTPUT_TRUNCATED"); + assertEquals( + runError?.message, + "The model stopped at its output token limit before it finished the response. Raise the model output token limit, or ask for a shorter response.", + ); + assertEquals(frames.some((frame) => frame.event === "RunFinished"), false); + } finally { + unregister(); + } + }); + + /** + * Regression coverage for #1467, end to end on the real call path: a real + * `ProviderOutputTruncatedError` raised inside a replay-checkpoint run, with + * the relay parked at a tool boundary that never reached its checkpoint. + * This is the staging shape, and it exercises the runtime wiring (the only + * production code that carries the sanitized cause into the relay) instead + * of invoking the failure hook by hand. + */ + it("surfaces a truncated provider response parked at a replay boundary", async () => { + const sessionManager = new AgentRunSessionManager(); + const privateProviderDetail = "incomplete tool_use input "; + const truncationError = new ProviderOutputTruncatedError({ + provider: "anthropic", + status: 200, + message: + `Anthropic request failed: provider output truncated at the max output token limit (${privateProviderDetail})`, + retryable: false, + }); + const unregister = registerModelProvider("issue-1467-parked", () => ({ + provider: "issue-1467-parked", + modelId: "issue-1467-parked/truncating", + doGenerate: () => Promise.reject(new Error("generate must not be called")), + doStream: () => { + // The turn produced one complete tool call before the provider cut the + // response at its output token limit, so the consumer reaches that tool + // boundary with no checkpoint frame behind it. The parts are delivered + // from `pull` because erroring a stream discards whatever is queued. + const parts: unknown[] = [ + { type: "tool-input-start", id: "tool-1", toolName: "lookup" }, + { type: "tool-input-available", toolCallId: "tool-1", toolName: "lookup", input: {} }, + ]; + let delivered = 0; + return Promise.resolve({ + stream: new ReadableStream({ + pull(controller) { + if (delivered < parts.length) { + controller.enqueue(parts[delivered++]); + return; + } + controller.error(truncationError); + }, + }), + }); + }, + })); + + try { + const runtimeAgent = createAgent({ + id: "issue-1467-parked-truncation", + model: "issue-1467-parked/truncating", + system: "Reply to the user.", + skills: false, + tools: { + lookup: tool({ + id: "lookup", + description: "Look up a value", + inputSchema: defineSchema((v) => v.object({}))(), + execute: () => ({ value: "found" }), + }), + }, + }); + const response = await createRuntimeAgentStreamResponse( + { + threadId: crypto.randomUUID(), + runId: "run_issue_1467_parked", + messageId: crypto.randomUUID(), + messages: [{ id: "message-1", role: "user", content: "Hello" }], + tools: [], + context: [], + }, + runtimeAgent, + { + sessionManager, + providerReplayCheckpointEmissionEnabled: true, + persistProviderReplayCheckpoint: () => Promise.resolve(), + }, + ); + const body = await response.text(); + const frames = parseSseFrames(body); + const runError = frames.find((frame) => frame.event === "RunError")?.data as + | Record + | undefined; + + assertEquals(runError?.code, "PROVIDER_OUTPUT_TRUNCATED"); + assertEquals( + runError?.message, + "The model stopped at its output token limit before it finished the response. Raise the model output token limit, or ask for a shorter response.", + ); + assertEquals(body.includes("Provider replay turn failed before its boundary"), false); + assertEquals(body.includes(privateProviderDetail), false); + assertEquals(frames.some((frame) => frame.event === "RunFinished"), false); + } finally { + unregister(); + } + }); + it("releases a pending tool boundary when the runtime turn fails", async () => { const sessionManager = new AgentRunSessionManager(); - let failProviderReplayTurn: (() => void | Promise) | undefined; + let failProviderReplayTurn: + | ((failure?: { message: string; code?: string }) => void | Promise) + | undefined; const agent = { id: "test", config: { @@ -1006,7 +1164,9 @@ describe("internal-agents/run-stream", () => { persistProviderReplayCheckpoint: () => Promise.resolve(), createRuntime: (runtimeAgent) => { failProviderReplayTurn = (runtimeAgent.config as Agent["config"] & { - __vfProviderReplayCheckpointTurnFailed?: () => void | Promise; + __vfProviderReplayCheckpointTurnFailed?: ( + failure?: { message: string; code?: string }, + ) => void | Promise; }).__vfProviderReplayCheckpointTurnFailed; return { stream: async () => @@ -1018,7 +1178,10 @@ describe("internal-agents/run-stream", () => { ), ); setTimeout(async () => { - await failProviderReplayTurn?.(); + // The runtime always reports the cause it is about to send + // on the stream; an unclassified provider failure is the + // sanitized "Provider stream failed". + await failProviderReplayTurn?.({ message: "Provider stream failed" }); controller.enqueue( new TextEncoder().encode( 'data: {"type":"error","error":"provider stream failed"}\n\n', @@ -1036,7 +1199,85 @@ describe("internal-agents/run-stream", () => { const body = await response.text(); assertStringIncludes(body, "event: RunError"); + assertStringIncludes(body, "Provider stream failed"); assertEquals(body.includes("event: RunFinished"), false); + assertEquals(body.includes("Provider replay turn failed before its boundary"), false); + }); + + it("surfaces the provider failure cause at a pending replay boundary", async () => { + const sessionManager = new AgentRunSessionManager(); + let failProviderReplayTurn: + | ((failure?: { message: string; code?: string }) => void | Promise) + | undefined; + const agent = { + id: "test", + config: { + id: "test", + model: "anthropic/claude-opus-4-8", + system: "test", + }, + } as unknown as Agent; + + const response = await createRuntimeAgentStreamResponse( + { + threadId: crypto.randomUUID(), + runId: "run_1", + messageId: crypto.randomUUID(), + messages: [], + tools: [], + context: [], + }, + agent, + { + sessionManager, + providerReplayCheckpointEmissionEnabled: true, + persistProviderReplayCheckpoint: () => Promise.resolve(), + createRuntime: (runtimeAgent) => { + failProviderReplayTurn = (runtimeAgent.config as Agent["config"] & { + __vfProviderReplayCheckpointTurnFailed?: ( + failure?: { message: string; code?: string }, + ) => void | Promise; + }).__vfProviderReplayCheckpointTurnFailed; + return { + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"step-start"}\n\ndata: {"type":"tool-input-start","toolCallId":"tool-1","toolName":"lookup"}\n\ndata: {"type":"tool-input-available","toolCallId":"tool-1","toolName":"lookup","input":{}}\n\n', + ), + ); + setTimeout(async () => { + await failProviderReplayTurn?.({ + message: + "Anthropic request failed: provider output truncated at the max output token limit (incomplete tool_use input)", + code: "PROVIDER_OUTPUT_TRUNCATED", + }); + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"error","error":"provider stream failed"}\n\n', + ), + ); + controller.close(); + }, 0); + }, + }), + }; + }, + }, + ); + + const frames = parseSseFrames(await response.text()); + const runError = frames.find((frame) => frame.event === "RunError")?.data as + | Record + | undefined; + + assertEquals( + runError?.message, + "Anthropic request failed: provider output truncated at the max output token limit (incomplete tool_use input)", + ); + assertEquals(runError?.code, "PROVIDER_OUTPUT_TRUNCATED"); + assertEquals(frames.some((frame) => frame.event === "RunFinished"), false); }); it("aborts a pending replay boundary when the run is cancelled", async () => { diff --git a/src/internal-agents/run-stream.ts b/src/internal-agents/run-stream.ts index 624fa97536..03fa4d94b0 100644 --- a/src/internal-agents/run-stream.ts +++ b/src/internal-agents/run-stream.ts @@ -74,6 +74,7 @@ import type { RuntimeRunAgentInput } from "./schema.ts"; import { serverLogger } from "#veryfront/utils"; import { compareStrings } from "#veryfront/utils/compare.ts"; import { type ProviderReplayCheckpoint } from "#veryfront/agent/runtime/provider-replay.ts"; +import { type ProviderReplayTurnFailure } from "#veryfront/agent/runtime/runtime-tool-config.ts"; import { DURABLE_RUN_EVENT_PERSISTENCE_FAILED } from "#veryfront/errors"; import type { ProviderReplayCheckpointPersister } from "./provider-replay-checkpoint-persister.ts"; import { createVeryfrontCloudInferenceModelResolver } from "#veryfront/agent/hosted/inference-credential.ts"; @@ -813,9 +814,18 @@ type ProviderReplayPrivateFrame = { payload: Record; }; +/** Error carrying the run error code the provider failure was classified with. */ +type ProviderReplayTurnError = Error & { vfRunErrorCode?: string }; + +/** Read the classified run error code a failed replay boundary carried. */ +function readProviderReplayTurnErrorCode(error: unknown): string | undefined { + const code = (error as ProviderReplayTurnError | null)?.vfRunErrorCode; + return typeof code === "string" && code.length > 0 ? code : undefined; +} + function createProviderReplayCheckpointRelay(): { complete: (messageId: string) => Promise; - fail: () => Promise; + fail: (failure?: ProviderReplayTurnFailure) => Promise; takeCompletedTurn: () => Promise; hasCompletedTurn: () => boolean; } { @@ -824,7 +834,7 @@ function createProviderReplayCheckpointRelay(): { | ((frames: ProviderReplayPrivateFrame[]) => void) | undefined; let rejectPending: ((error: Error) => void) | undefined; - let terminalError: Error | undefined; + let terminalError: ProviderReplayTurnError | undefined; const takeReadyTurn = (): ProviderReplayPrivateFrame[] | undefined => { const boundaryIndex = buffered.findIndex((frame) => @@ -853,9 +863,17 @@ function createProviderReplayCheckpointRelay(): { }); resolveIfReady(); }, - fail: async () => { + fail: async (failure) => { if (terminalError) return; - terminalError = new Error("Provider replay turn failed before its boundary"); + // Carry the runtime's already-sanitized cause instead of manufacturing a + // message: the parked boundary waiter is what the consumer surfaces, so a + // fixed string here masks the real provider failure. The fallback stays + // neutral — a caller that reports no cause (a cancellation, a Veryfront + // persistence failure) must not be attributed to the provider. + terminalError = new Error( + failure?.message ?? "Provider replay turn failed before its boundary", + ); + if (failure?.code) terminalError.vfRunErrorCode = failure.code; buffered.splice(0); const reject = rejectPending; resolvePending = undefined; @@ -1501,7 +1519,7 @@ export async function createRuntimeAgentStreamResponse( error: errorMessage, }); enqueueIfAttached("RunError", { - code: "RUNTIME_ERROR", + code: readProviderReplayTurnErrorCode(error) ?? "RUNTIME_ERROR", message: errorMessage, }); } diff --git a/src/provider/runtime-loader.ts b/src/provider/runtime-loader.ts index c4e493019f..c95e634ff1 100644 --- a/src/provider/runtime-loader.ts +++ b/src/provider/runtime-loader.ts @@ -31,6 +31,7 @@ export { export type { JsonSnapshotOptions, JsonSnapshotValue } from "./runtime-loader/json-snapshot.ts"; export { ProviderError, + ProviderOutputTruncatedError, ProviderOverloadedError, ProviderQuotaError, ProviderRateLimitError, diff --git a/src/provider/runtime-loader/provider-http.ts b/src/provider/runtime-loader/provider-http.ts index 28d3ed7245..dfdcce171a 100644 --- a/src/provider/runtime-loader/provider-http.ts +++ b/src/provider/runtime-loader/provider-http.ts @@ -136,6 +136,15 @@ export class ProviderQuotaError extends ProviderError {} /** Non-retryable 4xx/5xx that doesn't fit another bucket. */ export class ProviderRequestError extends ProviderError {} +/** + * Provider stopped generating at the output token limit, leaving the response + * incomplete (for example a `tool_use` block whose input JSON never closed). + * + * Non-retryable: the same request and the same output token budget truncate + * again. Raise the budget or shorten the requested output instead. + */ +export class ProviderOutputTruncatedError extends ProviderError {} + function readRequestRoute(url: string): string | undefined { try { const parsed = new URL(url); diff --git a/src/provider/shared/index.ts b/src/provider/shared/index.ts index 0f360c0801..8fa8a2efb4 100644 --- a/src/provider/shared/index.ts +++ b/src/provider/shared/index.ts @@ -50,6 +50,7 @@ export { mergeUsage, parseRetryAfterMs, ProviderError, + ProviderOutputTruncatedError, ProviderOverloadedError, ProviderQuotaError, ProviderRateLimitError,