diff --git a/CHANGELOG.md b/CHANGELOG.md index ec634535ef..11b5b59963 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ ### Added +- `VEYYON_DEBUG_STARTUP=1` writes one line per phase of a prompt submission (compaction check, plan arm, context build, memory context), so a slow submit names the phase that spent the time. - `read` takes `depth` and `limit` arguments for directory listings, and a read of the session working directory root with neither now returns a concise top-level listing with per-subdirectory entry counts instead of the recursive tree. - A tool result that carries an image now states whether the picture reached the screen, so a model reading a file describes what it shows instead of reporting that it displayed it. - A picture the block gives up on after the fact, because the session's image budget demoted it or a Kitty session could not convert it, is stated to the model as undrawn instead of being reported as displayed. @@ -39,6 +40,8 @@ - Daemon completion parsing and eval-store serialization errors use shared type guards; behavior is unchanged. - Superseded and useless tool results are now pruned as a batch whose combined size pays for the prompt-cache rewrite it forces, instead of only when a single result sits within 8,000 tokens of the end of the conversation. - The Anthropic provider reads its endpoint, credential placement, rejected betas and retry policy from the catalog's wire-capability table instead of comparing provider ids at seventeen call sites. +- A streaming request no longer pins a parsed clone of its wire payload for the life of the stream: every provider's diagnostic dump retains only the exact sent bytes and materializes a body when a 400/413 dump is built. +- The OpenAI-family, pi-native and Codex request builders serialize the request body once instead of deep-cloning the request graph, which took attempt preparation on a 32MiB context from 82ms to 9ms. - A message that names a dead socket reads the same everywhere: `namesDeadSocket` in `@veyyon/ai/error/flags` is the one list of errnos and phrases, and `ENETUNREACH`, `EHOSTUNREACH` and `EAI_AGAIN` now count as transient transport failures like the rest of them. - `MNEMOPI_NO_EMBEDDINGS=0`, `false`, `no` or `off` now leaves embeddings on everywhere instead of disabling them on the API path. - Every `MNEMOPI_*` value is read by `config.ts` alone; the local-model, extraction and embedding modules ask it instead of parsing the variable again. diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 3893b16f70..05b7bea09c 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -10,6 +10,8 @@ ### Changed - The Anthropic provider reads its endpoint, credential placement, rejected betas and retry policy from the catalog's wire-capability table instead of comparing provider ids at seventeen call sites. +- A streaming request no longer pins a parsed clone of its wire payload for the life of the stream: every provider's diagnostic dump retains only the exact sent bytes and materializes a body when a 400/413 dump is built. +- The OpenAI-family, pi-native and Codex request builders serialize the request body once instead of deep-cloning the request graph, which took attempt preparation on a 32MiB context from 82ms to 9ms. - A message that names a dead socket reads the same everywhere: `namesDeadSocket` in `@veyyon/ai/error/flags` is the one list of errnos and phrases, and `ENETUNREACH`, `EHOSTUNREACH` and `EAI_AGAIN` now count as transient transport failures like the rest of them. ### Fixed diff --git a/packages/ai/src/providers/amazon-bedrock.ts b/packages/ai/src/providers/amazon-bedrock.ts index 00b4e3e7b4..ad3f23de91 100644 --- a/packages/ai/src/providers/amazon-bedrock.ts +++ b/packages/ai/src/providers/amazon-bedrock.ts @@ -42,7 +42,7 @@ import { kStreamingPartialJson, } from "../utils/block-symbols"; import { AssistantMessageEventStream } from "../utils/event-stream"; -import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; import { armPreResponseTimeout, getStreamFirstEventTimeoutMs } from "../utils/idle-iterator"; import { fetchProviderWithRetry } from "../utils/provider-fetch"; import { notifyProviderResponse } from "../utils/provider-response"; @@ -307,6 +307,8 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = ( const blocks = output.content as Block[]; let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let wireBodyJson: string | undefined; const region = resolveBedrockRegion(model.id, options); try { @@ -404,9 +406,11 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = ( model: model.id, method: "POST", url, - body: commandInput, }; - const body = new TextEncoder().encode(JSON.stringify(commandInput)); + // Retain the exact sent BYTES, not the parsed object: a dump body is + // read only on the 400/413 path. + wireBodyJson = JSON.stringify(commandInput); + const body = new TextEncoder().encode(wireBodyJson); if (bearerToken) { return { @@ -590,7 +594,11 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = ( diagnostics = `\n[thinking-diag] ${JSON.stringify(thinkingBlocks)}`; } } - const result = await AIError.finalize(error, { api: model.api, signal: options.signal, rawRequestDump }); + const result = await AIError.finalize(error, { + api: model.api, + signal: options.signal, + rawRequestDump: materializeDumpBody(rawRequestDump, wireBodyJson), + }); output.stopReason = result.stopReason; output.errorStatus = result.status; output.errorId = result.id; diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 760d9341d3..274c73cd93 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -75,7 +75,7 @@ import { withEmptyCompletionRetry } from "../utils/empty-completion-retry"; import { AssistantMessageEventStream } from "../utils/event-stream"; import { isPreResponseStall, openStallLadderBudget } from "../utils/first-event-budget"; import { isFoundryEnabled } from "../utils/foundry"; -import { finalizeErrorMessage, type RawHttpRequestDump } from "../utils/http-inspector"; +import { finalizeErrorMessage, materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs, iterateWithIdleTimeout } from "../utils/idle-iterator"; import { notifyProviderResponse } from "../utils/provider-response"; import { COMBINATOR_KEYS, NO_STRICT, toolWireSchema } from "../utils/schema"; @@ -1879,6 +1879,8 @@ const streamAnthropicOnce = ( timestamp: Date.now(), }; let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let anthropicWireBodyJson: string | undefined; let activeAbortTracker = createAbortSourceTracker(options?.signal); const onSseEvent = options?.onSseEvent; @@ -2035,14 +2037,17 @@ const streamAnthropicOnce = ( nextParams = replacementPayload as typeof nextParams; } nextParams = toWellFormedDeep(nextParams) as typeof nextParams; + // Retain the exact sent BYTES, not the parsed object: a dump body is + // read only on the 400/413 path, and holding the graph here pinned a + // full context-sized object for the whole stream. rawRequestDump = { provider: model.provider, api: output.api, model: model.id, method: "POST", url: `${baseUrl}/v1/messages${isOAuthToken ? "?beta=true" : ""}`, - body: nextParams, }; + anthropicWireBodyJson = JSON.stringify(nextParams); return nextParams; }; let params = await prepareParams(); @@ -2698,7 +2703,10 @@ const streamAnthropicOnce = ( // success (consumers treat its presence as failure). logger.warn("anthropic: strict tools rejected, retrying without strict tools", { model: model.id, - error: await finalizeErrorMessage(streamFailure, rawRequestDump), + error: await finalizeErrorMessage( + streamFailure, + materializeDumpBody(rawRequestDump, anthropicWireBodyJson), + ), }); if (providerSessionState) { providerSessionState.strictToolsDisabled = true; @@ -2843,10 +2851,11 @@ const streamAnthropicOnce = ( api: model.api, provider: model.provider, abortTracker: activeAbortTracker, - rawRequestDump, + rawRequestDump: materializeDumpBody(rawRequestDump, anthropicWireBodyJson), }); output.stopReason = result.stopReason; output.errorStatus = result.status; + output.errorId = result.id; output.errorMessage = maybeAddReplayUnsignedThinkingHint(model, result.message); output.duration = performance.now() - startTime; diff --git a/packages/ai/src/providers/azure-openai-responses.ts b/packages/ai/src/providers/azure-openai-responses.ts index 54bfa6c7fd..4c1c9c1fb3 100644 --- a/packages/ai/src/providers/azure-openai-responses.ts +++ b/packages/ai/src/providers/azure-openai-responses.ts @@ -16,6 +16,7 @@ import { createAbortSourceTracker } from "../utils/abort"; import { withEmptyCompletionRetry } from "../utils/empty-completion-retry"; import { AssistantMessageEventStream } from "../utils/event-stream"; import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody } from "../utils/http-inspector"; import { getOpenAIStreamFirstEventTimeoutMs, getOpenAIStreamIdleTimeoutMs, @@ -100,6 +101,8 @@ const streamAzureOpenAIResponsesOnce = ( model.id, ); let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let wireBodyJson: string | undefined; const abortTracker = createAbortSourceTracker(options?.signal); const firstEventTimeoutAbortError = new AIError.StreamTimeoutError( AZURE_OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE, @@ -129,7 +132,6 @@ const streamAzureOpenAIResponsesOnce = ( model: model.id, method: "POST", url, - body: params, }; let activeRequestParams = params; let activeReasoningEffortFallbackKey = createOpenAIReasoningEffortFallbackKey( @@ -138,17 +140,29 @@ const streamAzureOpenAIResponsesOnce = ( typeof params.model === "string" ? params.model : model.id, ); const prepareRequest = async (): Promise => { - const attemptParams = structuredClone(params); - const replacementPayload = await options?.onPayload?.(attemptParams, model); - const wireParams = replacementPayload !== undefined ? (replacementPayload as typeof params) : attemptParams; + // Serialize once; the hook gets an isolated parse of exactly those + // bytes, and when no extension handles the event the wire reuses + // `bodyJson` instead of re-serializing (structuredClone + stringify + // measured 82ms on a 32MiB context where serialize-once costs 9ms). + const bodyJson = JSON.stringify(params); + let wireParams = params; + if (options?.onPayload) { + const attemptParams = JSON.parse(bodyJson) as typeof params; + const replacementPayload = await options.onPayload(attemptParams, model); + wireParams = + replacementPayload !== undefined && replacementPayload !== attemptParams + ? (replacementPayload as typeof params) + : attemptParams; + } activeRequestParams = wireParams; activeReasoningEffortFallbackKey = createOpenAIReasoningEffortFallbackKey( "azure-responses", url, typeof wireParams.model === "string" ? wireParams.model : model.id, ); - if (rawRequestDump) rawRequestDump.body = wireParams; - return { body: JSON.stringify(wireParams) }; + const body = wireParams === params ? bodyJson : JSON.stringify(wireParams); + wireBodyJson = body; + return { body }; }; const attemptedReasoningEffortFallbacks = new Set(); let openaiHandle: OpenAIStreamHandle; @@ -191,8 +205,9 @@ const streamAzureOpenAIResponsesOnce = ( const retryMarker = `${activeReasoningEffortFallbackKey}:${String(reasoningEffortFallback)}`; if (attemptedReasoningEffortFallbacks.has(retryMarker)) throw error; attemptedReasoningEffortFallbacks.add(retryMarker); + // The fallback-applied params reach `wireBodyJson` when the retried + // attempt's prepareRequest serializes them; no eager copy needed. applyOpenAIReasoningEffortFallback(params, reasoningEffortFallback); - rawRequestDump.body = params; } finally { clearTimeout(requestTimeout); } @@ -249,7 +264,11 @@ const streamAzureOpenAIResponsesOnce = ( stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { - const result = await AIError.finalize(error, { api: model.api, abortTracker, rawRequestDump }); + const result = await AIError.finalize(error, { + api: model.api, + abortTracker, + rawRequestDump: materializeDumpBody(rawRequestDump, wireBodyJson), + }); output.stopReason = result.stopReason; output.errorStatus = result.status; output.errorId = result.id; diff --git a/packages/ai/src/providers/google-gemini-cli.ts b/packages/ai/src/providers/google-gemini-cli.ts index 53532140c9..4a3c04fb39 100644 --- a/packages/ai/src/providers/google-gemini-cli.ts +++ b/packages/ai/src/providers/google-gemini-cli.ts @@ -38,7 +38,8 @@ import type { import { normalizeSystemPrompts } from "../utils"; import { AssistantMessageEventStream } from "../utils/event-stream"; import { extractGoogleValidationUrl, formatGoogleValidationRequiredMessage } from "../utils/google-validation"; -import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; + import { armPreResponseTimeout, getStreamFirstEventTimeoutMs } from "../utils/idle-iterator"; import { fetchProviderWithRetry } from "../utils/provider-fetch"; // Refresh is the sole responsibility of AuthStorage (broker-aware, single-flighted); @@ -538,6 +539,8 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( timestamp: Date.now(), }; let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let wireBodyJson: string | undefined; try { const apiKeyRaw = options?.apiKey; @@ -630,9 +633,9 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( api: output.api, model: model.id, method: "POST", - body: requestBody, headers: requestHeaders, }; + wireBodyJson = requestBodyJson; // Direct callers that skip `register-builtins` (which installs the // iterator-level watchdog) need a pre-response timer alongside @@ -1080,7 +1083,11 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { - const result = await AIError.finalize(error, { api: model.api, signal: options?.signal, rawRequestDump }); + const result = await AIError.finalize(error, { + api: model.api, + signal: options?.signal, + rawRequestDump: materializeDumpBody(rawRequestDump, wireBodyJson), + }); output.stopReason = result.stopReason; output.errorStatus = result.status; output.errorId = result.id; diff --git a/packages/ai/src/providers/google-shared.ts b/packages/ai/src/providers/google-shared.ts index 7b85ee75fd..6c73929e2d 100644 --- a/packages/ai/src/providers/google-shared.ts +++ b/packages/ai/src/providers/google-shared.ts @@ -32,7 +32,7 @@ import type { import { shouldSendServiceTier } from "../types"; import { normalizeSystemPrompts } from "../utils"; import { AssistantMessageEventStream } from "../utils/event-stream"; -import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; import { notifyProviderResponse } from "../utils/provider-response"; import { normalizeSchemaForCCA, normalizeSchemaForGoogle, toolWireSchema } from "../utils/schema"; import type { @@ -1076,6 +1076,8 @@ export function streamGoogleGenAI> => { const response = await fetchImpl(requestUrl, { @@ -1189,7 +1194,11 @@ export function streamGoogleGenAI => { - const attemptBody = structuredCloneJSON(body); - const replacementWireBody = await options?.onPayload?.(attemptBody, model); - wireBody = replacementWireBody !== undefined ? (replacementWireBody as RequestBody) : attemptBody; + // Serialize once. The hook, when present, gets an isolated parse of + // exactly those bytes; when no extension handles the event the wire + // object is the untouched original and the recorded bytes are reused. + // structuredCloneJSON + stringify measured 82ms on a 32MiB context where + // serialize-once costs 9ms — paid on every attempt before the first byte. + const bodyJson = JSON.stringify(body); + let wireParams = body; + if (options?.onPayload) { + const attemptBody = JSON.parse(bodyJson) as RequestBody; + const replacementWireBody = await options.onPayload(attemptBody, model); + wireParams = + replacementWireBody !== undefined && replacementWireBody !== attemptBody + ? (replacementWireBody as RequestBody) + : attemptBody; + } + wireBody = wireParams; // Keep the 400 dump honest: record the body actually sent on this attempt. - requestContext.rawRequestDump.body = wireBody; - return wireBody; + requestContext.wireBodyJson = wireParams === body ? bodyJson : JSON.stringify(wireParams); + return wireParams; }; // Preserve payload capture for callers that intentionally use an // already-aborted signal without issuing a physical request. @@ -1762,7 +1779,7 @@ async function handleCodexStreamFailure(context: CodexStreamFailureContext, erro const result = await AIError.finalize(error, { api: context.model.api, signal: context.options?.signal, - rawRequestDump: context.requestContext.rawRequestDump, + rawRequestDump: materializeDumpBody(context.requestContext.rawRequestDump, context.requestContext.wireBodyJson), }); output.stopReason = result.stopReason; output.errorStatus = result.status; diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index fedaae3619..6b05ab35be 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -47,7 +47,7 @@ import { withEmptyCompletionRetry, } from "../utils/empty-completion-retry"; import { AssistantMessageEventStream } from "../utils/event-stream"; -import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; import { getOpenAIStreamFirstEventTimeoutMs, getOpenAIStreamIdleTimeoutMs, @@ -608,6 +608,8 @@ const streamOpenAICompletionsOnce = ( const output: AssistantMessage = createInitialResponsesAssistantMessage(model.api, model.provider, model.id); let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let wireBodyJson: string | undefined; const abortTracker = createAbortSourceTracker(options?.signal); const firstEventTimeoutAbortError = new AIError.StreamTimeoutError( OPENAI_COMPLETIONS_FIRST_EVENT_TIMEOUT_MESSAGE, @@ -684,11 +686,27 @@ const streamOpenAICompletionsOnce = ( } activeReasoningEffortFallbackKey = reasoningEffortFallbackKey; const prepareRequest = async (): Promise => { - const attemptParams = structuredClone(params); - const replacementPayload = await options?.onPayload?.(attemptParams, model); - const wireParams = - replacementPayload !== undefined ? (replacementPayload as OpenAICompletionsParams) : attemptParams; + // Serialize once. The hook, when present, gets an isolated parse of + // exactly those bytes; when no extension handles the event it + // returns that same object, and the wire reuses `bodyJson` instead + // of re-serializing. structuredClone + stringify measured 82ms on a + // 32MiB context where serialize-once costs 9ms — paid on every + // submit before the first byte leaves the process. + const bodyJson = JSON.stringify(params); + let wireParams = params; + if (options?.onPayload) { + const attemptParams = JSON.parse(bodyJson) as OpenAICompletionsParams; + const replacementPayload = await options.onPayload(attemptParams, model); + wireParams = + replacementPayload !== undefined && replacementPayload !== attemptParams + ? (replacementPayload as OpenAICompletionsParams) + : attemptParams; + } activeRequestParams = wireParams; + const body = wireParams === params ? bodyJson : JSON.stringify(wireParams); + // Retain the exact sent BYTES, not the parsed object: a dump body + // is read only on the 400/413 path, and holding the graph here + // pinned a full context-sized clone for the whole stream. rawRequestDump = { provider: model.provider, api: output.api, @@ -696,9 +714,9 @@ const streamOpenAICompletionsOnce = ( method: "POST", url: completionsUrl, headers: requestHeaders, - body: wireParams, }; - return { body: JSON.stringify(wireParams) }; + wireBodyJson = body; + return { body }; }; if (captureOnly) { await prepareRequest(); @@ -1429,7 +1447,7 @@ const streamOpenAICompletionsOnce = ( api: model.api, provider: model.provider, abortTracker, - rawRequestDump, + rawRequestDump: materializeDumpBody(rawRequestDump, wireBodyJson), capturedErrorResponse, }); output.stopReason = result.stopReason; diff --git a/packages/ai/src/providers/openai-responses.ts b/packages/ai/src/providers/openai-responses.ts index af68e274ad..f6b1f88daa 100644 --- a/packages/ai/src/providers/openai-responses.ts +++ b/packages/ai/src/providers/openai-responses.ts @@ -29,7 +29,7 @@ import { import { createAbortSourceTracker } from "../utils/abort"; import { withEmptyCompletionRetry } from "../utils/empty-completion-retry"; import { AssistantMessageEventStream } from "../utils/event-stream"; -import type { RawHttpRequestDump } from "../utils/http-inspector"; +import { materializeDumpBody, type RawHttpRequestDump } from "../utils/http-inspector"; import { getOpenAIStreamFirstEventTimeoutMs, getOpenAIStreamIdleTimeoutMs, @@ -381,6 +381,9 @@ const streamOpenAIResponsesOnce = ( const output: AssistantMessage = createInitialResponsesAssistantMessage(model.api, model.provider, model.id); let rawRequestDump: RawHttpRequestDump | undefined; + /** Exact bytes of the last sent request body; materialized into a dump only on the 400/413 path. */ + let wireBodyJson: string | undefined; + let chainState: OpenAIResponsesChainState | undefined; let sentPreviousResponseId: string | undefined; const abortTracker = createAbortSourceTracker(options?.signal); @@ -454,13 +457,32 @@ const streamOpenAIResponsesOnce = ( const requestTimeoutMs = firstEventTimeoutMs !== undefined && firstEventTimeoutMs > 0 ? firstEventTimeoutMs : undefined; const requestUrl = `${resolvedBaseUrl}/responses`; - const applyPayloadReplacement = async (requestParams: OpenAIResponsesSamplingParams) => { - const attemptParams = structuredCloneJSON(requestParams); - const replacementPayload = await options?.onPayload?.(attemptParams, model); - const payload = - replacementPayload !== undefined ? (replacementPayload as OpenAIResponsesSamplingParams) : attemptParams; - applyReasoningEffortFallbackForRequest(payload); - return payload; + const applyPayloadReplacement = async ( + requestParams: OpenAIResponsesSamplingParams, + ): Promise<{ wireParams: OpenAIResponsesSamplingParams; bodyJson: string }> => { + // Serialize once; the hook gets an isolated parse of exactly those + // bytes, and when no extension handles the event the caller reuses + // `bodyJson` instead of re-serializing. The reasoning-effort + // fallback may still mutate the parsed object afterwards, which is + // what `reused` guards. + const bodyJson = JSON.stringify(requestParams); + let attemptParams = requestParams; + if (options?.onPayload) { + const hookView = JSON.parse(bodyJson) as OpenAIResponsesSamplingParams; + const replacementPayload = await options.onPayload(hookView, model); + attemptParams = + replacementPayload !== undefined && replacementPayload !== hookView + ? (replacementPayload as OpenAIResponsesSamplingParams) + : hookView; + } + const fallbackKey = applyReasoningEffortFallbackForRequest(attemptParams); + const fallbackApplied = + requestReasoningEffortFallbacks.has(fallbackKey) || + getOpenAIReasoningEffortFallback(providerSessionState, fallbackKey) !== undefined; + return { + wireParams: attemptParams, + bodyJson: fallbackApplied || attemptParams !== requestParams ? JSON.stringify(attemptParams) : bodyJson, + }; }; rawRequestDump = { provider: model.provider, @@ -468,19 +490,18 @@ const streamOpenAIResponsesOnce = ( model: model.id, method: "POST", url: requestUrl, - body: chained.params, }; const openResponsesStream = async (requestParams: OpenAIResponsesSamplingParams, captureOnly = false) => { const prepareRequest = async (): Promise => { - const wireParams = await applyPayloadReplacement(requestParams); + const { wireParams, bodyJson } = await applyPayloadReplacement(requestParams); activeReasoningEffortFallbackKey = createOpenAIReasoningEffortFallbackKey( "responses", resolvedBaseUrl, typeof wireParams.model === "string" ? wireParams.model : model.id, ); activeRequestParams = wireParams; - if (rawRequestDump) rawRequestDump.body = wireParams; - return { body: JSON.stringify(wireParams) }; + wireBodyJson = bodyJson; + return { body: bodyJson }; }; if (captureOnly) { await prepareRequest(); @@ -561,7 +582,6 @@ const streamOpenAIResponsesOnce = ( requestReasoningEffortFallbacks.set(activeReasoningEffortFallbackKey, reasoningEffortFallback); applyOpenAIReasoningEffortFallback(chained.params, reasoningEffortFallback); applyOpenAIReasoningEffortFallback(activeParams, reasoningEffortFallback); - rawRequestDump.body = chained.params; pendingReasoningEffortFallback = { key: activeReasoningEffortFallbackKey, fallback: reasoningEffortFallback, @@ -601,7 +621,6 @@ const streamOpenAIResponsesOnce = ( : { params: fallbackParams }; sentPreviousResponseId = fallbackChained.previousResponseId; chained = fallbackChained; - rawRequestDump.body = chained.params; activeParams = fallbackParams; activeStrictToolsApplied = fallbackBuilt.strictToolsApplied; continue; @@ -642,7 +661,6 @@ const streamOpenAIResponsesOnce = ( // breaker only trips when each retry stores and the next turn re-chains. currentParams.store = !zdrRejection; chained = { params: currentParams }; - rawRequestDump.body = currentParams; activeParams = currentParams; activeStrictToolsApplied = currentBuilt.strictToolsApplied; } @@ -747,7 +765,7 @@ const streamOpenAIResponsesOnce = ( api: model.api, provider: model.provider, abortTracker, - rawRequestDump, + rawRequestDump: materializeDumpBody(rawRequestDump, wireBodyJson), capturedErrorResponse, }); output.stopReason = result.stopReason; diff --git a/packages/ai/src/providers/pi-native-client.ts b/packages/ai/src/providers/pi-native-client.ts index eeda6bff60..fe1f45e587 100644 --- a/packages/ai/src/providers/pi-native-client.ts +++ b/packages/ai/src/providers/pi-native-client.ts @@ -197,25 +197,31 @@ export function streamPiNative( options: buildWireOptions(options), stream: true, }; - try { - const onPayload = options?.onPayload; - if (onPayload) { - // The hook is a JSON seam: a host's secret redactor walks the payload - // rewriting every string and refuses any value JSON cannot express. - // `context` carries live arktype schemas in `tools[].parameters`, - // which are function objects, so the raw object is never that shape. - // The wire form is JSON by construction (the body is stringified - // below), so the hook sees exactly the wire shape. - const wirePayload: unknown = JSON.parse(JSON.stringify(bodyPayload)); - const replacementPayload = await onPayload(wirePayload, model as Model); - if (replacementPayload !== undefined) bodyPayload = replacementPayload; + const onPayload = options?.onPayload; + // The hook is a JSON seam: a host's secret redactor walks the payload + // rewriting every string and refuses any value JSON cannot express. + // `context` carries live arktype schemas in `tools[].parameters`, + // which are function objects, so the raw object is never that shape. + // Serialize once up front; the hook gets an isolated parse of those + // bytes, and when it leaves the payload alone the wire reuses them — + // a full-context body is never serialized twice. + let body = JSON.stringify(bodyPayload); + if (onPayload) { + const wirePayload: unknown = JSON.parse(body); + let replacementPayload: unknown; + try { + replacementPayload = await onPayload(wirePayload, model as Model); + } catch (error) { + // Payload sanitization is a local policy decision, not an upstream + // authentication failure. Keep the rejection out of the + // auth-retry classifier even when its original error resembles a 401. + throw new PiNativePayloadHookError(error); + } + if (replacementPayload !== undefined) { + bodyPayload = replacementPayload; + body = JSON.stringify(bodyPayload); } - } catch (error) { - // Payload sanitization is a local policy decision, not an upstream authentication failure. Keep - // the rejection out of the auth-retry classifier even when its original error resembles a 401. - throw new PiNativePayloadHookError(error); } - const body = JSON.stringify(bodyPayload); response = await fetchImpl(url, { method: "POST", headers, body, signal: abortTracker.requestSignal }); if (!response.ok) { diff --git a/packages/ai/src/utils/http-inspector.ts b/packages/ai/src/utils/http-inspector.ts index 7815a0bcf2..633be8bd11 100644 --- a/packages/ai/src/utils/http-inspector.ts +++ b/packages/ai/src/utils/http-inspector.ts @@ -15,6 +15,28 @@ export type RawHttpRequestDump = { body?: unknown; }; +/** + * Attach the request body to a dump at error time. Providers retain only the + * exact sent bytes for the life of a stream — holding the parsed object pinned + * a full context-sized clone on every in-flight request just to serve a + * hypothetical 400/413 dump. This parses those bytes back into `body` once, + * when a dump is actually being built; bytes that never parse (never sent, or + * truncated by an abort) leave `body` unset rather than fabricating one. + */ +export function materializeDumpBody( + dump: RawHttpRequestDump | undefined, + wireBodyJson: string | undefined, +): RawHttpRequestDump | undefined { + if (!dump || wireBodyJson === undefined) return dump; + if (dump.body !== undefined) return dump; + try { + dump.body = JSON.parse(wireBodyJson) as unknown; + } catch { + // Unparseable body: the dump still carries method/url/headers. + } + return dump; +} + export type CapturedHttpErrorResponse = { status: number; headers?: Headers; diff --git a/packages/ai/test/a-failed-request-still-reports-its-duration.test.ts b/packages/ai/test/a-failed-request-still-reports-its-duration.test.ts new file mode 100644 index 0000000000..6345b941cd --- /dev/null +++ b/packages/ai/test/a-failed-request-still-reports-its-duration.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, test } from "bun:test"; +import { streamBedrock } from "@veyyon/ai/providers/amazon-bedrock"; +import type { Context, FetchImpl, Model } from "@veyyon/ai/types"; +import { buildModel } from "@veyyon/catalog/build"; + +/** + * WHY: every provider assigns `output.duration` twice — once where the stream + * finishes and once where it errors. Refactoring the error path is how one of + * those assignments disappears: the success path still reports a duration, the + * suite stays green, and only failed turns lose their timing. That happened to + * `streamBedrock`, where a change to the diagnostic dump removed the error-path + * assignment and left `ttft` beside it, so nothing looked missing. + * + * The class this closes: a provider that reports a duration when it succeeds + * and reports none when it fails. It is asserted against the real + * `streamBedrock` driven through a transport that rejects, not against a fake. + * + * WHAT IT DOES NOT CATCH, stated plainly: this drives Bedrock only. The other + * ten providers carry the same pair of assignments and the same exposure, and + * a sweep over all of them needs a per-provider driver (credentials, model + * identity, transport shape) that does not exist yet. A provider added or + * refactored outside Bedrock can still lose its error-path duration silently. + */ + +const model: Model<"bedrock-converse-stream"> = buildModel({ + id: "anthropic.claude-3-5-sonnet-20241022-v2:0", + name: "Bedrock duration probe", + api: "bedrock-converse-stream", + provider: "amazon-bedrock", + baseUrl: "https://bedrock-runtime.us-east-1.amazonaws.com", + reasoning: true, + input: ["text", "image"], + cost: { input: 5, output: 25, cacheRead: 0.5, cacheWrite: 6.25 }, + contextWindow: 1000000, + maxTokens: 128000, +}); + +function userContext(): Context { + return { messages: [{ role: "user", content: "Say hello", timestamp: 0 }] }; +} +/** + * A transport that always rejects with a status the retry policy does NOT + * re-attempt, so the turn reaches its error path once instead of backing off. + */ +function rejectingFetch(status: number): FetchImpl { + return Object.assign( + async (_input: string | URL | Request, _init?: RequestInit) => new Response("upstream refused", { status }), + { preconnect: fetch.preconnect }, + ); +} + +describe("a failed provider request still reports its duration", () => { + test("bedrock records a duration on the error path, not only the success path", async () => { + const result = await streamBedrock(model, userContext(), { + bearerToken: "test-token", + fetch: rejectingFetch(400), + maxTokens: 16, + }).result(); + + // The turn must be a failure, or this asserts the success path by accident. + expect(result.stopReason).toBe("error"); + expect(typeof result.duration).toBe("number"); + expect(result.duration).toBeGreaterThan(0); + }); + + test("the duration is the elapsed turn, not a value carried from somewhere else", async () => { + // A duration that never moves would satisfy "is a number greater than + // zero" while measuring nothing. Two turns cannot both report the same + // monotonic elapsed time unless the field is constant. + const run = async (): Promise => { + const result = await streamBedrock(model, userContext(), { + bearerToken: "test-token", + fetch: rejectingFetch(418), + maxTokens: 16, + }).result(); + expect(result.stopReason).toBe("error"); + return result.duration ?? -1; + }; + + const first = await run(); + const second = await run(); + expect(first).toBeGreaterThan(0); + expect(second).toBeGreaterThan(0); + // Bounded: a turn against an in-process transport cannot take a minute, + // so a duration that large means the field is a timestamp, not a span. + expect(first).toBeLessThan(60_000); + expect(second).toBeLessThan(60_000); + }); +}); diff --git a/packages/ai/test/http-inspector.test.ts b/packages/ai/test/http-inspector.test.ts index 487b51cba6..e62f5b960b 100644 --- a/packages/ai/test/http-inspector.test.ts +++ b/packages/ai/test/http-inspector.test.ts @@ -3,6 +3,7 @@ import { buildHttp400DumpPayload, captureHttpErrorResponse, finalizeErrorMessage, + materializeDumpBody, type RawHttpRequestDump, shouldDumpRejectedRequest, } from "@veyyon/ai/utils/http-inspector"; @@ -53,6 +54,40 @@ describe("buildHttp400DumpPayload", () => { }); }); +describe("materializeDumpBody", () => { + it("parses the retained wire bytes into the dump body at error time", () => { + const dump: RawHttpRequestDump = { provider: "openai", api: "openai-completions", model: "gpt-test" }; + const materialized = materializeDumpBody(dump, '{"model":"gpt-test","messages":[{"role":"user"}]}'); + expect(materialized).toBe(dump); + expect(dump.body).toEqual({ model: "gpt-test", messages: [{ role: "user" }] }); + }); + + it("returns a body-less dump unchanged when bytes were never sent", () => { + const dump: RawHttpRequestDump = { provider: "openai", api: "openai-completions", model: "gpt-test" }; + expect(materializeDumpBody(dump, undefined)).toBe(dump); + expect(dump.body).toBeUndefined(); + expect(materializeDumpBody(undefined, "{}")).toBeUndefined(); + }); + + it("never overwrites a body that is already present", () => { + const existing = { kept: true }; + const dump: RawHttpRequestDump = { + provider: "openai", + api: "openai-completions", + model: "gpt-test", + body: existing, + }; + materializeDumpBody(dump, '{"replaced":true}'); + expect(dump.body).toBe(existing); + }); + + it("tolerates unparseable bytes instead of throwing out of an error path", () => { + const dump: RawHttpRequestDump = { provider: "openai", api: "openai-completions", model: "gpt-test" }; + expect(() => materializeDumpBody(dump, "{truncated")).not.toThrow(); + expect(dump.body).toBeUndefined(); + }); +}); + describe("shouldDumpRejectedRequest", () => { it("captures request-content rejections (400 bad request, 413 payload too large)", () => { expect(shouldDumpRejectedRequest(new HttpError(400, "bad request"))).toBe(true); diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index e4e771363a..e9c1a22682 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -7,6 +7,7 @@ - Classified runner output (cargo, bun, Go, ctest, dotnet, clippy, golangci-lint, Gradle lint, pytest, and tsc/eslint-family) now opens with a result-contract header: `[clean] ` or `[errors]` / `[errors N] `. The header is the verdict and the body contains retained diagnostics. ### Added +- `VEYYON_DEBUG_STARTUP=1` writes one line per phase of a prompt submission (compaction check, plan arm, context build, memory context), so a slow submit names the phase that spent the time. - `read` takes `depth` and `limit` arguments for directory listings, and a read of the session working directory root with neither now returns a concise top-level listing with per-subdirectory entry counts instead of the recursive tree. - A tool result that carries an image now states whether the picture reached the screen, so a model reading a file describes what it shows instead of reporting that it displayed it. - A picture the block gives up on after the fact, because the session's image budget demoted it or a Kitty session could not convert it, is stated to the model as undrawn instead of being reported as displayed. diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 85cdc64f67..55c18fe20c 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -178,6 +178,7 @@ import { withScopedTimeoutSignal, withTimeout, } from "@veyyon/utils"; +import { startupMarker } from "@veyyon/utils/startup-marker"; import type { ArgotSession } from "argot"; import { ADVISOR_DEFAULT_TOOL_NAMES, @@ -11010,6 +11011,7 @@ export class AgentSession { }, ): Promise { this.#beginInFlight(); + startupMarker("prompt:start"); const generation = this.#promptGeneration; try { // Flush any pending bash messages before the new prompt @@ -11087,16 +11089,24 @@ export class AgentSession { ); } + // Phase markers for the submit path: VEYYON_DEBUG_STARTUP=1 writes one + // synchronous stderr line per phase, so a "submit feels slow" report + // names the phase that spent the time instead of offering a guess. + startupMarker("prompt:compaction-check:start"); // Check whether an aborted response left enough context pressure to require // in-place compaction before this prompt starts its agent loop. const lastAssistant = this.#findLastAssistantMessage(); if (lastAssistant && !options?.skipCompactionCheck) { await this.#checkCompaction(lastAssistant, false, false); } + startupMarker("prompt:compaction-check:done"); + startupMarker("prompt:plan-arm:start"); await this.#armPlanYoloIfNeeded(); + startupMarker("prompt:plan-arm:done"); // Build messages array (session context, eager todo prelude, then active prompt message) + startupMarker("prompt:context-build:start"); const messages: AgentMessage[] = []; const planReferenceMessage = await this.#buildPlanReferenceMessage?.(); if (planReferenceMessage) { @@ -11149,15 +11159,19 @@ export class AgentSession { // should already know when it reads the question, which is how the eager-task // prelude is placed too. Position within the turn is free either way — the // cache prefix ends before all of it. + startupMarker("prompt:memory-context:start"); const memoryContextMessage = await this.#collectVolatileMemoryContext(expandedText); if (memoryContextMessage) messages.unshift(memoryContextMessage); + startupMarker("prompt:memory-context:done"); + startupMarker("prompt:context-build:done"); + // Ahead of the memories for the same reason the memories are ahead of the // question: the date and the working directory are what the model should // already know when it reads either. const sessionStateMessage = this.#buildSessionStateMessage(); if (sessionStateMessage) messages.unshift(sessionStateMessage); const beforeAgentStartSystemPrompt = this.#baseSystemPrompt; - + startupMarker("prompt:before-agent-start:start"); // Emit before_agent_start extension event if (this.#extensionRunner) { const result = await this.#extensionRunner.emitBeforeAgentStart( @@ -11200,6 +11214,8 @@ export class AgentSession { this.agent.setSystemPrompt(beforeAgentStartSystemPrompt); } + startupMarker("prompt:before-agent-start:done"); + // Bail out if a newer abort/prompt cycle has started since we began setup if (this.#promptGeneration !== generation) { return; @@ -11216,7 +11232,9 @@ export class AgentSession { } } + startupMarker("prompt:pre-prompt-compaction:start"); await this.#runPrePromptCompactionIfNeeded(messages); + startupMarker("prompt:pre-prompt-compaction:done"); if (this.#promptGeneration !== generation) { return; } diff --git a/packages/coding-agent/test/a-superseded-prompt-stops-before-classifying-thinking.test.ts b/packages/coding-agent/test/a-superseded-prompt-stops-before-classifying-thinking.test.ts new file mode 100644 index 0000000000..a7ad49a32f --- /dev/null +++ b/packages/coding-agent/test/a-superseded-prompt-stops-before-classifying-thinking.test.ts @@ -0,0 +1,116 @@ +/** + * WHY: `AgentSession.prompt` awaits the `before_agent_start` extension hook, + * and a host extension holds that await open for as long as it likes. The + * generation check that follows it was once deleted in favour of a startup + * marker, which let a turn the user had already aborted go on to issue an + * auto-thinking classifier request and to run pre-prompt compaction before the + * next check stopped it. That request is billed and that compaction pass + * rewrites session context, both for a turn that no longer exists. + * + * The class this closes: an awaited setup stage between prompt entry and the + * model request re-checks the prompt generation before spending anything else + * on the turn. Driven here at the `before_agent_start` seam, the one a host + * controls directly, with a positive control proving the classifier does run + * for a turn that is not superseded — without it, a suite that only asserts + * "not called" passes when the classifier is unreachable for any reason. + * + * What it does not catch: a new awaited stage inserted between two existing + * checks. The stages are statements in one method rather than a registry, so + * they cannot be enumerated at run time and a new one is not detected here. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test"; +import * as path from "node:path"; +import { Agent } from "@veyyon/agent-core"; +import { Effort } from "@veyyon/ai"; +import { createMockModel } from "@veyyon/ai/providers/mock"; +import { getBundledModel } from "@veyyon/catalog/models"; +import * as classifier from "@veyyon/coding-agent/auto-thinking/classifier"; +import { ModelRegistry } from "@veyyon/coding-agent/config/model-registry"; +import { Settings } from "@veyyon/coding-agent/config/settings"; +import type { ExtensionRunner } from "@veyyon/coding-agent/extensibility/extensions"; +import { AgentSession } from "@veyyon/coding-agent/session/agent-session"; +import { AuthStorage } from "@veyyon/coding-agent/session/auth-storage"; +import { SessionManager } from "@veyyon/coding-agent/session/session-manager"; +import { AUTO_THINKING } from "@veyyon/coding-agent/thinking"; +import { TempDir } from "@veyyon/utils"; + +describe("a prompt superseded inside before_agent_start", () => { + let tempDir: TempDir; + let session: AgentSession; + let modelRegistry: ModelRegistry; + let authStorage: AuthStorage | undefined; + + beforeEach(async () => { + tempDir = TempDir.createSync("@pi-superseded-before-agent-start-"); + authStorage = await AuthStorage.create(path.join(tempDir.path(), "testauth.db")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + modelRegistry = new ModelRegistry(authStorage); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + if (session) await session.dispose(); + authStorage?.close(); + authStorage = undefined; + tempDir.removeSync(); + }); + + /** + * `onHook` runs while the session is suspended inside the extension hook, + * which is the window an abort has to land in for this contract to matter. + */ + function createSession(onHook?: () => void) { + const emitBeforeAgentStart = vi.fn(async () => { + onHook?.(); + return undefined; + }); + const extensionRunner = { + emitBeforeAgentStart, + emit: vi.fn().mockResolvedValue(undefined), + } as unknown as ExtensionRunner; + + const model = getBundledModel("anthropic", "claude-sonnet-4-5"); + if (!model) throw new Error("Expected claude-sonnet-4-5 model to exist"); + + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [], messages: [] }, + streamFn: createMockModel({ responses: [{ content: ["Done"] }] }).stream, + }); + + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings: Settings.isolated({ "compaction.enabled": false }), + modelRegistry, + extensionRunner, + thinkingLevel: AUTO_THINKING, + }); + + return { emitBeforeAgentStart }; + } + + it("classifies the thinking level when the turn is not superseded", async () => { + const classify = vi.spyOn(classifier, "classifyDifficulty").mockResolvedValue(Effort.Medium); + const { emitBeforeAgentStart } = createSession(); + + await session.prompt("write the parser"); + + expect(emitBeforeAgentStart).toHaveBeenCalledTimes(1); + expect(classify).toHaveBeenCalledTimes(1); + }); + + it("issues no classifier request when an abort lands inside the hook", async () => { + const classify = vi.spyOn(classifier, "classifyDifficulty").mockResolvedValue(Effort.Medium); + let aborted: Promise | undefined; + const { emitBeforeAgentStart } = createSession(() => { + aborted = session.abort({ reason: "user-interrupt" }); + }); + + await session.prompt("write the parser"); + await aborted; + + expect(emitBeforeAgentStart).toHaveBeenCalledTimes(1); + expect(classify).not.toHaveBeenCalled(); + }); +});