From 1c9acde3c2ef3f39eeaff25cffb481608bd7f900 Mon Sep 17 00:00:00 2001 From: Sinabina Date: Mon, 13 Jul 2026 12:46:23 -0700 Subject: [PATCH] fix(gateway): consolidate tool-loop resume + provider fixes (fix-wave A) Collector branch superseding the gateway tool-loop duplicate cluster and adjacent provider fixes. Re-implemented from the best of each PR (deduped by content, not file-overlap); every fix carries test coverage. Gateway tool-loop resume (supersedes #1934 #2062 #2065 #2112 #2274 #2487 #2336 #2257 #2499 #2491, test #2063): - toolLoop now persists the tool-result user turn per round (onToolResultTurn), so a resumed subagent job reloads a balanced transcript instead of dangling assistant tool-calls that non-Anthropic providers reject with AI_MissingToolResultsError. - runSubagentViaGateway reconciles an already-corrupted transcript on resume: it heals every dangling assistant tool-call turn (not just the tail) from the settled subagent_tool_executions rows, re-dispatching idempotent-pending tools and throwing on non-idempotent, mirroring the legacy Anthropic path. Terminal early-return for a transcript that already reached end_turn. - repairToolPairing() is a last-resort normalization at the chat() boundary (from #2336): back-fills error stubs for any assistant tool-call still unanswered (partial turns, provider-duplicated/dropped IDs on local models, length-truncated batches). No-op on balanced input. - toModelMessages is Date-safe (Postgres timestamptz -> ISO via a JSON round trip at the SDK boundary, never a ::jsonb cast; degrades bigint/circular to a string instead of throwing) and drops non-string text blocks reasoning models emit that AI SDK v6 rejects (#2488). Adjacent provider fixes: - Model-aware default max output tokens: thinking-by-default Claude 5 models get headroom (gateway 32000, think 16000) while everything else stays 4096/4000, so DeepSeek/OpenAI subagents don't exceed provider caps (#2614 #2806). - DeepSeek: promote reasoning_content into content when content is empty, via a fail-open recipe fetch shim (#2617). - OpenRouter: map openrouter_api_key (config + env) into OPENROUTER_API_KEY through buildGatewayConfig; register agent.use_gateway_loop, zeroentropy and openrouter keys in KNOWN_CONFIG_KEYS so `config set` accepts them (#2572, config key from #2112). Preserves JSONB (no JSON.stringify into ::jsonb), engine parity, source isolation, and trust-boundary invariants. Verified with the gbrain-pr-test-env consumer matrix (clawlancer/Postgres, gstack + hivemindos/PGLite) baseline-FAIL -> candidate-PASS on the same repro, plus bun run verify (31/31) and 439 targeted unit + e2e tests. Co-Authored-By: thomaskong119 Co-Authored-By: maxpetrusenkoagent Co-Authored-By: brettdavies Co-Authored-By: ivandebot <187176982+ivandebot@users.noreply.github.com> Co-Authored-By: Rafael Reis Co-Authored-By: fbal23 Co-Authored-By: javieraldape Co-Authored-By: David Carolan Co-Authored-By: Masashi-Ono0611 Co-Authored-By: spiky02plateau <155588579+spiky02plateau@users.noreply.github.com> Co-Authored-By: psam-717 Co-Authored-By: Claude Opus 4.8 (1M context) --- src/core/ai/build-gateway-config.ts | 4 + src/core/ai/gateway.ts | 155 +++++++++- src/core/ai/recipes/deepseek.ts | 60 ++++ src/core/ai/types.ts | 12 + src/core/config.ts | 15 + src/core/minions/handlers/subagent.ts | 226 +++++++++++++- src/core/think/index.ts | 15 +- test/ai/build-gateway-config.test.ts | 20 ++ test/ai/deepseek-reasoning-content.test.ts | 124 ++++++++ test/ai/gateway-tool-loop.test.ts | 65 ++++ test/ai/gateway-toolcall-pairing.test.ts | 100 +++++++ test/config-set.test.ts | 9 + ...gent-gateway-resume-reconciliation.test.ts | 280 ++++++++++++++++++ test/gateway-model-messages.test.ts | 57 ++++ test/think-max-output-tokens.test.ts | 28 ++ 15 files changed, 1143 insertions(+), 27 deletions(-) create mode 100644 test/ai/deepseek-reasoning-content.test.ts create mode 100644 test/ai/gateway-toolcall-pairing.test.ts create mode 100644 test/e2e/subagent-gateway-resume-reconciliation.test.ts create mode 100644 test/think-max-output-tokens.test.ts diff --git a/src/core/ai/build-gateway-config.ts b/src/core/ai/build-gateway-config.ts index 1b5eb9613..9996d3ffa 100644 --- a/src/core/ai/build-gateway-config.ts +++ b/src/core/ai/build-gateway-config.ts @@ -34,6 +34,10 @@ export function buildGatewayConfig(c: GBrainConfig): AIGatewayConfig { // plane field now exists (GBrainConfig type) and gets mapped here, so // setting it via `~/.gbrain/config.json` propagates into the gateway. if (c.zeroentropy_api_key) envFromConfig.ZEROENTROPY_API_KEY = c.zeroentropy_api_key; + // Same seam for OpenRouter: `gbrain config set openrouter_api_key X` (or + // config.json) must reach the openrouter recipe's OPENROUTER_API_KEY. + // process.env still wins via the later spread. + if (c.openrouter_api_key) envFromConfig.OPENROUTER_API_KEY = c.openrouter_api_key; // v0.32 codex finding #4+#5 fix: thread local-server _BASE_URL env vars // into base_urls so the gateway hits the user's configured port. Without diff --git a/src/core/ai/gateway.ts b/src/core/ai/gateway.ts index fdcae8779..48cfa2015 100644 --- a/src/core/ai/gateway.ts +++ b/src/core/ai/gateway.ts @@ -377,7 +377,8 @@ export function applyOpenAICompatConfig( cfg: AIGatewayConfig, ): { baseURL: string; fetch?: typeof fetch } { if (recipe.resolveOpenAICompatConfig) { - return recipe.resolveOpenAICompatConfig(cfg.env); + const resolved = recipe.resolveOpenAICompatConfig(cfg.env); + return { ...resolved, fetch: resolved.fetch ?? recipe.compat?.fetch }; } const baseURL = cfg.base_urls?.[recipe.id] ?? recipe.base_url_default; if (!baseURL) { @@ -386,7 +387,7 @@ export function applyOpenAICompatConfig( recipe.setup_hint, ); } - return { baseURL }; + return { baseURL, fetch: recipe.compat?.fetch }; } /** @@ -2383,6 +2384,58 @@ export interface ChatToolDef { * production subagent jobs) throws "messages do not match the ModelMessage[] * schema" the moment the model calls a tool. Surfaced by the SkillOpt eval. */ +/** + * Default per-call max output tokens. Thinking-by-default Claude 5 models + * (`anthropic:claude-*-5`) burn a large chunk of the budget on internal + * reasoning before emitting any text, so a 4096 default leaves them with empty + * final text on the subagent tool loop. Give those models headroom; providers + * bill actual tokens, not the cap, so it is free for the models that don't use + * it. Everything else keeps 4096 on purpose: raising the default blanket-wide + * would exceed some openai-compat providers' hard max-output caps (DeepSeek + * 8192, gpt-4o 16384) and 400 on them — a regression for exactly the + * non-Anthropic subagent users the gateway loop exists to serve. + */ +const DEFAULT_MAX_OUTPUT_TOKENS = 4096; +const THINKING_MODEL_MAX_OUTPUT_TOKENS = 32000; +const THINKING_BY_DEFAULT_MODEL_RE = /^anthropic[:/]claude-[a-z0-9]+-5(?:[.-]|$)/i; +function defaultMaxOutputTokens(modelStr: string | undefined): number { + return modelStr && THINKING_BY_DEFAULT_MODEL_RE.test(modelStr) + ? THINKING_MODEL_MAX_OUTPUT_TOKENS + : DEFAULT_MAX_OUTPUT_TOKENS; +} + +/** + * Deep-serialize a tool output into a plain JSON value for the AI SDK v6 + * ModelMessage schema. node-postgres returns `timestamptz` columns as JS + * `Date` instances, and AI SDK v6's `JSONValue` schema rejects a raw Date, + * throwing "Invalid prompt ... ModelMessage[] schema" the moment a + * timestamp-bearing tool result (e.g. `brain_get_page`, `brain_list_pages`) + * is fed back — dead-lettering the whole multi-tool loop. The JSON round-trip + * runs `Date.prototype.toJSON` (ISO string) recursively and drops `undefined`. + * This is a serialization fix at the SDK boundary, NOT a `::jsonb` DB cast — + * it never touches Postgres. (BigInt / circular outputs still throw in + * JSON.stringify; those aren't LLM-serializable and are out of scope.) + */ +function toJsonSafe(value: unknown): unknown { + try { + return JSON.parse(JSON.stringify(value ?? null)); + } catch { + // BigInt / circular output isn't LLM-serializable; degrade to a string + // rather than throwing and dead-lettering the whole tool loop. + return safeStringify(value); + } +} + +/** Stringify that never throws (bigint/circular fall back to String()). */ +function safeStringify(value: unknown): string { + if (typeof value === 'string') return value; + try { + return JSON.stringify(value ?? null); + } catch { + return String(value); + } +} + export function toModelMessages(messages: ChatMessage[]): unknown[] { return messages.map((m) => { if (typeof m.content === 'string') return { role: m.role, content: m.content }; @@ -2398,24 +2451,86 @@ export function toModelMessages(messages: ChatMessage[]): unknown[] { toolCallId: b.toolCallId, toolName: b.toolName, output: b.isError - ? { type: 'error-text' as const, value: typeof b.output === 'string' ? b.output : JSON.stringify(b.output) } + ? { type: 'error-text' as const, value: safeStringify(b.output) } : (typeof b.output === 'string' ? { type: 'text' as const, value: b.output } - : { type: 'json' as const, value: (b.output ?? null) as never }), + : { type: 'json' as const, value: toJsonSafe(b.output) as never }), })), }; } return { role: m.role, - content: blocks.map((b) => { - if (b.type === 'text') return { type: 'text' as const, text: b.text }; - if (b.type === 'tool-call') return { type: 'tool-call' as const, toolCallId: b.toolCallId, toolName: b.toolName, input: b.input }; - return b; - }), + // Drop text blocks whose `text` isn't a string: reasoning models + // (DeepSeek v4, etc.) surface `text: null/undefined` thinking parts that + // AI SDK v6's Zod schema rejects, poisoning the whole call. `''` is valid + // and kept. + content: blocks + .filter((b) => b.type !== 'text' || typeof b.text === 'string') + .map((b) => { + if (b.type === 'text') return { type: 'text' as const, text: b.text }; + if (b.type === 'tool-call') return { type: 'tool-call' as const, toolCallId: b.toolCallId, toolName: b.toolName, input: b.input }; + return b; + }), }; }); } +/** + * Last-resort normalization at the `chat()` boundary: back-fill error stubs for + * any assistant tool-call that isn't answered by the immediately-following + * tool-result turn. The subagent handler already balances its own transcript + * (see reconcileGatewayReplay), so this is a no-op there — it exists for the + * paths reconcile can't reach: a partially-answered turn, a provider that + * duplicates or drops tool-call IDs (local vLLM), or a `finishReason:'length'` + * truncation mid-batch. Without it those histories throw + * AI_MissingToolResultsError inside `generateText`. No-op on balanced input. + * + * @internal exported for tests. + */ +export function repairToolPairing(messages: ChatMessage[]): ChatMessage[] { + const out: ChatMessage[] = []; + for (let i = 0; i < messages.length; i++) { + const m = messages[i]; + out.push(m); + if (typeof m.content === 'string' || m.role !== 'assistant') continue; + + const calls = m.content.filter( + (b): b is Extract => b.type === 'tool-call', + ); + if (calls.length === 0) continue; + + // v6 only accepts results in the immediately-following message. + const next = messages[i + 1]; + const nextBlocks = next && typeof next.content !== 'string' ? next.content : []; + const resolved = new Set( + nextBlocks + .filter((b): b is Extract => b.type === 'tool-result') + .map((b) => b.toolCallId), + ); + + const missing = calls.filter((c) => !resolved.has(c.toolCallId)); + if (missing.length === 0) continue; + + const stubs: ChatBlock[] = missing.map((c) => ({ + type: 'tool-result', + toolCallId: c.toolCallId, + toolName: c.toolName, + output: 'tool result unavailable (recovered after interrupted run)', + isError: true, + })); + + if (resolved.size > 0) { + // A tool-result message follows but is incomplete — merge the stubs in. + out.push({ role: next!.role, content: [...(nextBlocks as ChatBlock[]), ...stubs] }); + i++; // the merged message replaces the original; don't emit it twice. + } else { + // No following tool-result message at all — synthesize one. + out.push({ role: 'user', content: stubs }); + } + } + return out; +} + export interface ChatResult { /** Final text content concatenated from text blocks. */ text: string; @@ -2697,7 +2812,7 @@ export async function chat(opts: ChatOpts): Promise { } } const estimatedInputTokens = estimateChatInputTokens(opts); - const maxOutputTokens = opts.maxTokens ?? 4096; + const maxOutputTokens = opts.maxTokens ?? defaultMaxOutputTokens(modelStrEarly); // TX5: reserve BEFORE the provider call. Throws BudgetExhausted on cost, // runtime, or no_pricing (when cap is set). Pre-resolution model id is @@ -2804,9 +2919,9 @@ export async function chat(opts: ChatOpts): Promise { const result = await generateText({ model, system: opts.system, - messages: toModelMessages(opts.messages) as any, + messages: toModelMessages(repairToolPairing(opts.messages)) as any, tools: opts.tools && opts.tools.length > 0 ? tools : undefined, - maxOutputTokens: opts.maxTokens ?? 4096, + maxOutputTokens: opts.maxTokens ?? defaultMaxOutputTokens(modelStr), // v0.42.20.0 — default a chat timeout (composes with the caller's signal, // shorter wins). Covers native-anthropic (the default provider + facts Haiku). abortSignal: withDefaultTimeout(opts.abortSignal, AI_CHAT_TIMEOUT_MS), @@ -2957,6 +3072,14 @@ export interface ToolLoopOpts { ) => Promise<{ gbrainToolUseId: string }>; onToolCallComplete?: (gbrainToolUseId: string, output: unknown) => Promise; onToolCallFailed?: (gbrainToolUseId: string, error: string) => Promise; + /** + * Persist the tool-result user turn that closes each tool round, BEFORE it is + * appended to the in-memory history. Without this the loop only kept the + * tool-result turn in memory, so a resumed job reloaded assistant tool-calls + * with no matching results and non-Anthropic providers rejected the + * unbalanced history (AI_MissingToolResultsError). Fires per completed round. + */ + onToolResultTurn?: (turnIdx: number, messageIdx: number, blocks: ChatBlock[]) => Promise; /** Optional per-call heartbeat for observability. */ onHeartbeat?: (event: string, data: Record) => void; @@ -2991,7 +3114,7 @@ export interface ToolLoopResult { */ export async function toolLoop(opts: ToolLoopOpts): Promise { const maxTurns = opts.maxTurns ?? 20; - const maxTokens = opts.maxTokens ?? 4096; + const maxTokens = opts.maxTokens ?? defaultMaxOutputTokens(opts.model ?? getChatModel()); const handlers = opts.toolHandlers; const totalUsage: ChatResult['usage'] = { input_tokens: 0, @@ -3180,9 +3303,11 @@ export async function toolLoop(opts: ToolLoopOpts): Promise { if (stopReason === 'aborted') break; - // Feed all tool results back as a single user message. + // Persist + feed all tool results back as a single user message. The + // persist-before-push mirrors onAssistantTurn's write-ordering: a crash + // after this leaves a balanced transcript for the next resume. const userMessageIdx = messageIdx++; - void userMessageIdx; + await opts.onToolResultTurn?.(turnIdx, userMessageIdx, toolResultBlocks); messages.push({ role: 'user', content: toolResultBlocks }); turnIdx++; diff --git a/src/core/ai/recipes/deepseek.ts b/src/core/ai/recipes/deepseek.ts index daa7cf2e9..d2dd4971d 100644 --- a/src/core/ai/recipes/deepseek.ts +++ b/src/core/ai/recipes/deepseek.ts @@ -1,5 +1,64 @@ import type { Recipe } from '../types.ts'; +/** + * `deepseek-reasoner` returns its answer in a separate `reasoning_content` + * field and leaves `content` empty/whitespace when the whole response was + * reasoning. The AI SDK's openai-compatible adapter reads only `content`, so + * the model appears to answer with nothing. This transport shim promotes + * `reasoning_content` into `content` when `content` is empty, before the + * adapter parses the body. Fail-open: any error returns the original response. + * Non-streaming JSON chat completions only. + * + * @internal exported for tests. + */ +// Cast through `unknown` because TS's `typeof fetch` includes a `preconnect` +// member the arrow function does not implement (matches azure-openai.ts). +export const deepseekReasoningContentCompatFetch = (async ( + input: RequestInfo | URL, + init?: RequestInit, +): Promise => { + const res = await fetch(input as any, init as any); + try { + if (!res.ok) return res; + const ctype = res.headers.get('content-type') ?? ''; + if (!ctype.includes('application/json')) return res; + const json = await res.clone().json(); + const choices = Array.isArray(json?.choices) ? json.choices : []; + let modified = false; + for (const choice of choices) { + const msg = choice?.message; + if (!msg) continue; + // A tool-call turn legitimately carries content:null — the answer is the + // tool call, not text. NEVER promote reasoning_content there: DeepSeek's + // chain-of-thought must not be fed back to the model (it would be + // persisted as assistant text and replayed every subsequent turn, + // contaminating context and inflating tokens). Only promote on a terminal + // text turn whose content is empty. + const hasToolCalls = Array.isArray(msg.tool_calls) && msg.tool_calls.length > 0; + const content = msg.content; + const reasoning = msg.reasoning_content; + const contentEmpty = content == null || (typeof content === 'string' && content.trim() === ''); + if (!hasToolCalls && contentEmpty && typeof reasoning === 'string' && reasoning.trim() !== '') { + msg.content = reasoning; + modified = true; + } + } + if (!modified) return res; + // Rebuild with a fresh header set: the body length changed, so the + // upstream content-length / content-encoding would now be wrong. + const headers = new Headers(res.headers); + headers.delete('content-length'); + headers.delete('content-encoding'); + return new Response(JSON.stringify(json), { + status: res.status, + statusText: res.statusText, + headers, + }); + } catch { + return res; + } +}) as unknown as typeof fetch; + /** * DeepSeek exposes an OpenAI-compatible /v1/chat/completions endpoint. * Useful as the second hop in a refusal-fallback chain and for cheap- @@ -29,4 +88,5 @@ export const deepseek: Recipe = { }, }, setup_hint: 'Get an API key at https://platform.deepseek.com/api_keys, then `export DEEPSEEK_API_KEY=...`', + compat: { fetch: deepseekReasoningContentCompatFetch }, }; diff --git a/src/core/ai/types.ts b/src/core/ai/types.ts index 4dced5c6d..8835c838f 100644 --- a/src/core/ai/types.ts +++ b/src/core/ai/types.ts @@ -326,6 +326,18 @@ export interface Recipe { baseURL: string; fetch?: typeof fetch; }; + /** + * Optional inbound-response rewriter for openai-compatible recipes whose wire + * shape needs normalizing before the AI SDK adapter parses it. `fetch` wraps + * the transport and MUST be fail-open (return the original response on any + * error). Used by DeepSeek to promote `reasoning_content` into `content` when + * the reasoner returns an empty `content` (the adapter reads only `content`). + * Applied by `applyOpenAICompatConfig`; a `resolveOpenAICompatConfig`-provided + * fetch takes precedence when both are present. + */ + compat?: { + fetch?: typeof fetch; + }; /** * v0.32 (D13=A): optional runtime readiness check for local-server * recipes (ollama, llama-server, future lmstudio-recipe). Returns diff --git a/src/core/config.ts b/src/core/config.ts index 8a79b0e06..aa05cdd62 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -41,6 +41,13 @@ export interface GBrainConfig { * merge → buildGatewayConfig env dict → recipe reads ZEROENTROPY_API_KEY. */ zeroentropy_api_key?: string; + /** + * OpenRouter API key. File-plane slot so `gbrain config set + * openrouter_api_key X` (or config.json) reaches the openrouter recipe: + * file plane → loadConfig env merge → buildGatewayConfig env dict → recipe + * reads OPENROUTER_API_KEY. + */ + openrouter_api_key?: string; /** AI gateway config (v0.14+). v0.36+ default: "zeroentropyai:zembed-1" / 1280 / "anthropic:claude-haiku-4-5-20251001". */ embedding_model?: string; embedding_dimensions?: number; @@ -526,6 +533,7 @@ export function loadConfig(): GBrainConfig | null { ...(process.env.OPENAI_API_KEY ? { openai_api_key: process.env.OPENAI_API_KEY } : {}), ...(process.env.ANTHROPIC_API_KEY ? { anthropic_api_key: process.env.ANTHROPIC_API_KEY } : {}), ...(process.env.ZEROENTROPY_API_KEY ? { zeroentropy_api_key: process.env.ZEROENTROPY_API_KEY } : {}), + ...(process.env.OPENROUTER_API_KEY ? { openrouter_api_key: process.env.OPENROUTER_API_KEY } : {}), ...(process.env.GBRAIN_EMBEDDING_MODEL ? { embedding_model: process.env.GBRAIN_EMBEDDING_MODEL } : {}), ...(process.env.GBRAIN_EMBEDDING_DIMENSIONS ? { embedding_dimensions: parseInt(process.env.GBRAIN_EMBEDDING_DIMENSIONS, 10) } : {}), ...(process.env.GBRAIN_EXPANSION_MODEL ? { expansion_model: process.env.GBRAIN_EXPANSION_MODEL } : {}), @@ -815,6 +823,8 @@ export const KNOWN_CONFIG_KEYS: readonly string[] = [ 'database_path', 'openai_api_key', 'anthropic_api_key', + 'zeroentropy_api_key', + 'openrouter_api_key', 'embedding_model', 'embedding_dimensions', 'embedding_disabled', @@ -836,6 +846,11 @@ export const KNOWN_CONFIG_KEYS: readonly string[] = [ 'sync', 'sync.repo_path', 'sync.last_commit', + // Gateway-native subagent loop toggle (routes subagent jobs through the + // provider-agnostic gateway.toolLoop for non-Anthropic providers). The + // subagent handler's error message tells users to `config set` this, so it + // must be a known key or `config set` rejects it without --force. + 'agent.use_gateway_loop', // DB-plane (v0.32.3 search modes + related) 'search.mode', 'search.cache.enabled', diff --git a/src/core/minions/handlers/subagent.ts b/src/core/minions/handlers/subagent.ts index 450dc8da0..f0b1ec251 100644 --- a/src/core/minions/handlers/subagent.ts +++ b/src/core/minions/handlers/subagent.ts @@ -777,22 +777,57 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise ({ - role: m.role as 'user' | 'assistant', - content: adaptContentBlocksToChatBlocks(m.content_blocks), - })); + // Token rollup across the prior transcript (returned as-is on the terminal + // early-return path; the loop adds only NEW-turn usage otherwise). + const priorTokens = { in: 0, out: 0, cache_read: 0, cache_create: 0 }; + for (const m of priorMessages) { + if (m.tokens_in) priorTokens.in += m.tokens_in; + if (m.tokens_out) priorTokens.out += m.tokens_out; + if (m.tokens_cache_read) priorTokens.cache_read += m.tokens_cache_read; + if (m.tokens_cache_create) priorTokens.cache_create += m.tokens_cache_create; + } + + // Reconcile an unbalanced transcript from a prior crashed/resumed run. The + // gateway loop persists each assistant turn but (pre-fix) never persisted the + // following tool-result user turn, so a resumed job reloads assistant + // tool-calls with no matching results — which non-Anthropic (openai-compat) + // providers reject with AI_MissingToolResultsError, dead-lettering the job. + // reconcileGatewayReplay heals every such dangling turn from settled tool + // executions (mirroring the legacy Anthropic path) and reports the terminal + // case where the prior run already reached end_turn. + const priorToolsV1 = await loadPriorTools(engine, ctx.id); + const { chatMessages: priorChatMessages, nextMessageIdx: reconciledNextIdx, terminalText } = + await reconcileGatewayReplay({ + engine, + jobId: ctx.id, + priorMessages, + priorTools: priorToolsV1, + toolDefs, + signal: ctx.signal, + }); + + // Terminal early-return (#1151 parity): the prior run already reached + // end_turn. Non-Anthropic providers reject a trailing assistant "prefill", + // so surface the persisted text and skip the loop entirely. + if (terminalText !== null) { + return { + result: terminalText, + turns_count: priorChatMessages.filter(m => m.role === 'assistant').length, + stop_reason: 'end_turn', + tokens: priorTokens, + }; + } // Initial seed message if no prior state. const initialMessages: ChatMessage[] = priorChatMessages.length === 0 ? [{ role: 'user', content: data.prompt }] : []; - // Persist seed user message at idx 0 if fresh start. - let nextMessageIdx = priorChatMessages.length; - if (nextMessageIdx === 0) { + // Persist seed user message at idx 0 if fresh start. reconciledNextIdx is + // max(known message_idx) + 1 (0 when no prior rows), which keeps the loop's + // subsequent writes clear of any healed tool-result turn we just inserted. + let nextMessageIdx = reconciledNextIdx; + if (priorChatMessages.length === 0) { await persistMessage(engine, ctx.id, { message_idx: 0, role: 'user', @@ -904,6 +939,22 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise { + await persistMessage(engine, ctx.id, { + message_idx: messageIdx, + role: 'user', + content_blocks: blocks as unknown as ContentBlock[], + tokens_in: null, + tokens_out: null, + tokens_cache_read: null, + tokens_cache_create: null, + model: null, + }); + }, onHeartbeat: heartbeat, }); @@ -934,6 +985,158 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise { + const { engine, jobId, priorMessages, priorTools, toolDefs, signal } = args; + + const work = priorMessages.map(m => { + const adapted = adaptContentBlocksToChatBlocks(m.content_blocks); + return { + message_idx: m.message_idx, + role: m.role, + blocks: typeof adapted === 'string' ? [{ type: 'text', text: adapted } as ChatBlock] : adapted, + }; + }); + + // Settled executions, looked up by (assistant message_idx, provider + // tool_use_id) with an ordinal-position fallback for legacy rows. + const execByKey = new Map(); + const execByMsg = new Map(); + for (const t of priorTools) { + execByKey.set(`${t.message_idx}:${t.tool_use_id}`, t); + const arr = execByMsg.get(t.message_idx) ?? []; + arr.push(t); + execByMsg.set(t.message_idx, arr); + } + + let maxIdx = work.reduce((mx, w) => Math.max(mx, w.message_idx), -1); + + for (let i = 0; i < work.length; i++) { + const msg = work[i]; + if (msg.role !== 'assistant') continue; + const toolCalls = msg.blocks.filter( + (b): b is Extract => b.type === 'tool-call', + ); + if (toolCalls.length === 0) continue; + + // Skip if a following tool-result user turn exists AT ALL. A fully-answered + // turn is already balanced; a PARTIALLY-answered turn (only reachable from + // externally-corrupted data — gbrain persists all of a turn's results in one + // message) is left for repairToolPairing() at the chat() boundary, which + // back-fills only the missing ids. Synthesizing a full turn here would + // duplicate the answered results and collide with the persisted row. + const next = work[i + 1]; + if (next && next.role === 'user' && next.blocks.some(b => b.type === 'tool-result')) continue; + + const results: ChatBlock[] = []; + for (let callIdx = 0; callIdx < toolCalls.length; callIdx++) { + const call = toolCalls[callIdx]; + // Prefer an exact (message_idx, provider tool_use_id) match. The + // positional fallback is used only when the row at that ordinal is for + // the SAME tool, so a missing row can't mis-attribute a sibling's output. + const fallback = execByMsg.get(msg.message_idx)?.[callIdx]; + const exec = execByKey.get(`${msg.message_idx}:${call.toolCallId}`) + ?? (fallback && fallback.tool_name === call.toolName ? fallback : undefined); + if (exec?.status === 'complete') { + results.push({ type: 'tool-result', toolCallId: call.toolCallId, toolName: call.toolName, output: exec.output ?? null }); + continue; + } + if (exec?.status === 'failed') { + results.push({ type: 'tool-result', toolCallId: call.toolCallId, toolName: call.toolName, output: exec.error ?? 'tool failed', isError: true }); + continue; + } + const toolDef = toolDefs.find(t => t.name === call.toolName); + if (!toolDef) { + await persistToolExecFailed(engine, jobId, msg.message_idx, call.toolCallId, call.toolName, call.input, `tool "${call.toolName}" is not in the registry for this subagent`); + results.push({ type: 'tool-result', toolCallId: call.toolCallId, toolName: call.toolName, output: `tool "${call.toolName}" is not available`, isError: true }); + continue; + } + if (exec?.status === 'pending' && !toolDef.idempotent) { + throw new Error(`non-idempotent tool "${call.toolName}" pending on resume; cannot safely re-run`); + } + await persistToolExecPending(engine, jobId, msg.message_idx, call.toolCallId, call.toolName, call.input); + try { + const output = await toolDef.execute(call.input, { engine, jobId, remote: true, signal }); + await persistToolExecComplete(engine, jobId, call.toolCallId, output); + results.push({ type: 'tool-result', toolCallId: call.toolCallId, toolName: call.toolName, output }); + } catch (e) { + const errText = e instanceof Error ? (e.stack ?? e.message) : String(e); + await persistToolExecFailed(engine, jobId, msg.message_idx, call.toolCallId, call.toolName, call.input, errText); + results.push({ type: 'tool-result', toolCallId: call.toolCallId, toolName: call.toolName, output: errText, isError: true }); + } + } + + const resultIdx = msg.message_idx + 1; + await persistMessage(engine, jobId, { + message_idx: resultIdx, + role: 'user', + content_blocks: results as unknown as ContentBlock[], + tokens_in: null, tokens_out: null, tokens_cache_read: null, tokens_cache_create: null, model: null, + }); + maxIdx = Math.max(maxIdx, resultIdx); + work.splice(i + 1, 0, { message_idx: resultIdx, role: 'user' as const, blocks: results }); + i++; // skip the turn we just inserted + } + + const chatMessages: ChatMessage[] = work.map(w => ({ role: w.role, content: w.blocks })); + + // Terminal case: the transcript already ends on an assistant turn that + // carried real text and made no tool calls (prior run reached end_turn). + // Surface its text; skip the loop. An assistant turn whose blocks are all + // empty (e.g. a reasoning-only/null-text turn that adaptation dropped) is NOT + // terminal — falling through lets the loop re-issue the call rather than + // returning an empty result. + const lastMsg = work[work.length - 1]; + let terminalText: string | null = null; + if (lastMsg && lastMsg.role === 'assistant' && !lastMsg.blocks.some(b => b.type === 'tool-call')) { + const text = lastMsg.blocks + .filter((b): b is Extract => b.type === 'text') + .map(b => b.text) + .join('\n'); + if (text.trim() !== '') terminalText = text; + } + + return { chatMessages, nextMessageIdx: maxIdx + 1, terminalText }; +} + function recipeIdFromModel(modelString: string): string { const idx = modelString.indexOf(':'); return idx > 0 ? modelString.slice(0, idx) : 'anthropic'; @@ -1087,7 +1290,8 @@ async function loadPriorTools(engine: BrainEngine, jobId: number): Promise>( `SELECT message_idx, tool_use_id, tool_name, input, status, output, error FROM subagent_tool_executions - WHERE job_id = $1`, + WHERE job_id = $1 + ORDER BY message_idx, COALESCE(ordinal, 0), id`, [jobId], ); return rows.map(r => ({ diff --git a/src/core/think/index.ts b/src/core/think/index.ts index 8f3ab94cc..9c31c81dc 100644 --- a/src/core/think/index.ts +++ b/src/core/think/index.ts @@ -152,6 +152,19 @@ export interface ThinkResult { const DEFAULT_MAX_OUTPUT_TOKENS = 4000; +// Thinking-by-default Claude 5 models (`anthropic:claude-*-5`) spend a large +// share of the output budget on internal reasoning before emitting any answer, +// so the 4000 default leaves `think` with empty or truncated text. Give those +// models headroom; providers bill actual tokens, not the cap. Everything else +// keeps 4000. +const THINKING_DEFAULT_MAX_OUTPUT_TOKENS = 16000; +const THINKING_BY_DEFAULT_MODEL_RE = /^anthropic[:/]claude-[a-z0-9]+-5(?:[.-]|$)/i; +export function maxOutputTokensFor(modelStr: string): number { + return THINKING_BY_DEFAULT_MODEL_RE.test(modelStr) + ? THINKING_DEFAULT_MAX_OUTPUT_TOKENS + : DEFAULT_MAX_OUTPUT_TOKENS; +} + function inferIntent(question: string, anchor?: string): string { if (anchor) return 'entity'; const q = question.toLowerCase(); @@ -465,7 +478,7 @@ export async function runThink( } const result = await client.create({ model: modelUsed, - max_tokens: DEFAULT_MAX_OUTPUT_TOKENS, + max_tokens: maxOutputTokensFor(normalizeModelId(modelUsed)), system: systemPrompt, messages: [{ role: 'user', content: userMessage }], }); diff --git a/test/ai/build-gateway-config.test.ts b/test/ai/build-gateway-config.test.ts index 5284c5bc8..b9ebb27d7 100644 --- a/test/ai/build-gateway-config.test.ts +++ b/test/ai/build-gateway-config.test.ts @@ -86,6 +86,26 @@ describe('buildGatewayConfig env-baseURL passthrough', () => { }); }); +describe('buildGatewayConfig config-plane API-key folding', () => { + test('openrouter_api_key folds into gateway env as OPENROUTER_API_KEY', async () => { + await withEnv({ OPENROUTER_API_KEY: undefined }, async () => { + const cfg = buildGatewayConfig({ + openrouter_api_key: 'sk-or-config-plane', + } as unknown as GBrainConfig); + expect(cfg.env.OPENROUTER_API_KEY).toBe('sk-or-config-plane'); + }); + }); + + test('a real OPENROUTER_API_KEY process.env value wins over the config-plane fallback', async () => { + await withEnv({ OPENROUTER_API_KEY: 'sk-or-env-plane' }, async () => { + const cfg = buildGatewayConfig({ + openrouter_api_key: 'sk-or-config-plane', + } as unknown as GBrainConfig); + expect(cfg.env.OPENROUTER_API_KEY).toBe('sk-or-env-plane'); + }); + }); +}); + describe('buildGatewayConfig env empty-string clobber guard (#1249)', () => { test('an empty-string process.env value does NOT clobber a valid config-plane key', async () => { // Claude Code injects ANTHROPIC_API_KEY='' to neuter subprocess LLM calls. diff --git a/test/ai/deepseek-reasoning-content.test.ts b/test/ai/deepseek-reasoning-content.test.ts new file mode 100644 index 000000000..8c1d6481b --- /dev/null +++ b/test/ai/deepseek-reasoning-content.test.ts @@ -0,0 +1,124 @@ +/** + * Pins the DeepSeek reasoning_content transport shim. `deepseek-reasoner` + * returns its answer in a separate `reasoning_content` field and leaves + * `content` empty when the whole turn was reasoning; the AI SDK's + * openai-compatible adapter reads only `content`, so the model appears to + * answer with nothing. The shim promotes `reasoning_content` into `content` + * when `content` is empty, fail-open on anything unexpected. + */ +import { describe, test, expect, afterEach } from 'bun:test'; +import { deepseekReasoningContentCompatFetch, deepseek } from '../../src/core/ai/recipes/deepseek.ts'; +import { applyOpenAICompatConfig } from '../../src/core/ai/gateway.ts'; +import type { Recipe, AIGatewayConfig } from '../../src/core/ai/types.ts'; + +const realFetch = globalThis.fetch; +afterEach(() => { globalThis.fetch = realFetch; }); + +function stubFetch(body: unknown, init?: { status?: number; contentType?: string }) { + globalThis.fetch = (async () => + new Response(JSON.stringify(body), { + status: init?.status ?? 200, + headers: { 'content-type': init?.contentType ?? 'application/json' }, + })) as unknown as typeof fetch; +} + +describe('deepseekReasoningContentCompatFetch', () => { + test('promotes reasoning_content when content is empty', async () => { + stubFetch({ choices: [{ message: { role: 'assistant', content: '', reasoning_content: 'the answer' } }] }); + const res = await deepseekReasoningContentCompatFetch('https://api.deepseek.com/v1/chat/completions'); + const json = await res.json(); + expect(json.choices[0].message.content).toBe('the answer'); + }); + + test('promotes when content is null or whitespace-only', async () => { + stubFetch({ choices: [{ message: { content: null, reasoning_content: 'from null' } }, { message: { content: ' ', reasoning_content: 'from ws' } }] }); + const res = await deepseekReasoningContentCompatFetch('u'); + const json = await res.json(); + expect(json.choices[0].message.content).toBe('from null'); + expect(json.choices[1].message.content).toBe('from ws'); + }); + + test('leaves non-empty content untouched (no duplication)', async () => { + stubFetch({ choices: [{ message: { content: 'real content', reasoning_content: 'ignored' } }] }); + const res = await deepseekReasoningContentCompatFetch('u'); + const json = await res.json(); + expect(json.choices[0].message.content).toBe('real content'); + }); + + test('both empty: stays empty, no crash', async () => { + stubFetch({ choices: [{ message: { content: '', reasoning_content: '' } }] }); + const res = await deepseekReasoningContentCompatFetch('u'); + const json = await res.json(); + expect(json.choices[0].message.content).toBe(''); + }); + + test('tool-call turn (content:null + tool_calls) is NOT promoted — never feed CoT back', async () => { + // content:null is the standard OpenAI shape on a tool-call turn. Promoting + // reasoning_content here would inject the whole chain-of-thought as assistant + // text, which the loop persists + replays every turn. Must be left alone. + stubFetch({ choices: [{ finish_reason: 'tool_calls', message: { + content: null, + reasoning_content: 'INTERNAL CHAIN OF THOUGHT — must not leak', + tool_calls: [{ id: 'call_1', type: 'function', function: { name: 'brain_search', arguments: '{}' } }], + } }] }); + const res = await deepseekReasoningContentCompatFetch('u'); + const json = await res.json(); + expect(json.choices[0].message.content).toBeNull(); + expect(json.choices[0].message.tool_calls).toHaveLength(1); + }); + + test('rebuilt response drops stale content-length header', async () => { + globalThis.fetch = (async () => new Response(JSON.stringify({ choices: [{ message: { content: '', reasoning_content: 'x' } }] }), { + status: 200, headers: { 'content-type': 'application/json', 'content-length': '999999' }, + })) as unknown as typeof fetch; + const res = await deepseekReasoningContentCompatFetch('u'); + expect(res.headers.get('content-length')).toBeNull(); + expect((await res.json()).choices[0].message.content).toBe('x'); + }); +}); + +describe('applyOpenAICompatConfig — compat.fetch wiring (gateway seam)', () => { + const cfg = { env: {}, base_urls: {} } as unknown as AIGatewayConfig; + + test('threads recipe.compat.fetch onto the resolved config (deepseek)', () => { + // Guards the src/core/ai/gateway.ts wiring: without `?? recipe.compat?.fetch` + // the DeepSeek shim would never install in production. + const resolved = applyOpenAICompatConfig(deepseek, cfg); + expect(resolved.fetch).toBe(deepseekReasoningContentCompatFetch); + }); + + test('a resolveOpenAICompatConfig-provided fetch takes precedence over compat.fetch', () => { + const ownFetch = (async () => new Response('{}')) as unknown as typeof fetch; + const recipe = { + id: 'x', name: 'X', tier: 'openai-compat', implementation: 'openai-compatible', + touchpoints: {}, + compat: { fetch: deepseekReasoningContentCompatFetch }, + resolveOpenAICompatConfig: () => ({ baseURL: 'http://x', fetch: ownFetch }), + } as unknown as Recipe; + expect(applyOpenAICompatConfig(recipe, cfg).fetch).toBe(ownFetch); + }); + + test('falls back to compat.fetch when resolveOpenAICompatConfig omits a fetch', () => { + const recipe = { + id: 'y', name: 'Y', tier: 'openai-compat', implementation: 'openai-compatible', + touchpoints: {}, + compat: { fetch: deepseekReasoningContentCompatFetch }, + resolveOpenAICompatConfig: () => ({ baseURL: 'http://y' }), + } as unknown as Recipe; + expect(applyOpenAICompatConfig(recipe, cfg).fetch).toBe(deepseekReasoningContentCompatFetch); + }); + + test('fail-open on non-ok / non-json responses', async () => { + stubFetch({ error: 'nope' }, { status: 500 }); + const res = await deepseekReasoningContentCompatFetch('u'); + expect(res.status).toBe(500); + globalThis.fetch = (async () => + new Response('plain text', { status: 200, headers: { 'content-type': 'text/plain' } })) as unknown as typeof fetch; + const res2 = await deepseekReasoningContentCompatFetch('u'); + expect(await res2.text()).toBe('plain text'); + }); + + test('recipe wires the shim via compat.fetch', () => { + expect(deepseek.compat?.fetch).toBe(deepseekReasoningContentCompatFetch); + }); +}); diff --git a/test/ai/gateway-tool-loop.test.ts b/test/ai/gateway-tool-loop.test.ts index b9700c379..4e6fb60c0 100644 --- a/test/ai/gateway-tool-loop.test.ts +++ b/test/ai/gateway-tool-loop.test.ts @@ -150,6 +150,47 @@ describe('gateway.toolLoop (v0.38 D11 — provider-agnostic loop control)', () = expect(events[4]).toBe('onAssistantTurn(1)'); // final assistant turn }); + it('persists the tool-result user turn via onToolResultTurn before the next chat', async () => { + let turn = 0; + __setChatTransportForTests(async () => { + turn++; + if (turn === 1) { + return { + text: '', + blocks: [{ type: 'tool-call', toolCallId: 'tc1', toolName: 'search', input: { q: 'x' } }] as ChatBlock[], + stopReason: 'tool_calls', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-sonnet-4-6', + providerId: 'anthropic', + }; + } + return { + text: 'done', + blocks: [{ type: 'text', text: 'done' }] as ChatBlock[], + stopReason: 'end', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-sonnet-4-6', + providerId: 'anthropic', + }; + }); + + const resultTurns: Array<{ turnIdx: number; messageIdx: number; blocks: ChatBlock[] }> = []; + await toolLoop({ + initialMessages: [{ role: 'user', content: 'go' }], + tools: [{ name: 'search', description: 's', inputSchema: { type: 'object' } }], + toolHandlers: new Map([['search', { idempotent: true, async execute() { return { hits: 1 }; } }]]), + onToolResultTurn: async (turnIdx, messageIdx, blocks) => { + resultTurns.push({ turnIdx, messageIdx, blocks }); + }, + }); + + // Fired exactly once, for the single tool round, carrying the tool-result. + expect(resultTurns).toHaveLength(1); + expect(resultTurns[0].turnIdx).toBe(0); + expect(resultTurns[0].blocks[0].type).toBe('tool-result'); + expect((resultTurns[0].blocks[0] as Extract).toolCallId).toBe('tc1'); + }); + it('replay short-circuits a complete prior tool execution', async () => { let chatCalls = 0; __setChatTransportForTests(async () => { @@ -230,6 +271,30 @@ describe('gateway.toolLoop (v0.38 D11 — provider-agnostic loop control)', () = ).rejects.toThrow(/non-idempotent.*pending/i); }); + it('defaults max output tokens per model: 4096 for non-thinking, 32000 for Claude 5', async () => { + const seen: Array = []; + __setChatTransportForTests(async (opts) => { + seen.push(opts.maxTokens); + return { + text: 'ok', + blocks: [{ type: 'text', text: 'ok' }] as ChatBlock[], + stopReason: 'end', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: opts.model ?? 'anthropic:claude-sonnet-4-6', + providerId: 'anthropic', + }; + }); + + await toolLoop({ model: 'openai:gpt-4o', initialMessages: [{ role: 'user', content: 'hi' }], tools: [], toolHandlers: new Map() }); + await toolLoop({ model: 'anthropic:claude-sonnet-4-6', initialMessages: [{ role: 'user', content: 'hi' }], tools: [], toolHandlers: new Map() }); + await toolLoop({ model: 'anthropic:claude-sonnet-5', initialMessages: [{ role: 'user', content: 'hi' }], tools: [], toolHandlers: new Map() }); + await toolLoop({ model: 'anthropic:claude-fable-5', initialMessages: [{ role: 'user', content: 'hi' }], tools: [], toolHandlers: new Map() }); + + // Non-thinking / non-Claude-5 stay 4096 (safe under openai-compat caps); + // thinking-by-default Claude 5 models get 32000 headroom. + expect(seen).toEqual([4096, 4096, 32000, 32000]); + }); + it('hits max_turns when the model keeps calling tools', async () => { __setChatTransportForTests(async () => ({ text: '', diff --git a/test/ai/gateway-toolcall-pairing.test.ts b/test/ai/gateway-toolcall-pairing.test.ts new file mode 100644 index 000000000..03681df60 --- /dev/null +++ b/test/ai/gateway-toolcall-pairing.test.ts @@ -0,0 +1,100 @@ +/** + * Pins `repairToolPairing` (the chat()-boundary safety net) and proves, against + * the REAL AI SDK v6 `generateText`, that the two failure modes this wave fixes + * are gone: + * - an unbalanced tool history (assistant tool-call with no tool-result) is + * back-filled so v6 no longer throws AI_MissingToolResultsError, and + * - a Date-bearing tool-result (Postgres timestamptz) passes v6's ModelMessage + * JSONValue schema after `toModelMessages` ISO-izes it, whereas a raw Date + * is still rejected (control). + * + * MockLanguageModelV3 = no network / no keys. + */ +import { describe, expect, it } from 'bun:test'; +import { generateText } from 'ai'; +import { MockLanguageModelV3 } from 'ai/test'; +import { repairToolPairing, toModelMessages, type ChatMessage } from '../../src/core/ai/gateway.ts'; + +function mockModel(): MockLanguageModelV3 { + return new MockLanguageModelV3({ + doGenerate: async () => ({ + content: [{ type: 'text', text: 'ok' }], + finishReason: 'stop', + usage: { inputTokens: 1, outputTokens: 1, totalTokens: 2 }, + warnings: [], + }), + } as any); +} + +describe('repairToolPairing', () => { + it('is a no-op on a balanced history', () => { + const msgs: ChatMessage[] = [ + { role: 'user', content: 'hi' }, + { role: 'assistant', content: [{ type: 'tool-call', toolCallId: 'c1', toolName: 'search', input: {} }] }, + { role: 'user', content: [{ type: 'tool-result', toolCallId: 'c1', toolName: 'search', output: { ok: 1 } }] }, + ]; + expect(repairToolPairing(msgs)).toEqual(msgs); + }); + + it('synthesizes a tool-result turn when the assistant tool-call is fully unanswered', () => { + const msgs: ChatMessage[] = [ + { role: 'assistant', content: [{ type: 'tool-call', toolCallId: 'c1', toolName: 'search', input: {} }] }, + ]; + const out = repairToolPairing(msgs); + expect(out).toHaveLength(2); + expect(out[1].role).toBe('user'); + const block = (out[1].content as any[])[0]; + expect(block).toMatchObject({ type: 'tool-result', toolCallId: 'c1', isError: true }); + }); + + it('merges stubs into a PARTIALLY-answered turn without duplicating the answered id', () => { + const msgs: ChatMessage[] = [ + { role: 'assistant', content: [ + { type: 'tool-call', toolCallId: 'a', toolName: 'search', input: {} }, + { type: 'tool-call', toolCallId: 'b', toolName: 'search', input: {} }, + ] }, + { role: 'user', content: [{ type: 'tool-result', toolCallId: 'a', toolName: 'search', output: { ok: 1 } }] }, + ]; + const out = repairToolPairing(msgs); + expect(out).toHaveLength(2); // merged in place, not appended + const ids = (out[1].content as any[]).map((x) => x.toolCallId); + expect(ids).toEqual(['a', 'b']); // 'a' kept once, 'b' back-filled + expect((out[1].content as any[]).filter((x) => x.toolCallId === 'a')).toHaveLength(1); + }); +}); + +describe('real AI SDK v6 validation', () => { + it('an unbalanced history passes generateText after repairToolPairing', async () => { + const model = mockModel(); + const unbalanced: ChatMessage[] = [ + { role: 'user', content: 'go' }, + { role: 'assistant', content: [{ type: 'tool-call', toolCallId: 'c1', toolName: 'search', input: {} }] }, + // no tool-result turn + ]; + const result = await generateText({ + model: model as any, + messages: toModelMessages(repairToolPairing(unbalanced)) as any, + }); + expect(result.text).toBe('ok'); + const prompt = model.doGenerateCalls[0]!.prompt as any[]; + expect(prompt.some((m) => m.role === 'tool')).toBe(true); // stub promoted to tool role + }); + + it('a Date-bearing tool-result passes after toModelMessages ISO-izes it; a raw Date is rejected', async () => { + const withDate: ChatMessage[] = [ + { role: 'user', content: 'go' }, + { role: 'assistant', content: [{ type: 'tool-call', toolCallId: 'c1', toolName: 'brain_get_page', input: {} }] }, + { role: 'user', content: [{ type: 'tool-result', toolCallId: 'c1', toolName: 'brain_get_page', output: { updated_at: new Date('2026-06-26T06:56:59.000Z') } }] }, + ]; + // Fixed path: converted history validates. + await expect(generateText({ model: mockModel() as any, messages: toModelMessages(withDate) as any })).resolves.toBeDefined(); + + // Control: a raw Date placed straight into a v6 ModelMessage json value is rejected. + const rawDateMessages = [ + { role: 'user', content: 'go' }, + { role: 'assistant', content: [{ type: 'tool-call', toolCallId: 'c1', toolName: 'brain_get_page', input: {} }] }, + { role: 'tool', content: [{ type: 'tool-result', toolCallId: 'c1', toolName: 'brain_get_page', output: { type: 'json', value: { updated_at: new Date('2026-06-26T06:56:59.000Z') } } }] }, + ]; + await expect(generateText({ model: mockModel() as any, messages: rawDateMessages as any })).rejects.toThrow(); + }); +}); diff --git a/test/config-set.test.ts b/test/config-set.test.ts index 8d353c838..4c0c0b372 100644 --- a/test/config-set.test.ts +++ b/test/config-set.test.ts @@ -38,6 +38,15 @@ describe('KNOWN_CONFIG_KEYS', () => { expect(KNOWN_CONFIG_KEYS).toContain('embed.backfill_max_usd'); }); + test('includes the gateway-loop toggle and provider API keys the wave wires', () => { + // The subagent handler's error message tells users to run + // `gbrain config set agent.use_gateway_loop true`; it must be a known key + // or `config set` rejects the wave's own enable command without --force. + expect(KNOWN_CONFIG_KEYS).toContain('agent.use_gateway_loop'); + expect(KNOWN_CONFIG_KEYS).toContain('openrouter_api_key'); + expect(KNOWN_CONFIG_KEYS).toContain('zeroentropy_api_key'); + }); + test('no duplicate entries', () => { const set = new Set(KNOWN_CONFIG_KEYS); expect(set.size).toBe(KNOWN_CONFIG_KEYS.length); diff --git a/test/e2e/subagent-gateway-resume-reconciliation.test.ts b/test/e2e/subagent-gateway-resume-reconciliation.test.ts new file mode 100644 index 000000000..d7c479a3a --- /dev/null +++ b/test/e2e/subagent-gateway-resume-reconciliation.test.ts @@ -0,0 +1,280 @@ +/** + * E2E: gateway-loop resume reconciliation (fix-wave A). + * + * The gateway-native subagent loop persists each assistant turn but historically + * never persisted the following tool-result user turn. A resumed job therefore + * reloaded assistant tool-calls with no matching tool-result, and non-Anthropic + * (openai-compat) providers reject that unbalanced history with + * AI_MissingToolResultsError — dead-lettering the job. This wave: + * 1. forward-persists the tool-result user turn (onToolResultTurn), and + * 2. reconciles an already-corrupted transcript on resume by rebuilding the + * missing tool-result turns from settled subagent_tool_executions, + * re-dispatching idempotent-pending tools and throwing on non-idempotent. + * + * Hermetic: PGLite in-memory engine, gateway transport stubbed. Seeds use the + * sanctioned `$N::text::jsonb` positional bind (NEVER JSON.stringify into a + * bare `::jsonb`) so the seed is Postgres-safe too. + * + * Supersedes the resume/replay work in #1934 #2062 #2065 #2112 #2274 #2487 + * #2802 #2336 #2257 #2499. + */ + +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../../src/core/pglite-engine.ts'; +import { resetPgliteState } from '../helpers/reset-pglite.ts'; +import { makeSubagentHandler } from '../../src/core/minions/handlers/subagent.ts'; +import type { MinionJobContext, ToolDef, ToolCtx } from '../../src/core/minions/types.ts'; +import { + __setChatTransportForTests, + configureGateway, + resetGateway, + toModelMessages, + type ChatBlock, + type ChatMessage, + type ChatResult, +} from '../../src/core/ai/gateway.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}); +afterAll(async () => { + __setChatTransportForTests(null); + resetGateway(); + await engine.disconnect(); +}); +beforeEach(async () => { + await resetPgliteState(engine); + await engine.setConfig('version', '85'); + await engine.setConfig('agent.use_gateway_loop', 'true'); + configureGateway({ + chat_model: 'anthropic:claude-sonnet-4-6', + embedding_model: 'openai:text-embedding-3-large', + embedding_dimensions: 1536, + expansion_model: 'anthropic:claude-haiku-4-5', + env: { ANTHROPIC_API_KEY: 'stub', OPENAI_API_KEY: 'stub' }, + }); +}); + +async function makeJob(prompt: string, model: string): Promise<{ jobId: number; ctx: MinionJobContext }> { + const rows = await engine.executeRaw<{ id: number }>( + `INSERT INTO minion_jobs (name, status, data, queue, priority, created_at) + VALUES ('subagent', 'active', $1::text::jsonb, 'default', 0, now()) RETURNING id`, + [JSON.stringify({ prompt, model })], + ); + const jobId = rows[0].id; + const ctx: MinionJobContext = { + id: jobId, name: 'subagent', data: { prompt, model }, attempts_made: 1, + signal: new AbortController().signal, shutdownSignal: new AbortController().signal, + updateProgress: async () => {}, updateTokens: async () => {}, log: async () => {}, + isActive: async () => true, readInbox: async () => [], + }; + return { jobId, ctx }; +} + +function makeTools(executions: string[]): ToolDef[] { + return [ + { name: 'search', description: 's', input_schema: { type: 'object' }, idempotent: true, + async execute(input: unknown, _c: ToolCtx) { executions.push('search'); return { results: ['fresh'] }; } }, + { name: 'put_page', description: 'p', input_schema: { type: 'object' }, idempotent: false, + async execute(_input: unknown, _c: ToolCtx) { executions.push('put_page'); return { saved: true }; } }, + ]; +} + +function buildHandler(toolRegistry: ToolDef[]) { + return makeSubagentHandler({ + engine, config: {} as any, toolRegistry, + makeAnthropic: () => ({ messages: { create: async () => { throw new Error('legacy path unused'); } } }) as any, + }); +} + +async function seedMessage(jobId: number, idx: number, role: string, blocks: ChatBlock[]): Promise { + await engine.executeRaw( + `INSERT INTO subagent_messages (job_id, message_idx, role, content_blocks, schema_version) + VALUES ($1, $2, $3, $4::text::jsonb, 2)`, + [jobId, idx, role, JSON.stringify(blocks)], + ); +} + +async function seedExec(jobId: number, msgIdx: number, toolUseId: string, name: string, status: string, output: unknown, ordinal: number, error?: string): Promise { + await engine.executeRaw( + `INSERT INTO subagent_tool_executions + (job_id, message_idx, tool_use_id, tool_name, input, status, output, error, schema_version, ordinal) + VALUES ($1, $2, $3, $4, $5::text::jsonb, $6, $7::text::jsonb, $8, 2, $9)`, + [jobId, msgIdx, toolUseId, name, JSON.stringify({}), status, output == null ? null : JSON.stringify(output), error ?? null, ordinal], + ); +} + +/** Assert every assistant tool-call turn is answered by a following tool-result. */ +function assertBalanced(messages: ChatMessage[]): void { + for (let i = 0; i < messages.length; i++) { + const m = messages[i]; + if (m.role !== 'assistant' || typeof m.content === 'string') continue; + const calls = m.content.filter((b): b is Extract => b.type === 'tool-call'); + if (calls.length === 0) continue; + const next = messages[i + 1]; + expect(next, `assistant tool-call turn at ${i} must be followed by a tool-result turn`).toBeDefined(); + const answered = new Set( + (typeof next!.content === 'string' ? [] : next!.content) + .filter((b): b is Extract => b.type === 'tool-result') + .map(b => b.toolCallId), + ); + for (const c of calls) expect(answered.has(c.toolCallId), `tool-call ${c.toolCallId} unanswered`).toBe(true); + } + // The real provider path: this must not throw the ModelMessage schema error. + expect(() => toModelMessages(messages)).not.toThrow(); +} + +describe('gateway resume reconciliation', () => { + it('forward-persists the tool-result user turn (idx 2) in a 2-turn flow', async () => { + let turn = 0; + __setChatTransportForTests(async () => { + turn++; + if (turn === 1) return { + text: '', blocks: [{ type: 'tool-call', toolCallId: 'tc1', toolName: 'search', input: { q: 'x' } }] as ChatBlock[], + stopReason: 'tool_calls', usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-sonnet-4-6', providerId: 'anthropic', + } satisfies ChatResult; + return { + text: 'done', blocks: [{ type: 'text', text: 'done' }] as ChatBlock[], + stopReason: 'end', usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-sonnet-4-6', providerId: 'anthropic', + } satisfies ChatResult; + }); + const { jobId, ctx } = await makeJob('go', 'anthropic:claude-sonnet-4-6'); + await buildHandler(makeTools([]))(ctx); + + const msgs = await engine.executeRaw<{ message_idx: number; role: string; content_blocks: unknown }>( + `SELECT message_idx, role, content_blocks FROM subagent_messages WHERE job_id = $1 ORDER BY message_idx`, [jobId]); + expect(msgs.map(m => [m.message_idx, m.role])).toEqual([[0, 'user'], [1, 'assistant'], [2, 'user'], [3, 'assistant']]); + const toolResultTurn = typeof msgs[2].content_blocks === 'string' ? JSON.parse(msgs[2].content_blocks as string) : msgs[2].content_blocks; + expect((toolResultTurn as any[])[0].type).toBe('tool-result'); + expect((toolResultTurn as any[])[0].toolCallId).toBe('tc1'); + }); + + it('self-heals a pre-fix corrupted job from the stored output (no re-execute), transcript balanced', async () => { + const { jobId, ctx } = await makeJob('resume me', 'openai:gpt-4o'); // non-Anthropic: strict pairing + // Corrupted pre-fix state: seed user + assistant(tool-call), a complete + // exec row, but NO tool-result user turn at idx 2. + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'resume me' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'tool-call', toolCallId: 'prov-tc-1', toolName: 'search', input: { q: 'x' } }]); + await seedExec(jobId, 1, 'prov-tc-1', 'search', 'complete', { results: ['from-prior-run'] }, 0); + + let captured: ChatMessage[] = []; + __setChatTransportForTests(async (opts) => { + captured = opts.messages; + return { + text: 'recovered and done', blocks: [{ type: 'text', text: 'recovered and done' }] as ChatBlock[], + stopReason: 'end', usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'openai:gpt-4o', providerId: 'openai', + } satisfies ChatResult; + }); + const executions: string[] = []; + const result = await buildHandler(makeTools(executions))(ctx); + + expect(result.result).toBe('recovered and done'); + expect(executions.length).toBe(0); // stored output reused, tool NOT re-run + assertBalanced(captured); + // The healed tool-result carries the real stored output. + const healed = (captured[2].content as ChatBlock[])[0] as Extract; + expect(healed.output).toEqual({ results: ['from-prior-run'] }); + + // Durably persisted so the next resume stays balanced. + const msgs = await engine.executeRaw<{ message_idx: number; role: string }>( + `SELECT message_idx, role FROM subagent_messages WHERE job_id = $1 ORDER BY message_idx`, [jobId]); + expect(msgs.map(m => [m.message_idx, m.role])).toEqual([[0, 'user'], [1, 'assistant'], [2, 'user'], [3, 'assistant']]); + }); + + it('heals MULTIPLE consecutive dangling assistant turns (pre-fix multi-turn corruption)', async () => { + const { jobId, ctx } = await makeJob('multi', 'openai:gpt-4o'); + // Pre-fix loop persisted assistants at 1 and 3 (gaps at 2 = skipped user idx). + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'multi' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'tool-call', toolCallId: 'tc-a', toolName: 'search', input: {} }]); + await seedExec(jobId, 1, 'tc-a', 'search', 'complete', { results: ['a'] }, 0); + await seedMessage(jobId, 3, 'assistant', [{ type: 'tool-call', toolCallId: 'tc-b', toolName: 'search', input: {} }]); + await seedExec(jobId, 3, 'tc-b', 'search', 'complete', { results: ['b'] }, 0); + + let captured: ChatMessage[] = []; + __setChatTransportForTests(async (opts) => { + captured = opts.messages; + return { text: 'ok', blocks: [{ type: 'text', text: 'ok' }] as ChatBlock[], stopReason: 'end', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'openai:gpt-4o', providerId: 'openai' } satisfies ChatResult; + }); + const executions: string[] = []; + await buildHandler(makeTools(executions))(ctx); + + expect(executions.length).toBe(0); + assertBalanced(captured); + // Both dangling turns healed and persisted (idx 2 and 4). + const msgs = await engine.executeRaw<{ message_idx: number; role: string }>( + `SELECT message_idx, role FROM subagent_messages WHERE job_id = $1 ORDER BY message_idx`, [jobId]); + expect(msgs.map(m => m.message_idx)).toContain(2); + expect(msgs.map(m => m.message_idx)).toContain(4); + }); + + it('re-dispatches an idempotent tool that was still pending on resume', async () => { + const { jobId, ctx } = await makeJob('redispatch', 'openai:gpt-4o'); + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'redispatch' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'tool-call', toolCallId: 'tc-pending', toolName: 'search', input: {} }]); + await seedExec(jobId, 1, 'tc-pending', 'search', 'pending', null, 0); + + __setChatTransportForTests(async () => ({ text: 'ok', blocks: [{ type: 'text', text: 'ok' }] as ChatBlock[], stopReason: 'end', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, model: 'openai:gpt-4o', providerId: 'openai' } satisfies ChatResult)); + const executions: string[] = []; + await buildHandler(makeTools(executions))(ctx); + expect(executions).toEqual(['search']); // idempotent-pending re-executed once + }); + + it('throws on a non-idempotent tool still pending on resume', async () => { + const { jobId, ctx } = await makeJob('unsafe', 'openai:gpt-4o'); + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'unsafe' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'tool-call', toolCallId: 'tc-mut', toolName: 'put_page', input: {} }]); + await seedExec(jobId, 1, 'tc-mut', 'put_page', 'pending', null, 0); + + __setChatTransportForTests(async () => ({ text: '', blocks: [] as ChatBlock[], stopReason: 'end', + usage: { input_tokens: 0, output_tokens: 0, cache_read_tokens: 0, cache_creation_tokens: 0 }, model: 'openai:gpt-4o', providerId: 'openai' } satisfies ChatResult)); + await expect(buildHandler(makeTools([]))(ctx)).rejects.toThrow(/non-idempotent tool "put_page" pending on resume/i); + }); + + it('error-stubs a dangling tool-call whose tool is no longer registered', async () => { + const { jobId, ctx } = await makeJob('gone tool', 'openai:gpt-4o'); + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'gone tool' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'tool-call', toolCallId: 'tc-gone', toolName: 'removed_tool', input: {} }]); + // No exec row and the tool isn't in the registry (only 'search'/'put_page'). + + let captured: ChatMessage[] = []; + __setChatTransportForTests(async (opts) => { + captured = opts.messages; + return { text: 'handled', blocks: [{ type: 'text', text: 'handled' }] as ChatBlock[], stopReason: 'end', + usage: { input_tokens: 1, output_tokens: 1, cache_read_tokens: 0, cache_creation_tokens: 0 }, model: 'openai:gpt-4o', providerId: 'openai' } satisfies ChatResult; + }); + const result = await buildHandler(makeTools([]))(ctx); + expect(result.result).toBe('handled'); + assertBalanced(captured); + const stub = (captured[2].content as ChatBlock[])[0] as Extract; + expect(stub.isError).toBe(true); + expect(String(stub.output)).toContain('removed_tool'); + // Persisted as a failed exec so the next resume is stable. + const rows = await engine.executeRaw<{ status: string }>( + `SELECT status FROM subagent_tool_executions WHERE job_id = $1 AND tool_use_id = 'tc-gone'`, [jobId]); + expect(rows[0].status).toBe('failed'); + }); + + it('terminal resume: a completed transcript returns its text without calling the model', async () => { + const { jobId, ctx } = await makeJob('already done', 'openai:gpt-4o'); + await seedMessage(jobId, 0, 'user', [{ type: 'text', text: 'already done' }]); + await seedMessage(jobId, 1, 'assistant', [{ type: 'text', text: 'the final answer' }]); + + let chatCalls = 0; + __setChatTransportForTests(async () => { chatCalls++; return { text: 'SHOULD NOT RUN', blocks: [] as ChatBlock[], stopReason: 'end', + usage: { input_tokens: 0, output_tokens: 0, cache_read_tokens: 0, cache_creation_tokens: 0 }, model: 'openai:gpt-4o', providerId: 'openai' } satisfies ChatResult; }); + const result = await buildHandler(makeTools([]))(ctx); + expect(chatCalls).toBe(0); + expect(result.result).toBe('the final answer'); + expect(result.stop_reason).toBe('end_turn'); + }); +}); diff --git a/test/gateway-model-messages.test.ts b/test/gateway-model-messages.test.ts index 3fe3dde3d..8e56465c8 100644 --- a/test/gateway-model-messages.test.ts +++ b/test/gateway-model-messages.test.ts @@ -107,6 +107,63 @@ describe('toModelMessages — v6 ModelMessage shape', () => { ]); }); + test('Date in tool-result json output serializes to ISO string (Postgres timestamptz)', () => { + // node-postgres returns timestamptz columns as JS Date; AI SDK v6's + // JSONValue schema rejects a raw Date, dead-lettering the tool loop. + const msgs: ChatMessage[] = [ + { + role: 'user', + content: [{ + type: 'tool-result', + toolCallId: 'c1', + toolName: 'brain_get_page', + output: { rows: [{ updated_at: new Date('2026-06-26T06:56:59.000Z'), nested: { created_at: new Date('2026-01-02T03:04:05.000Z') } }] }, + }], + }, + ]; + const out = toModelMessages(msgs) as any[]; + const value = out[0].content[0].output.value; + expect(out[0].content[0].output.type).toBe('json'); + expect(value.rows[0].updated_at).toBe('2026-06-26T06:56:59.000Z'); + expect(value.rows[0].nested.created_at).toBe('2026-01-02T03:04:05.000Z'); + // No Date instance survives (would throw in AI SDK v6). + expect(value.rows[0].updated_at instanceof Date).toBe(false); + }); + + test('non-string text block is dropped (reasoning-model null-text guard)', () => { + // DeepSeek v4 / reasoning models can emit text:null/undefined thinking + // parts; AI SDK v6 rejects them. Dropped here; tool-call sibling kept. + const msgs: ChatMessage[] = [ + { + role: 'assistant', + content: [ + { type: 'text', text: null as unknown as string }, + { type: 'text', text: undefined as unknown as string }, + { type: 'text', text: 'kept' }, + { type: 'text', text: '' }, // empty string is valid — kept + { type: 'tool-call', toolCallId: 'c1', toolName: 'search', input: {} }, + ], + }, + ]; + const out = toModelMessages(msgs) as any[]; + expect(out[0].content).toEqual([ + { type: 'text', text: 'kept' }, + { type: 'text', text: '' }, + { type: 'tool-call', toolCallId: 'c1', toolName: 'search', input: {} }, + ]); + }); + + test('errored tool-result never throws on circular/bigint output (safeStringify)', () => { + const circular: any = {}; + circular.self = circular; + const msgs: ChatMessage[] = [ + { role: 'user', content: [{ type: 'tool-result', toolCallId: 'c1', toolName: 'x', output: circular, isError: true }] }, + ]; + const out = toModelMessages(msgs) as any[]; + expect(out[0].content[0].output.type).toBe('error-text'); + expect(typeof out[0].content[0].output.value).toBe('string'); + }); + test('full multi-turn conversation: user → assistant(tool-call) → tool(result)', () => { const msgs: ChatMessage[] = [ { role: 'user', content: 'find widget' }, diff --git a/test/think-max-output-tokens.test.ts b/test/think-max-output-tokens.test.ts new file mode 100644 index 000000000..879217ad6 --- /dev/null +++ b/test/think-max-output-tokens.test.ts @@ -0,0 +1,28 @@ +/** + * Pins `maxOutputTokensFor` — the per-model output-token budget `runThink` + * passes to `client.create`. Thinking-by-default Claude 5 models + * (`anthropic:claude-*-5`) spend a large share of the budget on internal + * reasoning before emitting an answer, so the 4000 default left `think` with + * empty/truncated text. They now get 16000; everything else stays 4000. + */ +import { describe, test, expect } from 'bun:test'; +import { maxOutputTokensFor } from '../src/core/think/index.ts'; + +describe('maxOutputTokensFor — thinking-default headroom', () => { + test('Claude 5 family gets 16000', () => { + expect(maxOutputTokensFor('anthropic:claude-sonnet-5')).toBe(16000); + expect(maxOutputTokensFor('anthropic:claude-opus-5')).toBe(16000); + expect(maxOutputTokensFor('anthropic:claude-fable-5')).toBe(16000); + expect(maxOutputTokensFor('anthropic:claude-haiku-5')).toBe(16000); + expect(maxOutputTokensFor('anthropic/claude-sonnet-5')).toBe(16000); // slash form + }); + + test('non-Claude-5 and non-Anthropic keep 4000', () => { + expect(maxOutputTokensFor('anthropic:claude-opus-4-8')).toBe(4000); + expect(maxOutputTokensFor('anthropic:claude-haiku-4-5')).toBe(4000); + expect(maxOutputTokensFor('anthropic:claude-sonnet-4-6')).toBe(4000); + expect(maxOutputTokensFor('anthropic:claude-3-haiku')).toBe(4000); + expect(maxOutputTokensFor('openai:gpt-4o')).toBe(4000); + expect(maxOutputTokensFor('deepseek:deepseek-reasoner')).toBe(4000); + }); +});