mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-31 04:07:52 +00:00
feat(agents): v0.38 Slice 1 — gateway.toolLoop() provider-agnostic loop control
Adds `gateway.toolLoop(opts)` as the provider-neutral loop wrapper over the
already-provider-neutral `gateway.chat()`. The Vercel AI SDK abstraction does
all the per-provider tool-def normalization, tool-call parsing, and tool-result
framing; this helper just sequences the assistant→tool-dispatch→tool-result
cycle with:
- D11 stable-ID callbacks (onToolCallStart returns the gbrain-owned UUID v7
that the caller persists at first observation; reread on replay)
- Write-ordering invariant (persist assistant → persist pending tool row →
execute side effect → settle complete/failed)
- Crash-replay reconciliation via `replayState.priorTools` keyed by
gbrainToolUseId (NOT provider IDs)
- Capability-driven cache_control (Anthropic only, via cacheSystem flag)
- Stop-reason mapping for refusal / content_filter / max_turns / aborted
The loop is stateless beyond the optional replay state — testable via the
existing `__setChatTransportForTests` seam without any DB.
This is the substrate Slice 1's `subagent.ts` rewire (S1.5) consumes.
Plan: ~/.claude/plans/system-instruction-you-are-working-shimmying-breeze.md
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
14d3f987e9
commit
cf91450dd8
@@ -2140,6 +2140,310 @@ export async function chat(opts: ChatOpts): Promise<ChatResult> {
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Tool loop (v0.38 — D11 + D6/D7 gateway-native subagent path) ----
|
||||
|
||||
/**
|
||||
* A tool handler runs a single tool invocation. `idempotent` lets the loop
|
||||
* safely re-execute a pending row on crash-replay; non-idempotent tools that
|
||||
* crashed mid-execute are surfaced as a hard error.
|
||||
*/
|
||||
export interface ToolHandler {
|
||||
idempotent?: boolean;
|
||||
execute(input: unknown, signal: AbortSignal): Promise<unknown>;
|
||||
}
|
||||
|
||||
/**
|
||||
* State the caller carries in from a prior crashed run. The reconciler keys
|
||||
* by gbrain-owned `gbrainToolUseId` (D11), NOT provider-supplied IDs.
|
||||
* `priorMessages` is the chat history up to the assistant's last turn;
|
||||
* `priorTools` maps gbrainToolUseId → outcome. The D5 read-time shim
|
||||
* synthesizes gbrainToolUseIds for legacy v1 rows so this Map sees both
|
||||
* shapes uniformly.
|
||||
*/
|
||||
export interface ToolLoopReplayState {
|
||||
priorMessages: ChatMessage[];
|
||||
priorTools: Map<string, { status: 'pending' | 'complete' | 'failed'; output?: unknown; error?: string }>;
|
||||
nextTurnIdx: number;
|
||||
nextMessageIdx: number;
|
||||
}
|
||||
|
||||
export interface ToolLoopOpts {
|
||||
/** "provider:modelId" — defaults to config.chat_model. */
|
||||
model?: string;
|
||||
/** System prompt (provider-neutral). Cached when caching supported + cacheSystem true. */
|
||||
system?: string;
|
||||
/**
|
||||
* Initial user message(s). When `replayState` is set, these are prepended only
|
||||
* if `replayState.priorMessages` is empty — typically empty on a fresh call,
|
||||
* non-empty on a fresh-from-scratch run.
|
||||
*/
|
||||
initialMessages: ChatMessage[];
|
||||
/** Tool definitions (provider-neutral JSON Schema). */
|
||||
tools: ChatToolDef[];
|
||||
/** Implementations keyed by tool name. */
|
||||
toolHandlers: Map<string, ToolHandler>;
|
||||
/** Hard cap on loop iterations. Default 20. */
|
||||
maxTurns?: number;
|
||||
/** Per-turn max output tokens. Default 4096. */
|
||||
maxTokens?: number;
|
||||
abortSignal?: AbortSignal;
|
||||
/** Apply Anthropic cache_control to system + last tool. Silently ignored elsewhere. */
|
||||
cacheSystem?: boolean;
|
||||
|
||||
/** Crash-replay state. When set, the loop resumes from the recorded position. */
|
||||
replayState?: ToolLoopReplayState;
|
||||
|
||||
/**
|
||||
* D11 + write-ordering invariant callbacks. Fire BEFORE side effects so a
|
||||
* crash mid-execute is reconcilable on the next replay.
|
||||
*
|
||||
* Ordering per turn:
|
||||
* 1. onAssistantTurn — assistant message persisted (D11 step 1)
|
||||
* 2. onToolCallStart — pending row persisted (D11 step 2)
|
||||
* 3. handler.execute — side effect
|
||||
* 4. onToolCallComplete / onToolCallFailed (D11 step 4)
|
||||
*/
|
||||
onAssistantTurn?: (turnIdx: number, messageIdx: number, blocks: ChatBlock[], usage: ChatResult['usage'], model: string) => Promise<void>;
|
||||
/**
|
||||
* Persist a pending tool execution. The caller assigns ordinal + uuid v7 and
|
||||
* returns them so the loop can key replay by gbrainToolUseId. The provider
|
||||
* supplies its own `providerToolCallId` (kept as a debug-only side channel).
|
||||
*/
|
||||
onToolCallStart?: (
|
||||
turnIdx: number,
|
||||
messageIdx: number,
|
||||
ordinal: number,
|
||||
toolName: string,
|
||||
input: unknown,
|
||||
providerToolCallId: string,
|
||||
) => Promise<{ gbrainToolUseId: string }>;
|
||||
onToolCallComplete?: (gbrainToolUseId: string, output: unknown) => Promise<void>;
|
||||
onToolCallFailed?: (gbrainToolUseId: string, error: string) => Promise<void>;
|
||||
|
||||
/** Optional per-call heartbeat for observability. */
|
||||
onHeartbeat?: (event: string, data: Record<string, unknown>) => void;
|
||||
}
|
||||
|
||||
export type ToolLoopStopReason = 'end' | 'max_turns' | 'refusal' | 'content_filter' | 'aborted' | 'unrecoverable';
|
||||
|
||||
export interface ToolLoopResult {
|
||||
finalText: string;
|
||||
totalTurns: number;
|
||||
totalUsage: ChatResult['usage'];
|
||||
stopReason: ToolLoopStopReason;
|
||||
/** Final messages array including all assistant + tool results. Caller persists if desired. */
|
||||
messages: ChatMessage[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Provider-agnostic tool-calling loop. Wraps `gateway.chat()` with:
|
||||
* - assistant→tool-dispatch→tool-result cycle
|
||||
* - gbrain-stable IDs (D11) at first observation
|
||||
* - write-ordering invariant (persist before side effect)
|
||||
* - crash-replay reconciliation via gbrainToolUseId
|
||||
* - capability-driven cache_control (Anthropic only)
|
||||
*
|
||||
* This replaces the direct `new Anthropic()` + `client.create()` path in
|
||||
* `src/core/minions/handlers/subagent.ts`. The provider abstraction lives in
|
||||
* `gateway.chat()` (Vercel AI SDK); this function is just the loop control.
|
||||
*
|
||||
* Designed so the caller (subagent handler) supplies persistence callbacks —
|
||||
* the loop itself is stateless beyond `replayState`. That keeps it testable
|
||||
* via `__setChatTransportForTests` without any DB.
|
||||
*/
|
||||
export async function toolLoop(opts: ToolLoopOpts): Promise<ToolLoopResult> {
|
||||
const maxTurns = opts.maxTurns ?? 20;
|
||||
const maxTokens = opts.maxTokens ?? 4096;
|
||||
const handlers = opts.toolHandlers;
|
||||
const totalUsage: ChatResult['usage'] = {
|
||||
input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
cache_read_tokens: 0,
|
||||
cache_creation_tokens: 0,
|
||||
};
|
||||
|
||||
// Seed messages: prior history (replay) or initial.
|
||||
const messages: ChatMessage[] = opts.replayState
|
||||
? [...opts.replayState.priorMessages]
|
||||
: [...opts.initialMessages];
|
||||
if (opts.replayState && opts.replayState.priorMessages.length === 0) {
|
||||
messages.push(...opts.initialMessages);
|
||||
}
|
||||
let turnIdx = opts.replayState?.nextTurnIdx ?? 0;
|
||||
let messageIdx = opts.replayState?.nextMessageIdx ?? 0;
|
||||
let finalText = '';
|
||||
let stopReason: ToolLoopStopReason = 'end';
|
||||
|
||||
while (turnIdx < maxTurns) {
|
||||
if (opts.abortSignal?.aborted) {
|
||||
stopReason = 'aborted';
|
||||
break;
|
||||
}
|
||||
|
||||
opts.onHeartbeat?.('turn_start', { turn_idx: turnIdx });
|
||||
|
||||
let chatResult: ChatResult;
|
||||
try {
|
||||
chatResult = await chat({
|
||||
model: opts.model,
|
||||
system: opts.system,
|
||||
messages,
|
||||
tools: opts.tools,
|
||||
maxTokens,
|
||||
abortSignal: opts.abortSignal,
|
||||
cacheSystem: opts.cacheSystem,
|
||||
});
|
||||
} catch (err) {
|
||||
opts.onHeartbeat?.('llm_call_failed', {
|
||||
turn_idx: turnIdx,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
throw err;
|
||||
}
|
||||
|
||||
totalUsage.input_tokens += chatResult.usage.input_tokens;
|
||||
totalUsage.output_tokens += chatResult.usage.output_tokens;
|
||||
totalUsage.cache_read_tokens += chatResult.usage.cache_read_tokens;
|
||||
totalUsage.cache_creation_tokens += chatResult.usage.cache_creation_tokens;
|
||||
|
||||
// D11 step 1: persist assistant turn BEFORE any tool dispatch.
|
||||
const assistantMessageIdx = messageIdx++;
|
||||
await opts.onAssistantTurn?.(turnIdx, assistantMessageIdx, chatResult.blocks, chatResult.usage, chatResult.model);
|
||||
messages.push({ role: 'assistant', content: chatResult.blocks });
|
||||
|
||||
// Check stop reason BEFORE tool dispatch. The loop only continues on tool_calls.
|
||||
if (chatResult.stopReason === 'refusal') {
|
||||
stopReason = 'refusal';
|
||||
finalText = chatResult.text;
|
||||
break;
|
||||
}
|
||||
if (chatResult.stopReason === 'content_filter') {
|
||||
stopReason = 'content_filter';
|
||||
finalText = chatResult.text;
|
||||
break;
|
||||
}
|
||||
|
||||
const toolCalls = chatResult.blocks.filter(
|
||||
(b): b is { type: 'tool-call'; toolCallId: string; toolName: string; input: unknown } =>
|
||||
b.type === 'tool-call',
|
||||
);
|
||||
|
||||
if (toolCalls.length === 0) {
|
||||
stopReason = 'end';
|
||||
finalText = chatResult.text;
|
||||
break;
|
||||
}
|
||||
|
||||
// D11 + write-ordering invariant: persist pending → execute → settle.
|
||||
const toolResultBlocks: ChatBlock[] = [];
|
||||
for (let callIdx = 0; callIdx < toolCalls.length; callIdx++) {
|
||||
const call = toolCalls[callIdx];
|
||||
if (opts.abortSignal?.aborted) {
|
||||
stopReason = 'aborted';
|
||||
break;
|
||||
}
|
||||
|
||||
const handler = handlers.get(call.toolName);
|
||||
if (!handler) {
|
||||
// Tool not registered. Synthesize an error result; don't persist.
|
||||
toolResultBlocks.push({
|
||||
type: 'tool-result',
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
output: `tool "${call.toolName}" is not in the registry for this subagent`,
|
||||
isError: true,
|
||||
});
|
||||
opts.onHeartbeat?.('tool_failed', { turn_idx: turnIdx, tool_name: call.toolName, error: 'not_registered' });
|
||||
continue;
|
||||
}
|
||||
|
||||
// Step 2: persist pending row + claim gbrainToolUseId. The caller's
|
||||
// callback handles uniqueness contention via ON CONFLICT DO NOTHING +
|
||||
// re-read pattern (see persistToolExecPending in subagent.ts).
|
||||
const { gbrainToolUseId } = (await opts.onToolCallStart?.(
|
||||
turnIdx,
|
||||
assistantMessageIdx,
|
||||
callIdx,
|
||||
call.toolName,
|
||||
call.input,
|
||||
call.toolCallId,
|
||||
)) ?? { gbrainToolUseId: `inline-${turnIdx}-${callIdx}` };
|
||||
|
||||
// Replay short-circuit: prior outcome wins, idempotent re-execute allowed.
|
||||
const prior = opts.replayState?.priorTools.get(gbrainToolUseId);
|
||||
if (prior?.status === 'complete') {
|
||||
toolResultBlocks.push({
|
||||
type: 'tool-result',
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
output: prior.output,
|
||||
});
|
||||
opts.onHeartbeat?.('tool_replay_complete', { turn_idx: turnIdx, tool_name: call.toolName });
|
||||
continue;
|
||||
}
|
||||
if (prior?.status === 'failed') {
|
||||
toolResultBlocks.push({
|
||||
type: 'tool-result',
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
output: prior.error ?? 'tool failed',
|
||||
isError: true,
|
||||
});
|
||||
opts.onHeartbeat?.('tool_replay_failed', { turn_idx: turnIdx, tool_name: call.toolName });
|
||||
continue;
|
||||
}
|
||||
if (prior?.status === 'pending' && !handler.idempotent) {
|
||||
// Non-idempotent crash-mid-execute. Surface as unrecoverable.
|
||||
stopReason = 'unrecoverable';
|
||||
throw new Error(
|
||||
`non-idempotent tool "${call.toolName}" pending on resume; gbrainToolUseId=${gbrainToolUseId} — cannot safely re-run`,
|
||||
);
|
||||
}
|
||||
|
||||
// Step 3: execute (side effect).
|
||||
opts.onHeartbeat?.('tool_called', { turn_idx: turnIdx, tool_name: call.toolName });
|
||||
try {
|
||||
const output = await handler.execute(call.input, opts.abortSignal ?? new AbortController().signal);
|
||||
// Step 4: settle complete.
|
||||
await opts.onToolCallComplete?.(gbrainToolUseId, output);
|
||||
toolResultBlocks.push({
|
||||
type: 'tool-result',
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
output,
|
||||
});
|
||||
opts.onHeartbeat?.('tool_result', { turn_idx: turnIdx, tool_name: call.toolName });
|
||||
} catch (err) {
|
||||
const errMsg = err instanceof Error ? err.message : String(err);
|
||||
await opts.onToolCallFailed?.(gbrainToolUseId, errMsg);
|
||||
toolResultBlocks.push({
|
||||
type: 'tool-result',
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
output: errMsg,
|
||||
isError: true,
|
||||
});
|
||||
opts.onHeartbeat?.('tool_failed', { turn_idx: turnIdx, tool_name: call.toolName, error: errMsg });
|
||||
}
|
||||
}
|
||||
|
||||
if (stopReason === 'aborted') break;
|
||||
|
||||
// Feed all tool results back as a single user message.
|
||||
const userMessageIdx = messageIdx++;
|
||||
void userMessageIdx;
|
||||
messages.push({ role: 'user', content: toolResultBlocks });
|
||||
|
||||
turnIdx++;
|
||||
}
|
||||
|
||||
if (turnIdx >= maxTurns && stopReason === 'end') {
|
||||
stopReason = 'max_turns';
|
||||
}
|
||||
|
||||
return { finalText, totalTurns: turnIdx, totalUsage, stopReason, messages };
|
||||
}
|
||||
|
||||
// ---- Reranker (v0.35.0.0+) ----
|
||||
|
||||
/** Tagged error class for gateway.rerank() failures. `reason` classifies into the
|
||||
|
||||
Reference in New Issue
Block a user