mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
fix(gateway): consolidate tool-loop resume + provider fixes (fix-wave A) (#2820)
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: Sinabina <sinabina@Sinabinas-MacBook-Pro-4.local> Co-authored-by: thomaskong119 <thomaskong119@hotmail.com> Co-authored-by: maxpetrusenkoagent <max.petrusenko.agent@gmail.com> Co-authored-by: brettdavies <brettdavies@users.noreply.github.com> Co-authored-by: ivandebot <187176982+ivandebot@users.noreply.github.com> Co-authored-by: Rafael Reis <rafael.reis@contabilizei.com.br> Co-authored-by: fbal23 <fbal.public@gmail.com> Co-authored-by: javieraldape <javieraldape@users.noreply.github.com> Co-authored-by: David Carolan <david@joyrestart.com> Co-authored-by: Masashi-Ono0611 <masashi.ono.0611@gmail.com> Co-authored-by: spiky02plateau <155588579+spiky02plateau@users.noreply.github.com> Co-authored-by: psam-717 <mphilannorbah@gmail.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Time Attakc
co-authored by
Sinabina
thomaskong119
maxpetrusenkoagent
brettdavies
ivandebot
Rafael Reis
fbal23
javieraldape
David Carolan
Masashi-Ono0611
spiky02plateau
psam-717
Claude Opus 4.8
Sinabina
thomaskong119
maxpetrusenkoagent
brettdavies
ivandebot
Rafael Reis
fbal23
javieraldape
David Carolan
Masashi-Ono0611
spiky02plateau
psam-717
Claude Opus 4.8
parent
ec3910afc4
commit
86962b242b
@@ -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
|
||||
|
||||
+140
-15
@@ -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<ChatBlock, { type: 'tool-call' }> => 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<ChatBlock, { type: 'tool-result' }> => 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<ChatResult> {
|
||||
}
|
||||
}
|
||||
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<ChatResult> {
|
||||
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<void>;
|
||||
onToolCallFailed?: (gbrainToolUseId: string, error: string) => Promise<void>;
|
||||
/**
|
||||
* 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<void>;
|
||||
|
||||
/** Optional per-call heartbeat for observability. */
|
||||
onHeartbeat?: (event: string, data: Record<string, unknown>) => void;
|
||||
@@ -2991,7 +3114,7 @@ export interface ToolLoopResult {
|
||||
*/
|
||||
export async function toolLoop(opts: ToolLoopOpts): Promise<ToolLoopResult> {
|
||||
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<ToolLoopResult> {
|
||||
|
||||
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++;
|
||||
|
||||
@@ -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<Response> => {
|
||||
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 },
|
||||
};
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -777,22 +777,57 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise<SubagentResu
|
||||
});
|
||||
}
|
||||
|
||||
// Convert prior Anthropic-shape messages → ChatMessage with ChatBlock content.
|
||||
// v1 rows store Anthropic content blocks ({type:'tool_use'|'tool_result'|...});
|
||||
// we adapt them to ChatBlock shape (type: 'tool-call' | 'tool-result' | 'text').
|
||||
const priorChatMessages: ChatMessage[] = priorMessages.map(m => ({
|
||||
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<SubagentResu
|
||||
[errorMsg, gbrainToolUseId],
|
||||
);
|
||||
},
|
||||
// Persist the tool-result user turn so a resume reloads a balanced
|
||||
// transcript. JSON.stringify inside persistMessage ISO-izes any Date in
|
||||
// tool output, and the value binds through $N::text::jsonb (never
|
||||
// JSON.stringify into a bare ::jsonb).
|
||||
onToolResultTurn: async (_turnIdx, messageIdx, blocks) => {
|
||||
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<SubagentResu
|
||||
};
|
||||
}
|
||||
|
||||
interface ReconcileArgs {
|
||||
engine: BrainEngine;
|
||||
jobId: number;
|
||||
priorMessages: PersistedMessage[];
|
||||
priorTools: PersistedToolExec[];
|
||||
toolDefs: ToolDef[];
|
||||
signal: AbortSignal;
|
||||
}
|
||||
|
||||
interface ReconcileResult {
|
||||
chatMessages: ChatMessage[];
|
||||
/** max(known/healed message_idx) + 1 — 0 when there is no prior transcript. */
|
||||
nextMessageIdx: number;
|
||||
/** Non-null when the transcript already ended on an assistant text turn. */
|
||||
terminalText: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Heal an unbalanced gateway replay transcript before it reaches the provider.
|
||||
*
|
||||
* The gateway loop persists each assistant turn but historically never
|
||||
* persisted the following tool-result user turn, so a resumed job reloaded
|
||||
* assistant tool-calls with no matching results. Non-Anthropic (openai-compat)
|
||||
* providers reject that with AI_MissingToolResultsError and the job
|
||||
* dead-letters. This walks EVERY prior assistant turn that carries tool-call
|
||||
* blocks (not just the tail — a pre-fix multi-turn job persisted several
|
||||
* consecutive dangling assistant turns) and, if the turn isn't already answered
|
||||
* by the following tool-result user turn, synthesizes that turn from the
|
||||
* settled `subagent_tool_executions` rows with the same semantics as the legacy
|
||||
* Anthropic replay path:
|
||||
* - complete → stored output
|
||||
* - failed → stored error (isError)
|
||||
* - idempotent-pending / missing row → re-dispatch, persist, use output
|
||||
* - non-idempotent-pending → throw (cannot safely re-run)
|
||||
* - tool no longer registered → persist failed + error stub
|
||||
* Each synthesized turn is persisted at `assistant.message_idx + 1` (the free
|
||||
* slot the pre-fix loop skipped) via `ON CONFLICT DO NOTHING`, so the next
|
||||
* resume stays balanced.
|
||||
*/
|
||||
async function reconcileGatewayReplay(args: ReconcileArgs): Promise<ReconcileResult> {
|
||||
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<string, PersistedToolExec>();
|
||||
const execByMsg = new Map<number, PersistedToolExec[]>();
|
||||
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<ChatBlock, { type: 'tool-call' }> => 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<ChatBlock, { type: 'text' }> => 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<Persi
|
||||
const rows = await engine.executeRaw<Record<string, unknown>>(
|
||||
`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 => ({
|
||||
|
||||
+14
-1
@@ -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 }],
|
||||
});
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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<ChatBlock, { type: 'tool-result' }>).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<number | undefined> = [];
|
||||
__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: '',
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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<ChatBlock, { type: 'tool-call' }> => 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<ChatBlock, { type: 'tool-result' }> => 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<ChatBlock, { type: 'tool-result' }>;
|
||||
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<ChatBlock, { type: 'tool-result' }>;
|
||||
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');
|
||||
});
|
||||
});
|
||||
@@ -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' },
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user