diff --git a/src/agents/claude-agent-sdk/index.ts b/src/agents/claude-agent-sdk/index.ts index 0692b038f..f125899cf 100644 --- a/src/agents/claude-agent-sdk/index.ts +++ b/src/agents/claude-agent-sdk/index.ts @@ -25,6 +25,10 @@ export { readSessionHistory, loadSessionHistoryForSdk } from "./sdk-session-hist export { appendSdkTextTurnToSessionTranscript, appendSdkTurnPairToSessionTranscript, + appendSdkToolCallsToSessionTranscript, + appendSdkToolUseToSessionTranscript, + appendSdkToolResultToSessionTranscript, + type SdkToolCallRecord, } from "./sdk-session-transcript.js"; // Hooks @@ -44,6 +48,7 @@ export { wrapToolHandler, type BridgeOptions, type BridgeResult, + type OnToolUpdateCallback, } from "./tool-bridge.js"; export type { McpCallToolResult, @@ -56,6 +61,7 @@ export type { SdkRunnerQueryOptions, } from "./tool-bridge.types.js"; export type { + SdkCompletedToolCall, SdkConversationTurn, SdkDoneEvent, SdkErrorEvent, @@ -70,4 +76,5 @@ export type { SdkTextEvent, SdkToolResultEvent, SdkToolUseEvent, + SdkUsageStats, } from "./types.js"; diff --git a/src/agents/claude-agent-sdk/sdk-agent-runtime.ts b/src/agents/claude-agent-sdk/sdk-agent-runtime.ts index e16d9f1b7..807c93d6c 100644 --- a/src/agents/claude-agent-sdk/sdk-agent-runtime.ts +++ b/src/agents/claude-agent-sdk/sdk-agent-runtime.ts @@ -12,7 +12,10 @@ import { runSdkAgent } from "./sdk-runner.js"; import { resolveProviderConfig } from "./provider-config.js"; import { isSdkAvailable } from "./sdk-loader.js"; import { loadSessionHistoryForSdk } from "./sdk-session-history.js"; -import { appendSdkTurnPairToSessionTranscript } from "./sdk-session-transcript.js"; +import { + appendSdkTurnPairToSessionTranscript, + appendSdkToolCallsToSessionTranscript, +} from "./sdk-session-transcript.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import type { SdkConversationTurn, SdkRunnerResult } from "./types.js"; @@ -39,6 +42,17 @@ export type CcSdkAgentRuntimeContext = { * Convert an SdkRunnerResult into an AgentRuntimeResult. */ function adaptSdkResult(result: SdkRunnerResult, sessionId: string): AgentRuntimeResult { + // Map SDK usage stats to the Pi-embedded format. + const usage = result.meta.usage + ? { + input: result.meta.usage.inputTokens, + output: result.meta.usage.outputTokens, + cacheRead: result.meta.usage.cacheReadInputTokens, + cacheWrite: result.meta.usage.cacheCreationInputTokens, + total: result.meta.usage.totalTokens, + } + : undefined; + return { payloads: result.payloads.map((p) => ({ text: p.text, @@ -51,12 +65,16 @@ function adaptSdkResult(result: SdkRunnerResult, sessionId: string): AgentRuntim sessionId, provider: result.meta.provider ?? "sdk", model: result.meta.model ?? "default", + usage, }, // SDK runner errors are rendered as text payloads with isError=true. // Avoid mapping to Pi-specific error kinds (context/compaction) because // downstream recovery logic would treat them incorrectly. error: undefined, }, + didSendViaMessagingTool: result.didSendViaMessagingTool, + messagingToolSentTexts: result.messagingToolSentTexts, + messagingToolSentTargets: result.messagingToolSentTargets, }; } @@ -120,6 +138,7 @@ export function createCcSdkAgentRuntime(context?: CcSdkAgentRuntimeContext): Age hooksEnabled: context?.ccsdkConfig?.hooksEnabled, sdkOptions: context?.ccsdkConfig?.options, modelTiers: context?.ccsdkConfig?.models, + thinkingLevel: params.thinkLevel, conversationHistory, tools: context?.tools, @@ -137,9 +156,17 @@ export function createCcSdkAgentRuntime(context?: CcSdkAgentRuntimeContext): Age onAgentEvent: params.onAgentEvent, }); - // Persist a minimal user/assistant turn pair so SDK main-agent mode has multi-turn continuity. - // This intentionally records only text, not tool call structures. + // Persist session transcript for multi-turn continuity. if (params.sessionFile) { + // Record tool calls with structured format (matching pi-agent). + if (sdkResult.completedToolCalls && sdkResult.completedToolCalls.length > 0) { + appendSdkToolCallsToSessionTranscript({ + sessionFile: params.sessionFile, + toolCalls: sdkResult.completedToolCalls, + }); + } + + // Record user/assistant text turn pair. appendSdkTurnPairToSessionTranscript({ sessionFile: params.sessionFile, prompt: params.prompt, diff --git a/src/agents/claude-agent-sdk/sdk-runner.ts b/src/agents/claude-agent-sdk/sdk-runner.ts index fb7317ac8..2b871c423 100644 --- a/src/agents/claude-agent-sdk/sdk-runner.ts +++ b/src/agents/claude-agent-sdk/sdk-runner.ts @@ -23,10 +23,116 @@ import { buildHistorySystemPromptSuffix } from "./sdk-history.js"; import { isSdkTerminalToolEventType } from "./sdk-event-checks.js"; import { buildMoltbotSdkHooks } from "./sdk-hooks.js"; import { normalizeToolName } from "../tool-policy.js"; -import type { SdkRunnerParams, SdkRunnerResult } from "./types.js"; +import type { + SdkRunnerParams, + SdkRunnerResult, + SdkUsageStats, + SdkCompletedToolCall, +} from "./types.js"; +import type { ThinkLevel } from "../../auto-reply/thinking.js"; +import { + isMessagingTool, + isMessagingToolSendAction, + type MessagingToolSend, +} from "../pi-embedded-messaging.js"; +import { extractMessagingToolSend } from "../pi-embedded-subscribe.tools.js"; +// Session transcript recording is done in sdk-agent-runtime.ts, which receives +// completedToolCalls from the result and passes them to appendSdkToolCallsToSessionTranscript. const log = createSubsystemLogger("agents/claude-agent-sdk/sdk-runner"); +// --------------------------------------------------------------------------- +// Thinking budget mapping +// --------------------------------------------------------------------------- + +/** + * Map thinking levels to SDK budget tokens. + * These values are tuned to match Claude's expected thinking depths. + */ +const THINKING_BUDGET_MAP: Record = { + off: 0, + minimal: 2_000, + low: 5_000, + medium: 10_000, + high: 20_000, + xhigh: 40_000, +}; + +function getThinkingBudget(level?: ThinkLevel): number | undefined { + if (!level || level === "off") return undefined; + return THINKING_BUDGET_MAP[level]; +} + +/** + * Extract usage statistics from an SDK event. + * Usage can appear in result events or as a separate usage event. + */ +function extractUsageFromEvent(event: unknown): SdkUsageStats | undefined { + if (!isRecord(event)) return undefined; + + // Check for usage in the event itself + const usageObj = event.usage as Record | undefined; + if (!usageObj || typeof usageObj !== "object") { + // Also check for usage nested in data + const data = event.data as Record | undefined; + const dataUsage = data?.usage as Record | undefined; + if (!dataUsage || typeof dataUsage !== "object") return undefined; + return normalizeUsageObject(dataUsage); + } + + return normalizeUsageObject(usageObj); +} + +function normalizeUsageObject(obj: Record): SdkUsageStats | undefined { + const inputTokens = + typeof obj.input_tokens === "number" + ? obj.input_tokens + : typeof obj.inputTokens === "number" + ? obj.inputTokens + : undefined; + const outputTokens = + typeof obj.output_tokens === "number" + ? obj.output_tokens + : typeof obj.outputTokens === "number" + ? obj.outputTokens + : undefined; + const cacheReadInputTokens = + typeof obj.cache_read_input_tokens === "number" + ? obj.cache_read_input_tokens + : typeof obj.cacheReadInputTokens === "number" + ? obj.cacheReadInputTokens + : undefined; + const cacheCreationInputTokens = + typeof obj.cache_creation_input_tokens === "number" + ? obj.cache_creation_input_tokens + : typeof obj.cacheCreationInputTokens === "number" + ? obj.cacheCreationInputTokens + : undefined; + + if ( + inputTokens === undefined && + outputTokens === undefined && + cacheReadInputTokens === undefined && + cacheCreationInputTokens === undefined + ) { + return undefined; + } + + const totalTokens = + (inputTokens ?? 0) + + (outputTokens ?? 0) + + (cacheReadInputTokens ?? 0) + + (cacheCreationInputTokens ?? 0); + + return { + inputTokens, + outputTokens, + cacheReadInputTokens, + cacheCreationInputTokens, + totalTokens: totalTokens > 0 ? totalTokens : undefined, + }; +} + // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- @@ -260,6 +366,15 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { + emitEvent("tool", { + phase: "update", + name: updateParams.toolName, + toolCallId: updateParams.toolCallId, + update: updateParams.update, + }); + }, }); } catch (err) { const message = err instanceof Error ? err.message : String(err); @@ -337,8 +452,22 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise; } + // Thinking budget (extended thinking support). + const thinkingBudget = getThinkingBudget(params.thinkingLevel); + if (thinkingBudget !== undefined) { + sdkOptions.budgetTokens = thinkingBudget; + log.debug(`Set thinking budget to ${thinkingBudget} tokens for level: ${params.thinkingLevel}`); + } + // ------------------------------------------------------------------------- // Step 4: Build the prompt // ------------------------------------------------------------------------- @@ -387,10 +523,20 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { name, args } + const pendingToolCalls = new Map }>(); + // Completed tool calls for session transcript recording. + const completedToolCalls: SdkCompletedToolCall[] = []; + // Set up timeout if configured. let timeoutId: ReturnType | undefined; const timeoutController = new AbortController(); @@ -441,7 +587,7 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise) : undefined; const type = record && typeof record.type === "string" ? record.type : undefined; @@ -474,6 +620,47 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise) ?? + (record.arguments as Record) ?? + {}; + // Track all tool calls, not just messaging tools. + pendingToolCalls.set(toolCallId, { name: normalizedName.name, args }); + } + + // On completion, record tool call and check for messaging tool send. + if (phase === "result" && toolCallId) { + const pending = pendingToolCalls.get(toolCallId); + if (pending) { + // Record completed tool call for session transcript. + completedToolCalls.push({ + toolCallId, + toolName: pending.name, + args: pending.args, + result: toolText, + isError, + }); + + // Check for messaging tool send (only on success). + if (!isError && isMessagingToolSendAction(pending.name, pending.args)) { + const sendTarget = extractMessagingToolSend(pending.name, pending.args); + if (sendTarget) { + messagingToolSentTargets.push(sendTarget); + } + // Extract text from args for deduplication. + const textArg = pending.args.text ?? pending.args.message ?? pending.args.content; + if (typeof textArg === "string" && textArg.trim()) { + messagingToolSentTexts.push(textArg.trim()); + } + } + } + pendingToolCalls.delete(toolCallId); + } + emitEvent("tool", { phase, name: normalizedName.name, @@ -510,6 +697,11 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise 0, + messagingToolSentTexts: messagingToolSentTexts.length > 0 ? messagingToolSentTexts : undefined, + messagingToolSentTargets: + messagingToolSentTargets.length > 0 ? messagingToolSentTargets : undefined, + completedToolCalls: completedToolCalls.length > 0 ? completedToolCalls : undefined, + claudeSessionId: extractedSessionId, }; } diff --git a/src/agents/claude-agent-sdk/sdk-session-transcript.ts b/src/agents/claude-agent-sdk/sdk-session-transcript.ts index 14ebc8ae4..48c0f3a89 100644 --- a/src/agents/claude-agent-sdk/sdk-session-transcript.ts +++ b/src/agents/claude-agent-sdk/sdk-session-transcript.ts @@ -91,3 +91,135 @@ export function appendSdkTurnPairToSessionTranscript(params: { }); } } + +/** + * A tracked tool call for session transcript recording. + */ +export type SdkToolCallRecord = { + /** Tool call ID from the model. */ + toolCallId: string; + /** Normalized tool name. */ + toolName: string; + /** Tool input arguments. */ + args: Record; + /** Tool result (if completed). */ + result?: unknown; + /** Whether the tool execution resulted in an error. */ + isError?: boolean; +}; + +/** + * Append a tool use block to the session transcript. + * + * Records tool calls in a structured format matching the pi-agent session format. + */ +export function appendSdkToolUseToSessionTranscript(params: { + sessionFile: string; + toolCallId: string; + toolName: string; + args: Record; + timestamp?: number; +}): void { + try { + appendJsonlLine({ + filePath: params.sessionFile, + value: { + message: { + role: "assistant", + content: [ + { + type: "tool_use", + id: params.toolCallId, + name: params.toolName, + input: params.args, + }, + ], + timestamp: params.timestamp ?? Date.now(), + }, + }, + }); + } catch (err) { + log.debug(`Failed to append tool use to transcript: ${String(err)}`); + } +} + +/** + * Append a tool result block to the session transcript. + * + * Records tool results in a structured format matching the pi-agent session format. + */ +export function appendSdkToolResultToSessionTranscript(params: { + sessionFile: string; + toolCallId: string; + result: unknown; + isError?: boolean; + timestamp?: number; +}): void { + try { + // Serialize result to text if it's an object. + let resultContent: string; + if (typeof params.result === "string") { + resultContent = params.result; + } else if (params.result !== undefined) { + try { + resultContent = JSON.stringify(params.result); + } catch { + resultContent = String(params.result); + } + } else { + resultContent = "(no output)"; + } + + appendJsonlLine({ + filePath: params.sessionFile, + value: { + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: params.toolCallId, + content: resultContent, + is_error: params.isError, + }, + ], + timestamp: params.timestamp ?? Date.now(), + }, + }, + }); + } catch (err) { + log.debug(`Failed to append tool result to transcript: ${String(err)}`); + } +} + +/** + * Append multiple tool calls from a single run to the session transcript. + * + * This records completed tool calls with both the tool_use and tool_result + * blocks for full transcript parity with pi-agent format. + */ +export function appendSdkToolCallsToSessionTranscript(params: { + sessionFile: string; + toolCalls: SdkToolCallRecord[]; + timestamp?: number; +}): void { + const ts = params.timestamp ?? Date.now(); + for (const call of params.toolCalls) { + appendSdkToolUseToSessionTranscript({ + sessionFile: params.sessionFile, + toolCallId: call.toolCallId, + toolName: call.toolName, + args: call.args, + timestamp: ts, + }); + if (call.result !== undefined) { + appendSdkToolResultToSessionTranscript({ + sessionFile: params.sessionFile, + toolCallId: call.toolCallId, + result: call.result, + isError: call.isError, + timestamp: ts, + }); + } + } +} diff --git a/src/agents/claude-agent-sdk/tool-bridge.ts b/src/agents/claude-agent-sdk/tool-bridge.ts index d5699ea30..031ca37d1 100644 --- a/src/agents/claude-agent-sdk/tool-bridge.ts +++ b/src/agents/claude-agent-sdk/tool-bridge.ts @@ -134,12 +134,14 @@ export function convertToolResult(result: AgentToolResult): McpCallTool * - Moltbot: `execute(toolCallId, params, signal?, onUpdate?)` * - MCP: `handler(args) → Promise` * - * Notable: MCP handlers have no abort signal or streaming update callback. - * We pass the shared `abortSignal` (if provided) and skip `onUpdate`. + * Notable: MCP handlers have no native streaming update callback. + * We pass the shared `abortSignal` (if provided) and create an `onUpdate` + * that forwards to the provided callback. */ export function wrapToolHandler( tool: AnyAgentTool, abortSignal?: AbortSignal, + onToolUpdate?: OnToolUpdateCallback, ): (args: Record) => Promise { const normalizedName = normalizeToolName(tool.name); @@ -151,13 +153,19 @@ export function wrapToolHandler( // model's response. const toolCallId = `mcp-bridge-${normalizedName}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + // Create an onUpdate callback that forwards to the bridge callback. + const onUpdate = onToolUpdate + ? (update: unknown) => { + void Promise.resolve( + onToolUpdate({ toolCallId, toolName: normalizedName, update }), + ).catch(() => { + // Don't let async callback errors break tool execution. + }); + } + : undefined; + try { - const result = await tool.execute( - toolCallId, - args, - abortSignal, // 3rd arg: AbortSignal (undefined if not provided) - undefined, // 4th arg: onUpdate — not supported in MCP tool protocol - ); + const result = await tool.execute(toolCallId, args, abortSignal, onUpdate); return convertToolResult(result); } catch (err) { // Propagate AbortError so the SDK runner can handle cancellation. @@ -234,6 +242,13 @@ export function buildMcpAllowedTools(serverName: string, tools: AnyAgentTool[]): // Main bridge: Moltbot tools → MCP server config // --------------------------------------------------------------------------- +/** Callback for tool execution updates (streaming progress). */ +export type OnToolUpdateCallback = (params: { + toolCallId: string; + toolName: string; + update: unknown; +}) => void | Promise; + export type BridgeOptions = { /** MCP server name (used in mcp__{name}__{tool} naming). */ name: string; @@ -241,6 +256,8 @@ export type BridgeOptions = { tools: AnyAgentTool[]; /** Optional shared abort signal for all tool executions. */ abortSignal?: AbortSignal; + /** Optional callback for tool execution updates (streaming progress). */ + onToolUpdate?: OnToolUpdateCallback; }; export type BridgeResult = { @@ -286,7 +303,7 @@ export async function bridgeMoltbotToolsToMcpServer(options: BridgeOptions): Pro try { const jsonSchema = extractJsonSchema(tool); - const handler = wrapToolHandler(tool, options.abortSignal); + const handler = wrapToolHandler(tool, options.abortSignal, options.onToolUpdate); server.tool(toolName, tool.description ?? `Moltbot tool: ${toolName}`, jsonSchema, handler); @@ -341,7 +358,7 @@ export function bridgeMoltbotToolsSync( try { const jsonSchema = extractJsonSchema(tool); - const handler = wrapToolHandler(tool, options.abortSignal); + const handler = wrapToolHandler(tool, options.abortSignal, options.onToolUpdate); server.tool(toolName, tool.description ?? `Moltbot tool: ${toolName}`, jsonSchema, handler); registered.push(toolName); } catch (err) { diff --git a/src/agents/claude-agent-sdk/tool-bridge.types.ts b/src/agents/claude-agent-sdk/tool-bridge.types.ts index bcbd9c84c..ee54f020e 100644 --- a/src/agents/claude-agent-sdk/tool-bridge.types.ts +++ b/src/agents/claude-agent-sdk/tool-bridge.types.ts @@ -71,8 +71,16 @@ export type SdkRunnerQueryOptions = { maxTurns?: number; /** Model to use. */ model?: string; + /** Thinking budget in tokens for extended thinking. */ + budgetTokens?: number; /** Additional directories the agent can access. */ additionalDirectories?: string[]; + /** Session ID to resume (loads conversation history from the session). */ + resume?: string; + /** When resuming, fork to a new session ID instead of continuing. */ + forkSession?: boolean; + /** Continue the most recent conversation (mutually exclusive with resume). */ + continue?: boolean; /** Where to load Claude Code settings from ("project", etc.). */ settingSources?: string[]; /** Include partial message events in the SDK stream. */ diff --git a/src/agents/claude-agent-sdk/types.ts b/src/agents/claude-agent-sdk/types.ts index c61246e77..30d917d14 100644 --- a/src/agents/claude-agent-sdk/types.ts +++ b/src/agents/claude-agent-sdk/types.ts @@ -5,6 +5,8 @@ import type { MoltbotConfig } from "../../config/config.js"; import type { CcSdkModelTiers } from "../../config/types.agents.js"; import type { AnyAgentTool } from "../tools/common.js"; +import type { ThinkLevel } from "../../auto-reply/thinking.js"; +import type { MessagingToolSend } from "../pi-embedded-messaging.js"; /** Provider environment variables for SDK authentication and model config. */ export type SdkProviderEnv = { @@ -91,6 +93,9 @@ export type SdkRunnerParams = { /** 3-tier model configuration (haiku/sonnet/opus). */ modelTiers?: CcSdkModelTiers; + /** Thinking level for extended thinking (maps to SDK budgetTokens). */ + thinkingLevel?: ThinkLevel; + /** * Pre-built Moltbot tools to expose to the agent. * These should already be policy-filtered. @@ -111,6 +116,21 @@ export type SdkRunnerParams = { /** Max agent turns before the SDK stops. */ maxTurns?: number; + // --- Native session resume parameters --- + + /** + * CCSDK session ID to resume. When provided, the SDK will load + * conversation history from the specified session instead of + * injecting history into the system prompt. + */ + claudeSessionId?: string; + + /** + * When resuming, fork to a new session ID instead of continuing + * the previous session. Use with claudeSessionId. + */ + forkSession?: boolean; + /** * MCP server name for the bridged Moltbot tools. * Defaults to "moltbot". @@ -221,6 +241,15 @@ export type SdkRunnerErrorKind = | "timeout" | "no_output"; +/** Usage statistics from the SDK execution. */ +export type SdkUsageStats = { + inputTokens?: number; + outputTokens?: number; + cacheReadInputTokens?: number; + cacheCreationInputTokens?: number; + totalTokens?: number; +}; + /** Metadata from an SDK runner execution. */ export type SdkRunnerMeta = { durationMs: number; @@ -230,6 +259,8 @@ export type SdkRunnerMeta = { extractedChars: number; truncated: boolean; aborted?: boolean; + /** Token usage statistics. */ + usage?: SdkUsageStats; error?: { kind: SdkRunnerErrorKind; message: string; @@ -242,6 +273,20 @@ export type SdkRunnerMeta = { }; }; +/** A completed tool call record for session transcript parity. */ +export type SdkCompletedToolCall = { + /** Tool call ID from the model. */ + toolCallId: string; + /** Normalized tool name. */ + toolName: string; + /** Tool input arguments. */ + args: Record; + /** Tool result text (if completed). */ + result?: string; + /** Whether the tool execution resulted in an error. */ + isError?: boolean; +}; + /** Result from SDK runner execution. */ export type SdkRunnerResult = { /** Extracted text payloads from the agent run. */ @@ -251,4 +296,17 @@ export type SdkRunnerResult = { }>; /** Run metadata. */ meta: SdkRunnerMeta; + /** True if a messaging tool successfully sent a message during the run. */ + didSendViaMessagingTool?: boolean; + /** Texts successfully sent via messaging tools during the run. */ + messagingToolSentTexts?: string[]; + /** Messaging tool targets that successfully sent a message during the run. */ + messagingToolSentTargets?: MessagingToolSend[]; + /** Completed tool calls for session transcript recording. */ + completedToolCalls?: SdkCompletedToolCall[]; + /** + * CCSDK session ID for the current run. Can be used to resume + * the session in a future run via the claudeSessionId parameter. + */ + claudeSessionId?: string; }; diff --git a/src/agents/unified-runtime-adapter.ts b/src/agents/unified-runtime-adapter.ts new file mode 100644 index 000000000..1ade4eed8 --- /dev/null +++ b/src/agents/unified-runtime-adapter.ts @@ -0,0 +1,401 @@ +/** + * Unified Runtime Adapter for cross-runtime failover. + * + * Provides a common abstraction layer above both pi-agent and CCSDK runtimes, + * supporting: + * - Auth profile rotation with cooldown across both runtimes + * - Model fallback chains that can span runtimes + * - Unified error handling and FailoverError propagation + * + * @module agents/unified-runtime-adapter + */ + +import { createSubsystemLogger } from "../logging/subsystem.js"; +import type { MoltbotConfig } from "../config/config.js"; +import type { AgentRuntimeKind, AgentRuntimeResult } from "./agent-runtime.js"; +import { + coerceToFailoverError, + describeFailoverError, + isFailoverError, + isTimeoutError, +} from "./failover-error.js"; +import { + ensureAuthProfileStore, + isProfileInCooldown, + markAuthProfileCooldown, + resolveAuthProfileOrder, + type AuthProfileStore, +} from "./auth-profiles.js"; +import type { FailoverReason } from "./pi-embedded-helpers.js"; + +const log = createSubsystemLogger("agents/unified-runtime-adapter"); + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +/** Configuration for a single runtime slot in the failover chain. */ +export type RuntimeSlot = { + /** Runtime backend to use. */ + runtime: AgentRuntimeKind; + /** Optional provider override for this slot. */ + provider?: string; + /** Optional model override for this slot. */ + model?: string; +}; + +/** Configuration for unified runtime failover. */ +export type UnifiedRuntimeConfig = { + /** Primary runtime to attempt first. */ + primaryRuntime: AgentRuntimeKind; + /** Primary provider (shared across runtimes unless overridden). */ + provider?: string; + /** Primary model (shared across runtimes unless overridden). */ + model?: string; + /** Fallback runtime slots to try when primary fails. */ + fallbackRuntimes?: RuntimeSlot[]; + /** Auth profile IDs to rotate through (applies to all runtimes). */ + authProfiles?: string[]; +}; + +/** A single failover attempt record. */ +export type UnifiedFallbackAttempt = { + runtime: AgentRuntimeKind; + provider: string; + model: string; + profileId?: string; + error: string; + reason?: FailoverReason; + status?: number; + code?: string; +}; + +/** Result from unified runtime execution with failover. */ +export type UnifiedRuntimeResult = { + result: AgentRuntimeResult; + runtime: AgentRuntimeKind; + provider: string; + model: string; + profileId?: string; + attempts: UnifiedFallbackAttempt[]; +}; + +/** Parameters for running with unified failover. */ +export type RunWithUnifiedFallbackParams = { + /** Unified runtime configuration. */ + config: UnifiedRuntimeConfig; + /** Agent directory for auth profile store. */ + agentDir?: string; + /** Moltbot config for auth profile resolution. */ + cfg?: MoltbotConfig; + /** The actual run function for a given runtime/provider/model/profile. */ + run: (params: { + runtime: AgentRuntimeKind; + provider: string; + model: string; + profileId?: string; + }) => Promise; + /** Optional callback for each error during failover. */ + onError?: (attempt: { + runtime: AgentRuntimeKind; + provider: string; + model: string; + profileId?: string; + error: unknown; + attempt: number; + total: number; + }) => void | Promise; +}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function isAbortError(err: unknown): boolean { + if (!err || typeof err !== "object") return false; + if (isFailoverError(err)) return false; + const name = "name" in err ? String(err.name) : ""; + return name === "AbortError"; +} + +function shouldRethrowAbort(err: unknown): boolean { + return isAbortError(err) && !isTimeoutError(err); +} + +function buildRuntimeSlots(config: UnifiedRuntimeConfig): RuntimeSlot[] { + const slots: RuntimeSlot[] = []; + const seen = new Set(); + + const addSlot = (slot: RuntimeSlot) => { + const key = `${slot.runtime}:${slot.provider ?? ""}:${slot.model ?? ""}`; + if (seen.has(key)) return; + seen.add(key); + slots.push(slot); + }; + + // Primary runtime. + addSlot({ + runtime: config.primaryRuntime, + provider: config.provider, + model: config.model, + }); + + // Fallback runtimes. + if (config.fallbackRuntimes) { + for (const fb of config.fallbackRuntimes) { + addSlot({ + runtime: fb.runtime, + provider: fb.provider ?? config.provider, + model: fb.model ?? config.model, + }); + } + } + + return slots; +} + +function resolveProfilesForSlot(params: { + slot: RuntimeSlot; + config: UnifiedRuntimeConfig; + cfg?: MoltbotConfig; + authStore?: AuthProfileStore; +}): string[] { + const { slot, config, cfg, authStore } = params; + + // If explicit profiles provided in config, use those. + if (config.authProfiles && config.authProfiles.length > 0) { + return config.authProfiles; + } + + // Otherwise resolve from config/store. + if (authStore && cfg) { + const provider = slot.provider ?? config.provider ?? ""; + return resolveAuthProfileOrder({ + cfg, + store: authStore, + provider, + }); + } + + return []; +} + +// --------------------------------------------------------------------------- +// Main Entry Point +// --------------------------------------------------------------------------- + +/** + * Run an agent with unified cross-runtime failover. + * + * Attempts the primary runtime first, rotating through auth profiles. + * On persistent failures, falls back to alternate runtimes in the chain. + * + * @example + * ```typescript + * const result = await runWithUnifiedFallback({ + * config: { + * primaryRuntime: "ccsdk", + * fallbackRuntimes: [{ runtime: "pi" }], + * provider: "anthropic", + * model: "claude-sonnet-4-20250514", + * }, + * run: async ({ runtime, provider, model, profileId }) => { + * // Execute the agent run with the specified parameters + * return runtime === "ccsdk" + * ? await ccsdkRuntime.run(...) + * : await piRuntime.run(...); + * }, + * }); + * ``` + */ +export async function runWithUnifiedFallback( + params: RunWithUnifiedFallbackParams, +): Promise { + const { config, cfg, agentDir, run, onError } = params; + + const slots = buildRuntimeSlots(config); + const authStore = cfg + ? ensureAuthProfileStore(agentDir, { allowKeychainPrompt: false }) + : undefined; + + const attempts: UnifiedFallbackAttempt[] = []; + let lastError: unknown; + let attemptNum = 0; + + // Count total possible attempts for progress reporting. + const countTotalAttempts = () => { + let total = 0; + for (const slot of slots) { + const profiles = resolveProfilesForSlot({ slot, config, cfg, authStore }); + total += Math.max(1, profiles.length); + } + return total; + }; + const totalAttempts = countTotalAttempts(); + + for (const slot of slots) { + const provider = slot.provider ?? config.provider ?? ""; + const model = slot.model ?? config.model ?? ""; + const profiles = resolveProfilesForSlot({ slot, config, cfg, authStore }); + + // Filter out profiles in cooldown. + const availableProfiles = + profiles.length > 0 && authStore + ? profiles.filter((id) => !isProfileInCooldown(authStore, id)) + : profiles; + + // If all profiles are in cooldown, record and skip this slot. + if (profiles.length > 0 && availableProfiles.length === 0) { + log.debug(`All profiles for ${slot.runtime}/${provider} are in cooldown, skipping slot`); + attempts.push({ + runtime: slot.runtime, + provider, + model, + error: `All auth profiles for ${provider} are in cooldown`, + reason: "rate_limit", + }); + continue; + } + + // Try each available profile (or once with no profile). + const profilesToTry = availableProfiles.length > 0 ? availableProfiles : [undefined]; + + for (const profileId of profilesToTry) { + attemptNum += 1; + + try { + log.debug( + `Attempting ${slot.runtime}/${provider}/${model}${profileId ? ` (profile: ${profileId})` : ""}`, + ); + + const result = await run({ + runtime: slot.runtime, + provider, + model, + profileId, + }); + + return { + result, + runtime: slot.runtime, + provider, + model, + profileId, + attempts, + }; + } catch (err) { + // Abort errors should propagate immediately (unless they're timeouts). + if (shouldRethrowAbort(err)) { + throw err; + } + + // Attempt to classify the error as a failover reason. + const failoverErr = coerceToFailoverError(err, { + provider, + model, + profileId, + }); + + // If the error is not a failover-eligible error, rethrow. + if (!failoverErr && !isFailoverError(err)) { + throw err; + } + + lastError = failoverErr ?? err; + const described = describeFailoverError(lastError); + + attempts.push({ + runtime: slot.runtime, + provider, + model, + profileId, + error: described.message, + reason: described.reason, + status: described.status, + code: described.code, + }); + + // Put the profile in cooldown if it's a rate limit or auth error. + if (profileId && authStore && described.reason) { + const shouldCooldown = + described.reason === "rate_limit" || + described.reason === "auth" || + described.reason === "billing"; + if (shouldCooldown) { + void markAuthProfileCooldown({ + store: authStore, + profileId, + agentDir, + }); + log.debug(`Put profile ${profileId} in cooldown: ${described.reason}`); + } + } + + // Notify error callback. + await onError?.({ + runtime: slot.runtime, + provider, + model, + profileId, + error: lastError, + attempt: attemptNum, + total: totalAttempts, + }); + + log.debug( + `Attempt ${attemptNum}/${totalAttempts} failed: ${described.message} (${described.reason ?? "unknown"})`, + ); + } + } + } + + // All attempts exhausted. + if (attempts.length === 1 && lastError) { + // Single attempt: throw the original error. + throw lastError; + } + + const summary = + attempts.length > 0 + ? attempts + .map( + (a) => + `${a.runtime}/${a.provider}/${a.model}${a.profileId ? `[${a.profileId}]` : ""}: ${a.error}${ + a.reason ? ` (${a.reason})` : "" + }`, + ) + .join(" | ") + : "unknown"; + + throw new Error(`All runtimes failed (${attempts.length} attempts): ${summary}`, { + cause: lastError instanceof Error ? lastError : undefined, + }); +} + +/** + * Check if a runtime slot is currently available (has non-cooldown profiles). + */ +export function isRuntimeSlotAvailable(params: { + slot: RuntimeSlot; + config: UnifiedRuntimeConfig; + cfg?: MoltbotConfig; + agentDir?: string; +}): boolean { + const { slot, config, cfg, agentDir } = params; + const authStore = cfg + ? ensureAuthProfileStore(agentDir, { allowKeychainPrompt: false }) + : undefined; + + const profiles = resolveProfilesForSlot({ slot, config, cfg, authStore }); + + if (profiles.length === 0) { + // No profiles configured, assume available. + return true; + } + + if (!authStore) { + return true; + } + + return profiles.some((id) => !isProfileInCooldown(authStore, id)); +}