romtuck/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add failover at AgentRuntime layer, fixed up tool bridge

Browse Source
This commit is contained in:
David Garson 2026-01-28 11:14:02 -07:00
parent c8b5337a71
commit b8c947a43e
8 changed files with 885 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/agents/claude-agent-sdk/index.ts
View File

@ -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";

33
src/agents/claude-agent-sdk/sdk-agent-runtime.ts
View File

@ -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,

226
src/agents/claude-agent-sdk/sdk-runner.ts
View File

@ -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<ThinkLevel, number> = {
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<string, unknown> | undefined;
if (!usageObj || typeof usageObj !== "object") {
// Also check for usage nested in data
const data = event.data as Record<string, unknown> | undefined;
const dataUsage = data?.usage as Record<string, unknown> | undefined;
if (!dataUsage || typeof dataUsage !== "object") return undefined;
return normalizeUsageObject(dataUsage);
}
return normalizeUsageObject(usageObj);
}
function normalizeUsageObject(obj: Record<string, unknown>): 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<SdkRunnerRes
name: mcpServerName,
tools,
abortSignal: params.abortSignal,
// Forward tool updates through the agent event stream.
onToolUpdate: (updateParams) => {
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<SdkRunnerRes
sdkOptions.permissionMode = params.permissionMode;
}
// System prompt (with optional conversation history suffix).
const historySuffix = buildHistorySystemPromptSuffix(params.conversationHistory);
// Native session resume (preferred) or history injection fallback.
// When claudeSessionId is provided, use the SDK's native resume feature
// instead of injecting history into the system prompt.
const useNativeSessionResume = Boolean(params.claudeSessionId);
if (useNativeSessionResume) {
sdkOptions.resume = params.claudeSessionId;
if (params.forkSession) {
sdkOptions.forkSession = true;
}
log.debug(`Using native session resume with ID: ${params.claudeSessionId}`);
}
// System prompt (with optional conversation history suffix for non-native resume).
const historySuffix = useNativeSessionResume
? "" // Skip history injection when using native resume
: buildHistorySystemPromptSuffix(params.conversationHistory);
const baseSystemPrompt = params.systemPrompt ?? params.extraSystemPrompt;
if (baseSystemPrompt || historySuffix) {
sdkOptions.systemPrompt = (baseSystemPrompt ?? "") + historySuffix;
@ -369,6 +498,13 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
}) as unknown as Record<string, unknown>;
}
// 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<SdkRunnerRes
let truncated = false;
let resultText: string | undefined;
let aborted = false;
let usage: SdkUsageStats | undefined;
let extractedSessionId: string | undefined;
const chunks: string[] = [];
let assistantSoFar = "";
let didAssistantMessageStart = false;
// Messaging tool tracking state.
const messagingToolSentTexts: string[] = [];
const messagingToolSentTargets: MessagingToolSend[] = [];
// Map of pending tool calls: toolCallId -> { name, args }
const pendingToolCalls = new Map<string, { name: string; args: Record<string, unknown> }>();
// Completed tool calls for session transcript recording.
const completedToolCalls: SdkCompletedToolCall[] = [];
// Set up timeout if configured.
let timeoutId: ReturnType<typeof setTimeout> | undefined;
const timeoutController = new AbortController();
@ -441,7 +587,7 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
const { kind } = classifyEvent(event);
// Emit tool results via callback.
// Emit tool results via callback and track messaging tools.
if (!hooksEnabled && kind === "tool") {
const record = isRecord(event) ? (event as Record<string, unknown>) : undefined;
const type = record && typeof record.type === "string" ? record.type : undefined;
@ -474,6 +620,47 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
: record && typeof record.isError === "boolean"
? record.isError
: Boolean(record?.error);
// Track tool calls for session transcript and messaging deduplication.
if (phase === "start" && toolCallId && record) {
// Extract args from the event (tool_use events typically have input field).
const args =
(record.input as Record<string, unknown>) ??
(record.arguments as Record<string, unknown>) ??
{};
// 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<SdkRunnerRes
if (typeof result === "string") {
resultText = result;
}
// Extract usage from result event.
const resultUsage = extractUsageFromEvent(event);
if (resultUsage) {
usage = resultUsage;
}
// Also check for error results.
const subtype = event.subtype;
if (subtype === "error" && typeof event.error === "string") {
@ -518,6 +710,24 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
break;
}
// Check for usage events (some SDK versions emit these separately).
if (isRecord(event) && (event.type === "usage" || event.type === "message_stop")) {
const eventUsage = extractUsageFromEvent(event);
if (eventUsage) {
usage = eventUsage;
}
}
// Extract session ID from SDK events.
// Session ID appears in system init messages, assistant messages, and result events.
if (!extractedSessionId && isRecord(event)) {
const sessionId = event.session_id;
if (typeof sessionId === "string" && sessionId) {
extractedSessionId = sessionId;
log.debug(`Extracted CCSDK session ID: ${sessionId}`);
}
}
// Extract text from assistant messages.
if (kind === "assistant" || kind === "unknown") {
const text = extractTextFromClaudeAgentSdkEvent(event);
@ -678,6 +888,7 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
extractedChars: 0,
truncated: false,
aborted,
usage,
error: aborted ? undefined : { kind: "no_output", message: "No text output" },
bridge: bridgeResult
? {
@ -728,6 +939,7 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
extractedChars,
truncated,
aborted,
usage,
bridge: bridgeResult
? {
toolCount: bridgeResult.toolCount,
@ -736,6 +948,12 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise<SdkRunnerRes
}
: undefined,
},
didSendViaMessagingTool: messagingToolSentTargets.length > 0,
messagingToolSentTexts: messagingToolSentTexts.length > 0 ? messagingToolSentTexts : undefined,
messagingToolSentTargets:
messagingToolSentTargets.length > 0 ? messagingToolSentTargets : undefined,
completedToolCalls: completedToolCalls.length > 0 ? completedToolCalls : undefined,
claudeSessionId: extractedSessionId,
};
}

132
src/agents/claude-agent-sdk/sdk-session-transcript.ts
View File

@ -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<string, unknown>;
/** 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<string, unknown>;
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,
});
}
}
}

37
src/agents/claude-agent-sdk/tool-bridge.ts
View File

@ -134,12 +134,14 @@ export function convertToolResult(result: AgentToolResult<unknown>): McpCallTool
* - Moltbot: `execute(toolCallId, params, signal?, onUpdate?)`
* - MCP: `handler(args) → Promise<CallToolResult>`
*
* 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<string, unknown>) => Promise<McpCallToolResult> {
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<void>;
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) {

8
src/agents/claude-agent-sdk/tool-bridge.types.ts
View File

@ -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. */

58
src/agents/claude-agent-sdk/types.ts
View File

@ -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<string, unknown>;
/** 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;
};

401
src/agents/unified-runtime-adapter.ts Normal file
View File

@ -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<AgentRuntimeResult>;
/** 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<void>;
};
// ---------------------------------------------------------------------------
// 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<string>();
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<UnifiedRuntimeResult> {
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));
}