From e5f7aa936c4441d2150fd8bc8a85c7e0ca79653c Mon Sep 17 00:00:00 2001 From: David Garson Date: Wed, 28 Jan 2026 11:07:45 -0700 Subject: [PATCH] feat: add ccsdk session adapter --- src/agents/claude-agent-sdk/error-handling.ts | 345 ++++++++++++ src/agents/claude-agent-sdk/index.ts | 28 + .../claude-agent-sdk/sdk-agent-runtime.ts | 106 +++- src/agents/claude-agent-sdk/sdk-runner.ts | 506 +++++++++++++++--- src/agents/claude-agent-sdk/system-prompt.ts | 169 ++++++ src/agents/claude-agent-sdk/types.ts | 72 ++- src/agents/sessions/ccsdk-session-adapter.ts | 502 +++++++++++++++++ src/agents/sessions/index.ts | 56 ++ src/agents/sessions/pi-session-adapter.ts | 377 +++++++++++++ src/agents/sessions/session-adapter.ts | 107 ++++ src/agents/sessions/types.ts | 109 ++++ src/config/zod-schema.agent-defaults.ts | 14 + src/config/zod-schema.agent-runtime.ts | 67 ++- 13 files changed, 2373 insertions(+), 85 deletions(-) create mode 100644 src/agents/claude-agent-sdk/error-handling.ts create mode 100644 src/agents/claude-agent-sdk/system-prompt.ts create mode 100644 src/agents/sessions/ccsdk-session-adapter.ts create mode 100644 src/agents/sessions/index.ts create mode 100644 src/agents/sessions/pi-session-adapter.ts create mode 100644 src/agents/sessions/session-adapter.ts create mode 100644 src/agents/sessions/types.ts diff --git a/src/agents/claude-agent-sdk/error-handling.ts b/src/agents/claude-agent-sdk/error-handling.ts new file mode 100644 index 000000000..9418d65eb --- /dev/null +++ b/src/agents/claude-agent-sdk/error-handling.ts @@ -0,0 +1,345 @@ +/** + * Error handling and retry logic for the Claude Agent SDK. + * + * Provides error classification, retry strategies, and failover support + * for CCSDK agent runs. + */ + +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("agents/claude-agent-sdk/error"); + +/** + * Error classification for CCSDK errors. + */ +export type CcsdkErrorKind = + | "rate_limit" + | "context_overflow" + | "auth_failure" + | "network" + | "timeout" + | "tool_error" + | "unknown"; + +/** + * Error patterns for classification. + */ +const ERROR_PATTERNS: Array<{ pattern: RegExp; kind: CcsdkErrorKind }> = [ + // Rate limiting + { pattern: /rate.?limit|too many requests|429/i, kind: "rate_limit" }, + { pattern: /overloaded|capacity|throttl/i, kind: "rate_limit" }, + { pattern: /resource has been exhausted/i, kind: "rate_limit" }, + + // Context overflow + { pattern: /context.?(?:window|length|overflow)|too.?(?:long|large)/i, kind: "context_overflow" }, + { pattern: /exceeds.*(?:context|token|max)/i, kind: "context_overflow" }, + { pattern: /maximum.*(?:context|token)/i, kind: "context_overflow" }, + { pattern: /prompt.*too.*long/i, kind: "context_overflow" }, + + // Auth failures + { pattern: /auth(?:entication|orization)?.*(?:failed|error|invalid)/i, kind: "auth_failure" }, + { pattern: /invalid.*(?:api.?key|token|credential)/i, kind: "auth_failure" }, + { pattern: /401|403|forbidden|unauthorized/i, kind: "auth_failure" }, + { pattern: /permission.*denied/i, kind: "auth_failure" }, + + // Network errors + { pattern: /network|connection|connect.*(?:error|failed|refused)/i, kind: "network" }, + { pattern: /ECONNREFUSED|ECONNRESET|ENOTFOUND|EHOSTUNREACH/i, kind: "network" }, + { pattern: /socket.*(?:error|closed|hang)/i, kind: "network" }, + { pattern: /fetch.*failed/i, kind: "network" }, + + // Timeout + { pattern: /timeout|timed.?out|deadline.*exceeded/i, kind: "timeout" }, + { pattern: /ETIMEDOUT|ESOCKETTIMEDOUT/i, kind: "timeout" }, + { pattern: /request.*abort/i, kind: "timeout" }, + + // Tool errors + { pattern: /tool.*(?:error|failed|execution)/i, kind: "tool_error" }, + { pattern: /mcp.*(?:error|failed)/i, kind: "tool_error" }, +]; + +/** + * Classify an error into a known category. + */ +export function classifyError(error: unknown): CcsdkErrorKind { + if (!error) return "unknown"; + + // Extract error details + const message = getErrorMessage(error); + const name = getErrorName(error); + const status = getStatusCode(error); + const code = getErrorCode(error); + + // Check status codes first + if (status === 429) return "rate_limit"; + if (status === 401 || status === 403) return "auth_failure"; + if (status === 408) return "timeout"; + + // Check error codes + if (code === "ETIMEDOUT" || code === "ESOCKETTIMEDOUT") return "timeout"; + if (code === "ECONNREFUSED" || code === "ECONNRESET" || code === "ENOTFOUND") return "network"; + + // Check error name + if (name === "TimeoutError") return "timeout"; + if (name === "AbortError") return "timeout"; + + // Check message patterns + const textToCheck = `${message} ${name} ${code ?? ""}`; + for (const { pattern, kind } of ERROR_PATTERNS) { + if (pattern.test(textToCheck)) { + return kind; + } + } + + return "unknown"; +} + +/** + * Extract error message from unknown error. + */ +function getErrorMessage(error: unknown): string { + if (error instanceof Error) return error.message; + if (typeof error === "string") return error; + if (error && typeof error === "object") { + const msg = (error as { message?: unknown }).message; + if (typeof msg === "string") return msg; + } + return ""; +} + +/** + * Extract error name from unknown error. + */ +function getErrorName(error: unknown): string { + if (error instanceof Error) return error.name; + if (error && typeof error === "object") { + const name = (error as { name?: unknown }).name; + if (typeof name === "string") return name; + } + return ""; +} + +/** + * Extract HTTP status code from error. + */ +function getStatusCode(error: unknown): number | undefined { + if (!error || typeof error !== "object") return undefined; + const status = + (error as { status?: unknown }).status ?? (error as { statusCode?: unknown }).statusCode; + if (typeof status === "number") return status; + if (typeof status === "string" && /^\d+$/.test(status)) { + return Number(status); + } + return undefined; +} + +/** + * Extract error code from error. + */ +function getErrorCode(error: unknown): string | undefined { + if (!error || typeof error !== "object") return undefined; + const code = (error as { code?: unknown }).code; + if (typeof code === "string" && code.trim()) { + return code.trim(); + } + return undefined; +} + +/** + * Retry options. + */ +export type RetryOptions = { + /** Maximum number of retry attempts. */ + maxRetries: number; + /** Base backoff delay in milliseconds. */ + backoffMs: number; + /** Maximum backoff delay in milliseconds. */ + maxBackoffMs?: number; + /** Backoff multiplier (default: 2 for exponential backoff). */ + backoffMultiplier?: number; + /** Error kinds that should trigger a retry. */ + retryOn: CcsdkErrorKind[]; + /** Optional abort signal. */ + abortSignal?: AbortSignal; + /** Optional callback for retry events. */ + onRetry?: (attempt: number, error: unknown, delayMs: number) => void; +}; + +/** + * Default retry options. + */ +export const DEFAULT_RETRY_OPTIONS: Partial = { + maxRetries: 3, + backoffMs: 1000, + maxBackoffMs: 30_000, + backoffMultiplier: 2, + retryOn: ["rate_limit", "network", "timeout"], +}; + +/** + * Execute a function with retry logic. + */ +export async function withRetry(fn: () => Promise, options: RetryOptions): Promise { + const maxRetries = options.maxRetries; + const backoffMs = options.backoffMs; + const maxBackoffMs = options.maxBackoffMs ?? 30_000; + const multiplier = options.backoffMultiplier ?? 2; + const retryOn = new Set(options.retryOn); + + let lastError: unknown; + let delay = backoffMs; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + // Check for abort before each attempt + if (options.abortSignal?.aborted) { + throw new Error("Aborted"); + } + + try { + return await fn(); + } catch (error) { + lastError = error; + const errorKind = classifyError(error); + + // Don't retry non-retryable errors + if (!retryOn.has(errorKind)) { + throw error; + } + + // Don't retry if we've exhausted attempts + if (attempt >= maxRetries) { + log.warn(`All ${maxRetries} retry attempts exhausted`, { + errorKind, + lastError: getErrorMessage(error), + }); + throw error; + } + + // Calculate delay with jitter + const jitter = Math.random() * 0.2 * delay; // 0-20% jitter + const effectiveDelay = Math.min(delay + jitter, maxBackoffMs); + + log.debug(`Retry attempt ${attempt + 1}/${maxRetries} in ${Math.round(effectiveDelay)}ms`, { + errorKind, + error: getErrorMessage(error), + }); + + // Notify callback + options.onRetry?.(attempt + 1, error, effectiveDelay); + + // Wait before retry + await sleep(effectiveDelay, options.abortSignal); + + // Increase delay for next attempt + delay = Math.min(delay * multiplier, maxBackoffMs); + } + } + + throw lastError; +} + +/** + * Sleep for a duration, respecting abort signal. + */ +async function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new Error("Aborted")); + return; + } + + const timer = setTimeout(resolve, ms); + + if (signal) { + const onAbort = () => { + clearTimeout(timer); + reject(new Error("Aborted")); + }; + signal.addEventListener("abort", onAbort, { once: true }); + } + }); +} + +/** + * Classify error and determine if it's retryable. + */ +export function isRetryableError( + error: unknown, + retryableKinds: CcsdkErrorKind[] = ["rate_limit", "network", "timeout"], +): boolean { + const kind = classifyError(error); + return retryableKinds.includes(kind); +} + +/** + * Get a human-readable description of an error kind. + */ +export function describeErrorKind(kind: CcsdkErrorKind): string { + switch (kind) { + case "rate_limit": + return "Rate limit exceeded"; + case "context_overflow": + return "Context window overflow"; + case "auth_failure": + return "Authentication failed"; + case "network": + return "Network error"; + case "timeout": + return "Request timed out"; + case "tool_error": + return "Tool execution error"; + case "unknown": + default: + return "Unknown error"; + } +} + +/** + * CCSDK-specific error with classification. + */ +export class CcsdkError extends Error { + readonly kind: CcsdkErrorKind; + readonly status?: number; + readonly code?: string; + + constructor( + message: string, + kind: CcsdkErrorKind, + options?: { + status?: number; + code?: string; + cause?: unknown; + }, + ) { + super(message, { cause: options?.cause }); + this.name = "CcsdkError"; + this.kind = kind; + this.status = options?.status; + this.code = options?.code; + } +} + +/** + * Check if an error is a CcsdkError. + */ +export function isCcsdkError(error: unknown): error is CcsdkError { + return error instanceof CcsdkError; +} + +/** + * Coerce an unknown error to a CcsdkError. + */ +export function toCcsdkError(error: unknown): CcsdkError { + if (isCcsdkError(error)) return error; + + const kind = classifyError(error); + const message = getErrorMessage(error) || describeErrorKind(kind); + const status = getStatusCode(error); + const code = getErrorCode(error); + + return new CcsdkError(message, kind, { + status, + code, + cause: error instanceof Error ? error : undefined, + }); +} diff --git a/src/agents/claude-agent-sdk/index.ts b/src/agents/claude-agent-sdk/index.ts index 9991d63b3..a79814409 100644 --- a/src/agents/claude-agent-sdk/index.ts +++ b/src/agents/claude-agent-sdk/index.ts @@ -47,8 +47,36 @@ export type { SdkEventType, SdkProviderConfig, SdkProviderEnv, + SdkReasoningLevel, + SdkRunnerCallbacks, SdkRunnerParams, SdkTextEvent, SdkToolResultEvent, SdkToolUseEvent, + SdkVerboseLevel, } from "./types.js"; + +// Error handling +export { + classifyError, + CcsdkError, + describeErrorKind, + isCcsdkError, + isRetryableError, + toCcsdkError, + withRetry, + DEFAULT_RETRY_OPTIONS, + type CcsdkErrorKind, + type RetryOptions, +} from "./error-handling.js"; + +// System prompt +export { + buildSystemPromptAdditions, + buildSystemPromptAdditionsFromParams, + clearPromptEnrichers, + getEnricherCount, + registerPromptEnricher, + type PromptEnricher, + type PromptEnricherContext, +} from "./system-prompt.js"; diff --git a/src/agents/claude-agent-sdk/sdk-agent-runtime.ts b/src/agents/claude-agent-sdk/sdk-agent-runtime.ts index a3b52e466..e28dac05f 100644 --- a/src/agents/claude-agent-sdk/sdk-agent-runtime.ts +++ b/src/agents/claude-agent-sdk/sdk-agent-runtime.ts @@ -7,6 +7,8 @@ import type { MoltbotConfig } from "../../config/config.js"; import type { AgentRuntime, AgentRuntimeRunParams, AgentRuntimeResult } from "../agent-runtime.js"; import type { AgentCcSdkConfig } from "../../config/types.agents.js"; +import type { ThinkLevel, VerboseLevel } from "../../auto-reply/thinking.js"; +import type { SdkReasoningLevel, SdkVerboseLevel } from "./types.js"; import { runSdkAgent } from "./sdk-runner.js"; import { resolveProviderConfig } from "./provider-config.js"; import { isSdkAvailable } from "./sdk-loader.js"; @@ -27,6 +29,77 @@ export type CcSdkAgentRuntimeContext = { baseUrl?: string; }; +/** + * Map ThinkLevel to SDK reasoning level. + */ +function mapThinkLevel(thinkLevel?: ThinkLevel): SdkReasoningLevel { + switch (thinkLevel) { + case "off": + return "off"; + case "minimal": + return "minimal"; + case "low": + return "low"; + case "medium": + return "medium"; + case "high": + case "xhigh": + return "high"; + default: + return "off"; + } +} + +/** + * Map VerboseLevel to SDK verbose level. + */ +function mapVerboseLevel(verboseLevel?: VerboseLevel): SdkVerboseLevel { + switch (verboseLevel) { + case "off": + return "off"; + case "on": + return "on"; + case "full": + return "full"; + default: + return "off"; + } +} + +/** + * Extract agent ID from session key. + */ +function extractAgentId(sessionKey?: string): string { + if (!sessionKey) return "main"; + const parts = sessionKey.split(":"); + return parts[0] || "main"; +} + +/** + * Resolve user timezone from config. + */ +function resolveTimezone(config?: MoltbotConfig): string | undefined { + // Check for explicit timezone in config + const tz = config?.agents?.defaults?.userTimezone; + if (tz) return tz; + + // Fallback to environment + return typeof process !== "undefined" ? process.env.TZ : undefined; +} + +/** + * Extract skill names from snapshot. + */ +function extractSkillNames( + skillsSnapshot?: AgentRuntimeRunParams["skillsSnapshot"], +): string[] | undefined { + if (!skillsSnapshot?.resolvedSkills) return undefined; + const names = skillsSnapshot.resolvedSkills + .map((s) => s.name) + .filter((n): n is string => Boolean(n)); + return names.length > 0 ? names : undefined; +} + /** * Create a Claude Code SDK runtime instance. * @@ -55,29 +128,58 @@ export function createCcSdkAgentRuntime(context?: CcSdkAgentRuntimeContext): Age displayName: `Claude Code SDK (${providerConfig.name})`, async run(params: AgentRuntimeRunParams): Promise { + const effectiveConfig = params.config ?? context?.config; + const agentId = extractAgentId(params.sessionKey); + log.debug("CCSDK runtime run", { sessionId: params.sessionId, runId: params.runId, provider: providerConfig.name, + agentId, + thinkLevel: params.thinkLevel, + verboseLevel: params.verboseLevel, }); return runSdkAgent({ + // ─── Session & Identity ────────────────────────────────────────────── sessionId: params.sessionId, sessionKey: params.sessionKey, sessionFile: params.sessionFile, workspaceDir: params.workspaceDir, agentDir: params.agentDir, - config: params.config ?? context?.config, + agentId, + + // ─── Configuration ─────────────────────────────────────────────────── + config: effectiveConfig, prompt: params.prompt, - model: params.model ? `${params.provider}/${params.model}` : undefined, + model: params.model ? `${params.provider ?? "anthropic"}/${params.model}` : undefined, providerConfig, timeoutMs: params.timeoutMs, runId: params.runId, abortSignal: params.abortSignal, + + // ─── Model Behavior ────────────────────────────────────────────────── + reasoningLevel: mapThinkLevel(params.thinkLevel), + verboseLevel: mapVerboseLevel(params.verboseLevel), + + // ─── System Prompt Context ─────────────────────────────────────────── extraSystemPrompt: params.extraSystemPrompt, + timezone: resolveTimezone(effectiveConfig), + messageChannel: params.messageChannel, + skills: extractSkillNames(params.skillsSnapshot), + + // ─── SDK Options ───────────────────────────────────────────────────── hooksEnabled: context?.ccsdkConfig?.hooksEnabled, sdkOptions: context?.ccsdkConfig?.options, modelTiers: context?.ccsdkConfig?.models, + + // ─── Streaming Callbacks ───────────────────────────────────────────── + onPartialReply: params.onPartialReply, + onAssistantMessageStart: params.onAssistantMessageStart, + onBlockReply: params.onBlockReply, + onBlockReplyFlush: params.onBlockReplyFlush, + onReasoningStream: params.onReasoningStream, + onToolResult: params.onToolResult, onAgentEvent: params.onAgentEvent, }); }, diff --git a/src/agents/claude-agent-sdk/sdk-runner.ts b/src/agents/claude-agent-sdk/sdk-runner.ts index c30e2913a..6f731d5cc 100644 --- a/src/agents/claude-agent-sdk/sdk-runner.ts +++ b/src/agents/claude-agent-sdk/sdk-runner.ts @@ -2,7 +2,7 @@ * Claude Agent SDK runner. * * Executes agent turns using the Claude Agent SDK, handling authentication, - * event streaming, and result adaptation. + * event streaming, hooks, and result adaptation. */ import type { SdkRunnerParams, SdkProviderEnv } from "./types.js"; @@ -10,6 +10,14 @@ import type { CcSdkModelTiers } from "../../config/types.agents.js"; import type { AgentRuntimeResult } from "../agent-runtime.js"; import { loadClaudeAgentSdk, type ClaudeAgentSdkModule } from "./sdk-loader.js"; import { resolveProviderConfig } from "./provider-config.js"; +import { buildSystemPromptAdditionsFromParams } from "./system-prompt.js"; +import { + classifyError, + withRetry, + DEFAULT_RETRY_OPTIONS, + type CcsdkErrorKind, +} from "./error-handling.js"; +import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; const log = createSubsystemLogger("agents/claude-agent-sdk"); @@ -36,30 +44,83 @@ function buildModelTierEnv(modelTiers?: CcSdkModelTiers): SdkProviderEnv { return env; } +/** + * Internal result type from SDK execution. + */ +type SdkInternalResult = { + texts: string[]; + usage?: { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + total?: number; + }; + aborted?: boolean; + error?: string; +}; + +/** + * SDK event handler for streaming callbacks. + */ +type SdkEventHandler = { + onText?: (text: string) => void; + onThinking?: (text: string) => void; + onToolUse?: (toolName: string, toolId: string, input: unknown) => void; + onToolResult?: (toolId: string, result: string, isError?: boolean) => void; + onError?: (error: string) => void; + onDone?: (usage?: SdkInternalResult["usage"]) => void; +}; + /** * Run an agent turn using the Claude Agent SDK. * * This function: - * 1. Loads the SDK dynamically - * 2. Configures authentication based on available credentials - * 3. Executes the agent turn with the given prompt - * 4. Streams events back to the caller - * 5. Returns results in the standard AgentRuntimeResult format + * 1. Runs before_agent_start hooks to allow context injection + * 2. Loads the SDK dynamically + * 3. Configures authentication based on available credentials + * 4. Builds system prompt additions for Moltbot-specific context + * 5. Executes the agent turn with the given prompt + * 6. Streams events back to the caller via callbacks + * 7. Runs agent_end hooks for post-processing + * 8. Returns results in the standard AgentRuntimeResult format */ export async function runSdkAgent(params: SdkRunnerParams): Promise { const started = Date.now(); + const hookRunner = getGlobalHookRunner(); // Emit lifecycle start event params.onAgentEvent?.({ stream: "lifecycle", data: { phase: "start", - runtime: "sdk", + runtime: "ccsdk", sessionId: params.sessionId, runId: params.runId, }, }); + // ─── Run before_agent_start hooks ────────────────────────────────────────── + let effectivePrompt = params.prompt; + if (hookRunner?.hasHooks("before_agent_start")) { + try { + const hookResult = await hookRunner.runBeforeAgentStart( + { prompt: params.prompt, messages: [] }, + { + agentId: params.agentId ?? params.sessionKey?.split(":")[0] ?? "main", + sessionKey: params.sessionKey, + workspaceDir: params.workspaceDir, + }, + ); + if (hookResult?.prependContext) { + effectivePrompt = `${hookResult.prependContext}\n\n${params.prompt}`; + log.debug(`hooks: prepended context to prompt (${hookResult.prependContext.length} chars)`); + } + } catch (hookErr) { + log.warn(`before_agent_start hook failed: ${String(hookErr)}`); + } + } + try { // Load the SDK const sdk = await loadClaudeAgentSdk(); @@ -72,6 +133,7 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { - // Forward events to the caller - params.onAgentEvent?.({ - stream: "sdk", - data: event as Record, - }); - }, + // Build system prompt additions + const systemPromptAdditions = buildSystemPromptAdditionsFromParams({ + agentId: params.agentId, + sessionKey: params.sessionKey, + workspaceDir: params.workspaceDir, + timezone: params.timezone, + messageChannel: params.messageChannel, + channelHints: params.channelHints, + skills: params.skills, }); + // Combine with any legacy extra system prompt + const combinedSystemPrompt = + [systemPromptAdditions, params.extraSystemPrompt].filter(Boolean).join("\n\n") || undefined; + + // Build event handler for streaming callbacks + const eventHandler = buildEventHandler(params); + + // Run the agent with retry logic + const result = await withRetry( + () => + runSdkAgentInternal( + sdk, + effectivePrompt, + { + model: params.model, + workingDirectory: params.workspaceDir, + sessionId: params.sessionId, + systemPrompt: combinedSystemPrompt, + timeout: params.timeoutMs, + signal: params.abortSignal, + reasoningLevel: params.reasoningLevel, + maxOutputTokens: params.maxOutputTokens, + maxThinkingTokens: params.maxThinkingTokens, + ...params.sdkOptions, + }, + eventHandler, + ), + { + ...DEFAULT_RETRY_OPTIONS, + maxRetries: 2, + backoffMs: 1000, + retryOn: ["rate_limit", "network", "timeout"] as CcsdkErrorKind[], + abortSignal: params.abortSignal, + onRetry: (attempt, error, delayMs) => { + log.warn(`SDK agent retry attempt ${attempt}`, { + error: error instanceof Error ? error.message : String(error), + delayMs, + sessionId: params.sessionId, + }); + params.onAgentEvent?.({ + stream: "lifecycle", + data: { phase: "retry", attempt, delayMs }, + }); + }, + }, + ); + // Build result const agentResult: AgentRuntimeResult = { payloads: result.texts.map((text) => ({ text })), @@ -135,13 +232,34 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { + log.warn(`agent_end hook failed: ${err}`); + }); + } + return agentResult; } finally { // Restore original environment @@ -159,9 +277,11 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { + log.warn(`agent_end hook failed: ${err}`); + }); + } + // Return error result return { payloads: [ @@ -193,84 +335,292 @@ export async function runSdkAgent(params: SdkRunnerParams): Promise { + // Signal assistant message start + if (!assistantStarted) { + assistantStarted = true; + params.onAssistantMessageStart?.(); + } + + // Stream partial reply + params.onPartialReply?.({ text }); + + // Forward as block reply if configured + params.onBlockReply?.({ text }); + }, + + onThinking: (text) => { + params.onReasoningStream?.({ text }); + }, + + onToolUse: (toolName, toolId, input) => { + params.onAgentEvent?.({ + stream: "tool", + data: { + phase: "start", + toolName, + toolId, + input, + }, + }); + }, + + onToolResult: (toolId, result, isError) => { + // Check if we should emit this tool result + const toolName = "unknown"; // Would need to track from onToolUse + if (params.shouldEmitToolResult && !params.shouldEmitToolResult(toolName)) { + return; + } + + params.onToolResult?.({ text: result }); + params.onAgentEvent?.({ + stream: "tool", + data: { + phase: "end", + toolId, + result: params.shouldEmitToolOutput?.(toolName) ? result : undefined, + isError, + }, + }); + }, + + onError: (error) => { + params.onAgentEvent?.({ + stream: "error", + data: { message: error }, + }); + }, + + onDone: (usage) => { + // Flush any pending block replies + params.onBlockReplyFlush?.(); + + params.onAgentEvent?.({ + stream: "done", + data: { usage }, + }); + }, + }; +} + /** * Internal SDK execution function. * - * This is a placeholder that needs to be implemented based on the actual - * Claude Agent SDK API. The SDK provides different interfaces for: - * - Simple completions - * - Agentic tool use - * - Streaming responses + * Calls the Claude Agent SDK's query function and processes the streaming response. + * The SDK handles: + * - Session resumption (via sessionId) + * - Workspace context building + * - Tool execution + * - Conversation history */ async function runSdkAgentInternal( sdk: ClaudeAgentSdkModule, - _prompt: string, + prompt: string, options: { model?: string; workingDirectory?: string; + sessionId?: string; systemPrompt?: string; timeout?: number; signal?: AbortSignal; + reasoningLevel?: string; + maxOutputTokens?: number; + maxThinkingTokens?: number; + [key: string]: unknown; }, - _callbacks: { - onEvent?: (event: unknown) => void; - }, -): Promise<{ - texts: string[]; - usage?: { - input?: number; - output?: number; - cacheRead?: number; - cacheWrite?: number; - total?: number; - }; - aborted?: boolean; -}> { - // Placeholder implementation - needs to be updated based on actual SDK API - // The SDK likely provides something like: - // - sdk.createAgent() or sdk.Agent class - // - agent.run(prompt) or agent.chat(prompt) - // - Event streaming via async iterators or callbacks - + eventHandler: SdkEventHandler, +): Promise { const texts: string[] = []; - const usage: - | { - input?: number; - output?: number; - cacheRead?: number; - cacheWrite?: number; - total?: number; - } - | undefined = undefined; + let usage: SdkInternalResult["usage"]; let aborted = false; + let error: string | undefined; try { - // Check if SDK has the expected interface - // This needs to be updated based on actual SDK API - if ("Claude" in sdk || "createAgent" in sdk || "Agent" in sdk) { - // TODO: Implement actual SDK call based on SDK's public API - // For now, throw to indicate SDK integration is not yet complete + // The Claude Agent SDK exports a `query` function that returns an async iterator + // of events. The exact API shape may vary by SDK version. + const queryFn = (sdk as { query?: unknown }).query; + + if (typeof queryFn !== "function") { + // Check for alternative API shapes + const claudeFn = (sdk as { claude?: unknown }).claude; + const agentFn = (sdk as { agent?: unknown }).agent; + + if (typeof claudeFn === "function" || typeof agentFn === "function") { + throw new Error( + "Found alternative SDK API. Please update runSdkAgentInternal to use the correct interface.", + ); + } + throw new Error( - "Claude Agent SDK integration not yet implemented. " + - "Please check SDK documentation for the correct API.", + "Claude Agent SDK query function not found. " + + "Ensure @anthropic-ai/claude-agent-sdk is installed correctly.", ); } - // Fallback: check for simple completion API - throw new Error("Unable to find compatible API in Claude Agent SDK"); - } catch (error) { + // Build query options + const queryOptions = { + prompt, + options: { + cwd: options.workingDirectory, + model: options.model, + systemPrompt: options.systemPrompt, + maxTurns: 50, // Reasonable default + resume: options.sessionId, // Resume existing session + abortController: options.signal ? { signal: options.signal } : undefined, + // Thinking/reasoning configuration + ...(options.reasoningLevel && options.reasoningLevel !== "off" + ? { + thinking: { + type: "enabled", + budgetTokens: options.maxThinkingTokens ?? 10000, + }, + } + : {}), + ...(options.maxOutputTokens ? { maxTokens: options.maxOutputTokens } : {}), + }, + }; + + // Execute query and process events + const response = await ( + queryFn as (opts: typeof queryOptions) => Promise> + )(queryOptions); + + // Process streaming events + for await (const event of response) { + if (options.signal?.aborted) { + aborted = true; + break; + } + + processEvent(event, texts, eventHandler, (u) => { + usage = u; + }); + } + } catch (err) { if (options.signal?.aborted) { aborted = true; - texts.push("Agent run was aborted."); } else { - throw error; + error = err instanceof Error ? err.message : String(err); + eventHandler.onError?.(error); + throw err; } } - return { texts, usage, aborted }; + // Notify completion + eventHandler.onDone?.(usage); + + // Ensure we have at least one text response + if (texts.length === 0 && !aborted && !error) { + texts.push("(No response from agent)"); + } + + return { texts, usage, aborted, error }; +} + +/** + * Process a single SDK event. + */ +function processEvent( + event: unknown, + texts: string[], + handler: SdkEventHandler, + setUsage: (u: SdkInternalResult["usage"]) => void, +): void { + if (!event || typeof event !== "object") return; + + const evt = event as Record; + const type = evt.type as string | undefined; + + switch (type) { + case "text": + case "content_block_delta": { + const text = (evt.text as string) ?? (evt.delta as { text?: string })?.text; + if (text) { + texts.push(text); + handler.onText?.(text); + } + break; + } + + case "thinking": + case "thinking_delta": { + const thinking = (evt.thinking as string) ?? (evt.delta as { thinking?: string })?.thinking; + if (thinking) { + handler.onThinking?.(thinking); + } + break; + } + + case "tool_use": + case "tool_use_start": { + const name = evt.name as string; + const id = evt.id as string; + const input = evt.input; + if (name && id) { + handler.onToolUse?.(name, id, input); + } + break; + } + + case "tool_result": { + const toolId = (evt.tool_use_id as string) ?? (evt.id as string); + const content = evt.content as string; + const isError = evt.is_error as boolean; + if (toolId) { + handler.onToolResult?.(toolId, content ?? "", isError); + } + break; + } + + case "message_stop": + case "done": + case "result": { + const eventUsage = evt.usage as + | { + input_tokens?: number; + output_tokens?: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + } + | undefined; + if (eventUsage) { + setUsage({ + input: eventUsage.input_tokens, + output: eventUsage.output_tokens, + cacheRead: eventUsage.cache_read_input_tokens, + cacheWrite: eventUsage.cache_creation_input_tokens, + }); + } + // Extract final text if present + const resultText = (evt.result as string) ?? (evt.text as string); + if (resultText && !texts.includes(resultText)) { + texts.push(resultText); + handler.onText?.(resultText); + } + break; + } + + case "error": { + const errorMsg = + (evt.error as { message?: string })?.message ?? (evt.message as string) ?? "Unknown error"; + handler.onError?.(errorMsg); + break; + } + } } diff --git a/src/agents/claude-agent-sdk/system-prompt.ts b/src/agents/claude-agent-sdk/system-prompt.ts new file mode 100644 index 000000000..f8df250a5 --- /dev/null +++ b/src/agents/claude-agent-sdk/system-prompt.ts @@ -0,0 +1,169 @@ +/** + * System prompt additions for Claude Code SDK. + * + * The SDK builds its own comprehensive system prompt with workspace context, + * tools, git status, etc. This module provides a small extension point for + * adding Moltbot-specific context the SDK can't know about. + * + * These additions are **prepended** to the SDK's native system prompt. + */ + +/** + * Context available to prompt enrichers. + */ +export type PromptEnricherContext = { + /** Moltbot agent ID (e.g., "main", "work"). */ + agentId?: string; + /** Session key for routing. */ + sessionKey?: string; + /** Agent workspace directory. */ + workspaceDir?: string; + /** User's timezone (e.g., "America/New_York"). */ + timezone?: string; + /** Messaging channel (e.g., "telegram", "slack", "whatsapp"). */ + channel?: string; + /** Channel-specific hints or instructions. */ + channelHints?: string; + /** Available Moltbot skills. */ + skills?: string[]; + /** Additional custom context. */ + custom?: Record; +}; + +/** + * A prompt enricher adds context to the system prompt. + * + * Returns a string to add, or undefined/null to skip. + */ +export type PromptEnricher = (context: PromptEnricherContext) => string | undefined | null; + +/** + * Registry of prompt enrichers. + */ +const enrichers: PromptEnricher[] = []; + +/** + * Register a prompt enricher. + * + * Enrichers are called in registration order. Each can add a line or + * paragraph of context to the system prompt additions. + */ +export function registerPromptEnricher(enricher: PromptEnricher): void { + enrichers.push(enricher); +} + +/** + * Clear all registered enrichers (for testing). + */ +export function clearPromptEnrichers(): void { + enrichers.length = 0; +} + +/** + * Get count of registered enrichers (for testing). + */ +export function getEnricherCount(): number { + return enrichers.length; +} + +// ─── Built-in Enrichers ────────────────────────────────────────────────────── + +/** + * Agent identity enricher. + */ +const agentIdentityEnricher: PromptEnricher = (ctx) => { + if (!ctx.agentId) return null; + return `You are Moltbot agent "${ctx.agentId}".`; +}; + +/** + * Timezone enricher. + */ +const timezoneEnricher: PromptEnricher = (ctx) => { + if (!ctx.timezone) return null; + return `User timezone: ${ctx.timezone}`; +}; + +/** + * Channel hints enricher. + */ +const channelEnricher: PromptEnricher = (ctx) => { + if (!ctx.channel && !ctx.channelHints) return null; + + const parts: string[] = []; + if (ctx.channel) { + parts.push(`Messaging channel: ${ctx.channel}`); + } + if (ctx.channelHints) { + parts.push(ctx.channelHints); + } + return parts.join("\n"); +}; + +/** + * Skills enricher. + */ +const skillsEnricher: PromptEnricher = (ctx) => { + if (!ctx.skills?.length) return null; + return `Available Moltbot skills: ${ctx.skills.join(", ")}`; +}; + +// Register built-in enrichers +registerPromptEnricher(agentIdentityEnricher); +registerPromptEnricher(timezoneEnricher); +registerPromptEnricher(channelEnricher); +registerPromptEnricher(skillsEnricher); + +// ─── Main API ──────────────────────────────────────────────────────────────── + +/** + * Build system prompt additions for the Claude Code SDK. + * + * Runs all registered enrichers and combines their output. + * Returns undefined if no additions are needed. + */ +export function buildSystemPromptAdditions(context: PromptEnricherContext): string | undefined { + const parts: string[] = []; + + for (const enricher of enrichers) { + try { + const result = enricher(context); + if (result) { + parts.push(result.trim()); + } + } catch { + // Skip failed enrichers silently + } + } + + if (parts.length === 0) { + return undefined; + } + + return parts.join("\n\n"); +} + +/** + * Build system prompt additions from run parameters. + * + * Convenience wrapper that extracts context from common param shapes. + */ +export function buildSystemPromptAdditionsFromParams(params: { + agentId?: string; + sessionKey?: string; + workspaceDir?: string; + timezone?: string; + messageChannel?: string; + channelHints?: string; + skills?: string[]; +}): string | undefined { + return buildSystemPromptAdditions({ + agentId: params.agentId ?? params.sessionKey?.split(":")[0], + sessionKey: params.sessionKey, + workspaceDir: params.workspaceDir, + timezone: params.timezone, + channel: params.messageChannel, + channelHints: params.channelHints, + skills: params.skills, + }); +} diff --git a/src/agents/claude-agent-sdk/types.ts b/src/agents/claude-agent-sdk/types.ts index 6005040dc..bde557727 100644 --- a/src/agents/claude-agent-sdk/types.ts +++ b/src/agents/claude-agent-sdk/types.ts @@ -35,8 +35,38 @@ export type SdkProviderConfig = { model?: string; }; +/** Reasoning/thinking level for the agent. */ +export type SdkReasoningLevel = "off" | "minimal" | "low" | "medium" | "high"; + +/** Verbose output level for tool results. */ +export type SdkVerboseLevel = "off" | "on" | "full"; + +/** Streaming callbacks for SDK runner. */ +export type SdkRunnerCallbacks = { + /** Called when the assistant message starts. */ + onAssistantMessageStart?: () => void | Promise; + /** Called when a partial reply chunk is available. */ + onPartialReply?: (payload: { text?: string; mediaUrls?: string[] }) => void | Promise; + /** Called for reasoning/thinking stream events. */ + onReasoningStream?: (payload: { text?: string }) => void | Promise; + /** Called for block-level reply delivery. */ + onBlockReply?: (payload: { + text?: string; + mediaUrls?: string[]; + audioAsVoice?: boolean; + replyToId?: string; + }) => void | Promise; + /** Called when block replies should be flushed. */ + onBlockReplyFlush?: () => void | Promise; + /** Called when a tool result is available. */ + onToolResult?: (payload: { text?: string; mediaUrls?: string[] }) => void | Promise; + /** Called for agent lifecycle events. */ + onAgentEvent?: (evt: { stream: string; data: Record }) => void; +}; + /** Parameters for SDK runner execution. */ export type SdkRunnerParams = { + // ─── Session & Identity ──────────────────────────────────────────────────── /** Session identifier. */ sessionId: string; /** Session key for routing. */ @@ -47,6 +77,10 @@ export type SdkRunnerParams = { workspaceDir: string; /** Agent data directory. */ agentDir?: string; + /** Moltbot agent ID (e.g., "main", "work"). */ + agentId?: string; + + // ─── Configuration ───────────────────────────────────────────────────────── /** Moltbot configuration. */ config?: MoltbotConfig; /** User prompt/message. */ @@ -61,17 +95,47 @@ export type SdkRunnerParams = { runId: string; /** Abort signal for cancellation. */ abortSignal?: AbortSignal; - /** Extra system prompt to append. */ + + // ─── Model Behavior ──────────────────────────────────────────────────────── + /** Reasoning/thinking level. */ + reasoningLevel?: SdkReasoningLevel; + /** Verbose level for tool output. */ + verboseLevel?: SdkVerboseLevel; + /** Maximum output tokens. */ + maxOutputTokens?: number; + /** Maximum thinking tokens (when reasoning is enabled). */ + maxThinkingTokens?: number; + + // ─── Tools ───────────────────────────────────────────────────────────────── + /** Allow elevated bash commands. */ + bashElevated?: boolean; + /** Tools to disable. */ + disableTools?: string[]; + /** Filter for emitting tool results. */ + shouldEmitToolResult?: (toolName: string) => boolean; + /** Filter for emitting tool output. */ + shouldEmitToolOutput?: (toolName: string) => boolean; + + // ─── System Prompt Context ───────────────────────────────────────────────── + /** Extra system prompt to append (legacy). */ extraSystemPrompt?: string; + /** User's timezone (e.g., "America/New_York"). */ + timezone?: string; + /** Messaging channel (e.g., "telegram", "slack"). */ + messageChannel?: string; + /** Channel-specific hints. */ + channelHints?: string; + /** Available Moltbot skills. */ + skills?: string[]; + + // ─── SDK Options ─────────────────────────────────────────────────────────── /** Enable Claude Code hooks. */ hooksEnabled?: boolean; /** Additional SDK options. */ sdkOptions?: Record; /** 3-tier model configuration (haiku/sonnet/opus). */ modelTiers?: CcSdkModelTiers; - /** Called for agent lifecycle events. */ - onAgentEvent?: (evt: { stream: string; data: Record }) => void; -}; +} & SdkRunnerCallbacks; /** SDK event types from the Claude Agent SDK. */ export type SdkEventType = diff --git a/src/agents/sessions/ccsdk-session-adapter.ts b/src/agents/sessions/ccsdk-session-adapter.ts new file mode 100644 index 000000000..2f496d84a --- /dev/null +++ b/src/agents/sessions/ccsdk-session-adapter.ts @@ -0,0 +1,502 @@ +/** + * Claude Code SDK session adapter. + * + * Reads and writes Claude Code SDK JSONL format with tree-structured messages. + * + * CCSDK JSONL Format: + * ```jsonl + * { + * "type": "assistant", + * "parentUuid": "...", + * "uuid": "...", + * "sessionId": "...", + * "cwd": "/path/to/workspace", + * "version": "2.1.22", + * "gitBranch": "...", + * "slug": "...", + * "timestamp": "...", + * "message": { + * "role": "assistant", + * "content": [ + * { "type": "text", "text": "..." }, + * { "type": "tool_use", "id": "...", "name": "...", "input": {} }, + * { "type": "thinking", "thinking": "..." } + * ], + * "usage": { "input_tokens": ..., "output_tokens": ... } + * } + * } + * ``` + * + * Key differences from Pi-Agent: + * - Tree structure (parentUuid/uuid) vs flat list + * - Rich metadata envelope (cwd, gitBranch, slug, version) + * - tool_use in content (not toolCall) + * - Tool results inline in next message + * - Thinking blocks preserved + */ + +import fs from "node:fs/promises"; +import readline from "node:readline"; +import { createReadStream } from "node:fs"; + +import type { SessionAdapter } from "./session-adapter.js"; +import type { + AssistantContent, + NormalizedContent, + NormalizedImageContent, + NormalizedMessage, + NormalizedToolResultContent, + SessionMetadata, + UsageInfo, +} from "./types.js"; + +/** + * CCSDK content block types. + */ +type CcsdkTextContent = { type: "text"; text: string }; +type CcsdkImageContent = { + type: "image"; + source: { type: "base64"; media_type: string; data: string }; +}; +type CcsdkToolUse = { + type: "tool_use"; + id: string; + name: string; + input: Record; +}; +type CcsdkToolResult = { + type: "tool_result"; + tool_use_id: string; + content: string | Array; + is_error?: boolean; +}; +type CcsdkThinking = { type: "thinking"; thinking: string }; + +type CcsdkContent = + | CcsdkTextContent + | CcsdkImageContent + | CcsdkToolUse + | CcsdkToolResult + | CcsdkThinking; + +/** + * CCSDK message structure. + */ +type CcsdkMessage = { + role: "user" | "assistant"; + content: string | CcsdkContent[]; + usage?: { + input_tokens?: number; + output_tokens?: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + }; +}; + +/** + * CCSDK session entry envelope. + */ +type CcsdkEntry = { + type: "user" | "assistant"; + uuid: string; + parentUuid?: string; + sessionId: string; + cwd?: string; + version?: string; + gitBranch?: string; + slug?: string; + timestamp?: string; + message: CcsdkMessage; +}; + +/** + * Options for creating the CCSDK session adapter. + */ +export type CcsdkSessionAdapterOptions = { + sessionId: string; + cwd?: string; + version?: string; + gitBranch?: string; + slug?: string; +}; + +/** + * Generate a UUID for CCSDK messages. + */ +function generateUuid(): string { + // Simple UUID v4-like generation + return "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx".replace(/[xy]/g, (c) => { + const r = (Math.random() * 16) | 0; + const v = c === "x" ? r : (r & 0x3) | 0x8; + return v.toString(16); + }); +} + +/** + * Convert CCSDK content to normalized content. + */ +function normalizeCcsdkContent(content: string | CcsdkContent[]): NormalizedContent[] { + if (typeof content === "string") { + return [{ type: "text", text: content }]; + } + + return content + .map((block): NormalizedContent | null => { + switch (block.type) { + case "text": + return { type: "text", text: block.text }; + case "image": + return { + type: "image", + data: block.source.data, + mimeType: block.source.media_type, + }; + case "tool_use": + return { + type: "tool_call", + id: block.id, + name: block.name, + arguments: block.input, + }; + case "thinking": + return { type: "thinking", text: block.thinking }; + case "tool_result": + // Tool results are handled separately + return null; + default: + return null; + } + }) + .filter((b): b is NormalizedContent => b !== null); +} + +/** + * Convert normalized assistant content to CCSDK format. + */ +function denormalizeAssistantContent(content: AssistantContent[]): CcsdkContent[] { + return content.map((block): CcsdkContent => { + switch (block.type) { + case "text": + return { type: "text", text: block.text }; + case "tool_call": + return { + type: "tool_use", + id: block.id, + name: block.name, + input: block.arguments, + }; + case "thinking": + return { type: "thinking", thinking: block.text }; + default: + return { type: "text", text: "" }; + } + }); +} + +/** + * CCSDK session adapter implementation. + */ +export class CcsdkSessionAdapter implements SessionAdapter { + readonly format = "ccsdk" as const; + readonly sessionFile: string; + + private metadata: SessionMetadata; + private currentParentUuid: string | undefined; + private entries: CcsdkEntry[] = []; + private entriesLoaded = false; + private pendingWrites: CcsdkEntry[] = []; + + constructor(sessionFile: string, options: CcsdkSessionAdapterOptions) { + this.sessionFile = sessionFile; + this.metadata = { + sessionId: options.sessionId, + cwd: options.cwd, + version: options.version, + gitBranch: options.gitBranch, + slug: options.slug, + runtime: "ccsdk", + }; + } + + getMetadata(): SessionMetadata { + return this.metadata; + } + + /** + * Set the parent UUID for tree-structured messages. + */ + setParentId(parentId: string): void { + this.currentParentUuid = parentId; + } + + /** + * Load entries from the session file. + */ + private async loadEntries(): Promise { + if (this.entriesLoaded) return; + + try { + await fs.access(this.sessionFile); + } catch { + // File doesn't exist yet + this.entriesLoaded = true; + return; + } + + const stream = createReadStream(this.sessionFile, { encoding: "utf8" }); + const rl = readline.createInterface({ input: stream, crlfDelay: Infinity }); + + for await (const line of rl) { + const trimmed = line.trim(); + if (!trimmed) continue; + + try { + const entry = JSON.parse(trimmed) as CcsdkEntry; + if (entry.type === "user" || entry.type === "assistant") { + this.entries.push(entry); + // Track the latest UUID as potential parent + if (entry.uuid) { + this.currentParentUuid = entry.uuid; + } + // Update metadata from entries + if (entry.cwd) this.metadata.cwd = entry.cwd; + if (entry.version) this.metadata.version = entry.version; + if (entry.gitBranch) this.metadata.gitBranch = entry.gitBranch; + if (entry.slug) this.metadata.slug = entry.slug; + } + } catch { + // Skip malformed lines + } + } + + this.entriesLoaded = true; + } + + async loadHistory(): Promise { + await this.loadEntries(); + + const messages: NormalizedMessage[] = []; + + for (const entry of this.entries) { + const content = normalizeCcsdkContent(entry.message.content); + + // Extract tool results from content (they're inline in CCSDK) + const contentArr = Array.isArray(entry.message.content) ? entry.message.content : []; + const toolResults = contentArr.filter((c): c is CcsdkToolResult => c.type === "tool_result"); + + // Add the main message + if (entry.type === "user") { + messages.push({ + id: entry.uuid, + role: "user", + content, + timestamp: entry.timestamp ? new Date(entry.timestamp).getTime() : undefined, + metadata: { + parentUuid: entry.parentUuid, + cwd: entry.cwd, + gitBranch: entry.gitBranch, + }, + }); + } else if (entry.type === "assistant") { + messages.push({ + id: entry.uuid, + role: "assistant", + content, + timestamp: entry.timestamp ? new Date(entry.timestamp).getTime() : undefined, + metadata: { + parentUuid: entry.parentUuid, + cwd: entry.cwd, + gitBranch: entry.gitBranch, + usage: entry.message.usage, + }, + }); + } + + // Add tool results as separate messages for consistency + for (const tr of toolResults) { + const resultContent = + typeof tr.content === "string" + ? [{ type: "text" as const, text: tr.content }] + : tr.content.map((c) => { + if (c.type === "text") { + return { type: "text" as const, text: c.text }; + } + return { + type: "image" as const, + data: c.source.data, + mimeType: c.source.media_type, + }; + }); + + messages.push({ + id: `${entry.uuid}-tr-${tr.tool_use_id}`, + role: "tool_result", + content: { + type: "tool_result", + toolCallId: tr.tool_use_id, + content: resultContent, + isError: tr.is_error, + }, + }); + } + } + + return messages; + } + + async appendUserMessage(content: string, images?: NormalizedImageContent[]): Promise { + const uuid = generateUuid(); + const messageContent: CcsdkContent[] = [{ type: "text", text: content }]; + + if (images && images.length > 0) { + for (const img of images) { + messageContent.push({ + type: "image", + source: { + type: "base64", + media_type: img.mimeType, + data: img.data, + }, + }); + } + } + + const entry: CcsdkEntry = { + type: "user", + uuid, + parentUuid: this.currentParentUuid, + sessionId: this.metadata.sessionId, + cwd: this.metadata.cwd, + version: this.metadata.version, + gitBranch: this.metadata.gitBranch, + slug: this.metadata.slug, + timestamp: new Date().toISOString(), + message: { + role: "user", + content: messageContent, + }, + }; + + this.entries.push(entry); + this.pendingWrites.push(entry); + this.currentParentUuid = uuid; + + return uuid; + } + + async appendAssistantMessage(content: AssistantContent[], usage?: UsageInfo): Promise { + const uuid = generateUuid(); + const ccsdkContent = denormalizeAssistantContent(content); + + const entry: CcsdkEntry = { + type: "assistant", + uuid, + parentUuid: this.currentParentUuid, + sessionId: this.metadata.sessionId, + cwd: this.metadata.cwd, + version: this.metadata.version, + gitBranch: this.metadata.gitBranch, + slug: this.metadata.slug, + timestamp: new Date().toISOString(), + message: { + role: "assistant", + content: ccsdkContent, + usage: usage + ? { + input_tokens: usage.inputTokens, + output_tokens: usage.outputTokens, + cache_read_input_tokens: usage.cacheReadTokens, + cache_creation_input_tokens: usage.cacheWriteTokens, + } + : undefined, + }, + }; + + this.entries.push(entry); + this.pendingWrites.push(entry); + this.currentParentUuid = uuid; + + return uuid; + } + + async appendToolResult( + toolCallId: string, + result: NormalizedToolResultContent, + isError?: boolean, + ): Promise { + // In CCSDK format, tool results are typically part of the next user message + // For standalone storage, we create a synthetic user message with the tool result + const uuid = generateUuid(); + + const toolResultContent: CcsdkToolResult = { + type: "tool_result", + tool_use_id: toolCallId, + content: result.content.map((c) => { + if (c.type === "text") { + return { type: "text" as const, text: c.text }; + } + return { + type: "image" as const, + source: { + type: "base64" as const, + media_type: c.mimeType, + data: c.data, + }, + }; + }), + is_error: isError ?? result.isError, + }; + + const entry: CcsdkEntry = { + type: "user", + uuid, + parentUuid: this.currentParentUuid, + sessionId: this.metadata.sessionId, + cwd: this.metadata.cwd, + version: this.metadata.version, + timestamp: new Date().toISOString(), + message: { + role: "user", + content: [toolResultContent], + }, + }; + + this.entries.push(entry); + this.pendingWrites.push(entry); + this.currentParentUuid = uuid; + + return uuid; + } + + async flush(): Promise { + if (this.pendingWrites.length === 0) return; + + // Ensure directory exists + const dir = this.sessionFile.substring(0, this.sessionFile.lastIndexOf("/")); + if (dir) { + await fs.mkdir(dir, { recursive: true }); + } + + // Append entries as JSONL + const lines = this.pendingWrites.map((entry) => JSON.stringify(entry)).join("\n") + "\n"; + await fs.appendFile(this.sessionFile, lines, "utf8"); + + this.pendingWrites = []; + } + + async close(): Promise { + await this.flush(); + this.entries = []; + this.entriesLoaded = false; + } +} + +/** + * Create a CCSDK session adapter. + */ +export function createCcsdkSessionAdapter( + sessionFile: string, + options: CcsdkSessionAdapterOptions, +): CcsdkSessionAdapter { + return new CcsdkSessionAdapter(sessionFile, options); +} diff --git a/src/agents/sessions/index.ts b/src/agents/sessions/index.ts new file mode 100644 index 000000000..e80625415 --- /dev/null +++ b/src/agents/sessions/index.ts @@ -0,0 +1,56 @@ +/** + * Session adapter module. + * + * Provides a unified interface for reading and writing session history + * across different runtime formats (Pi-Agent, Claude Code SDK). + */ + +export type { SessionAdapter, SessionAdapterFactory } from "./session-adapter.js"; +export type { + AssistantContent, + NormalizedContent, + NormalizedImageContent, + NormalizedMessage, + NormalizedTextContent, + NormalizedThinking, + NormalizedToolCall, + NormalizedToolResultContent, + SessionMetadata, + UsageInfo, +} from "./types.js"; + +export { + createPiSessionAdapter, + PiSessionAdapter, + type PiSessionAdapterOptions, +} from "./pi-session-adapter.js"; + +export { + createCcsdkSessionAdapter, + CcsdkSessionAdapter, + type CcsdkSessionAdapterOptions, +} from "./ccsdk-session-adapter.js"; + +import type { SessionAdapter } from "./session-adapter.js"; +import { createPiSessionAdapter } from "./pi-session-adapter.js"; +import { createCcsdkSessionAdapter } from "./ccsdk-session-adapter.js"; + +/** + * Factory function to create the appropriate session adapter. + */ +export function createSessionAdapter( + runtime: "pi-agent" | "ccsdk", + sessionFile: string, + options: { + sessionId: string; + cwd?: string; + version?: string; + gitBranch?: string; + slug?: string; + }, +): SessionAdapter { + if (runtime === "ccsdk") { + return createCcsdkSessionAdapter(sessionFile, options); + } + return createPiSessionAdapter(sessionFile, options); +} diff --git a/src/agents/sessions/pi-session-adapter.ts b/src/agents/sessions/pi-session-adapter.ts new file mode 100644 index 000000000..f3cc3a378 --- /dev/null +++ b/src/agents/sessions/pi-session-adapter.ts @@ -0,0 +1,377 @@ +/** + * Pi-Agent session adapter. + * + * Wraps the pi-coding-agent SessionManager to provide a unified session interface. + * Pi-Agent uses a flat JSONL format with session header and message entries. + * + * Pi-Agent JSONL Format: + * ```jsonl + * {"type":"session","version":"1","id":"...","cwd":"..."} + * {"type":"message","message":{"role":"user","content":"..."}} + * {"type":"message","message":{"role":"assistant","content":[{"type":"text","text":"..."},{"type":"toolCall","id":"...","name":"...","arguments":{}}]}} + * {"type":"message","message":{"role":"toolResult","toolCallId":"...","content":[{"type":"text","text":"..."}]}} + * ``` + */ + +import type { SessionAdapter } from "./session-adapter.js"; +import type { + AssistantContent, + NormalizedContent, + NormalizedImageContent, + NormalizedMessage, + NormalizedToolResultContent, + SessionMetadata, + UsageInfo, +} from "./types.js"; + +/** + * Pi-Agent message content types. + */ +type PiTextContent = { type: "text"; text: string }; +type PiImageContent = { type: "image"; data: string; mimeType: string }; +type PiToolCall = { + type: "toolCall" | "toolUse" | "functionCall"; + id: string; + name: string; + arguments: Record; +}; +type PiToolResult = { + type: "toolResult"; + toolCallId: string; + content: Array; + isError?: boolean; +}; +type PiContent = PiTextContent | PiImageContent | PiToolCall | PiToolResult; + +/** + * Pi-Agent message structure. + */ +type PiMessage = { + role: "user" | "assistant" | "toolResult"; + content: string | PiContent[]; +}; + +/** + * Pi-Agent session entry types. + */ +type PiSessionHeader = { + type: "session"; + version?: string; + id?: string; + cwd?: string; +}; + +type PiMessageEntry = { + type: "message"; + id?: string; + parentId?: string; + message: PiMessage; +}; + +type PiEntry = PiSessionHeader | PiMessageEntry; + +/** + * SessionManager interface (subset we need from pi-coding-agent). + */ +interface PiSessionManager { + sessionId: string; + fileEntries: PiEntry[]; + appendMessage(message: PiMessage): void; + buildSessionContext(): { messages: PiMessage[] }; + getLeafEntry(): PiEntry | null; + branch(parentId: string): void; + resetLeaf(): void; +} + +/** + * Options for creating the Pi-Agent session adapter. + */ +export type PiSessionAdapterOptions = { + sessionId: string; + cwd?: string; + /** Existing SessionManager instance to wrap. */ + sessionManager?: PiSessionManager; +}; + +/** + * Generate a unique message ID. + */ +function generateMessageId(): string { + return `pi-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; +} + +/** + * Convert Pi-Agent content to normalized content. + */ +function normalizePiContent(content: string | PiContent[]): NormalizedContent[] { + if (typeof content === "string") { + return [{ type: "text", text: content }]; + } + + return content + .map((block): NormalizedContent | null => { + switch (block.type) { + case "text": + return { type: "text", text: block.text }; + case "image": + return { + type: "image", + data: block.data, + mimeType: block.mimeType, + }; + case "toolCall": + case "toolUse": + case "functionCall": + return { + type: "tool_call", + id: block.id, + name: block.name, + arguments: block.arguments, + }; + default: + return null; + } + }) + .filter((b): b is NormalizedContent => b !== null); +} + +/** + * Convert normalized assistant content to Pi-Agent format. + */ +function denormalizeAssistantContent(content: AssistantContent[]): PiContent[] { + return content.map((block): PiContent => { + switch (block.type) { + case "text": + return { type: "text", text: block.text }; + case "tool_call": + return { + type: "toolCall", + id: block.id, + name: block.name, + arguments: block.arguments, + }; + case "thinking": + // Thinking blocks are typically not persisted in Pi-Agent format + // Convert to text with marker + return { type: "text", text: `${block.text}` }; + default: + return { type: "text", text: "" }; + } + }); +} + +/** + * Pi-Agent session adapter implementation. + */ +export class PiSessionAdapter implements SessionAdapter { + readonly format = "pi-agent" as const; + readonly sessionFile: string; + + private sessionManager: PiSessionManager | null; + private metadata: SessionMetadata; + private pendingWrites: PiEntry[] = []; + + constructor(sessionFile: string, options: PiSessionAdapterOptions) { + this.sessionFile = sessionFile; + this.sessionManager = options.sessionManager ?? null; + this.metadata = { + sessionId: options.sessionId, + cwd: options.cwd, + version: "1", + runtime: "pi-agent", + }; + } + + /** + * Set the SessionManager instance (for use with existing pi-coding-agent setup). + */ + setSessionManager(sm: PiSessionManager): void { + this.sessionManager = sm; + // Update metadata from SessionManager + const header = sm.fileEntries.find((e): e is PiSessionHeader => e.type === "session"); + if (header) { + this.metadata.sessionId = header.id ?? this.metadata.sessionId; + this.metadata.cwd = header.cwd ?? this.metadata.cwd; + this.metadata.version = header.version ?? this.metadata.version; + } + } + + getMetadata(): SessionMetadata { + return this.metadata; + } + + async loadHistory(): Promise { + if (!this.sessionManager) { + return []; + } + + const context = this.sessionManager.buildSessionContext(); + const messages: NormalizedMessage[] = []; + + for (let i = 0; i < context.messages.length; i++) { + const msg = context.messages[i]; + const id = generateMessageId(); + + if (msg.role === "user") { + messages.push({ + id, + role: "user", + content: normalizePiContent(msg.content), + }); + } else if (msg.role === "assistant") { + messages.push({ + id, + role: "assistant", + content: normalizePiContent(msg.content), + }); + } else if (msg.role === "toolResult") { + // Tool results in Pi-Agent are separate messages + const content = msg.content as PiContent[]; + const toolResult = content.find((c): c is PiToolResult => c.type === "toolResult"); + if (toolResult) { + messages.push({ + id, + role: "tool_result", + content: { + type: "tool_result", + toolCallId: toolResult.toolCallId, + content: toolResult.content.map((c) => ({ + type: c.type as "text" | "image", + ...(c.type === "text" ? { text: (c as PiTextContent).text } : {}), + ...(c.type === "image" + ? { + data: (c as PiImageContent).data, + mimeType: (c as PiImageContent).mimeType, + } + : {}), + })) as Array< + { type: "text"; text: string } | { type: "image"; data: string; mimeType: string } + >, + isError: toolResult.isError, + }, + }); + } + } + } + + return messages; + } + + async appendUserMessage(content: string, images?: NormalizedImageContent[]): Promise { + const id = generateMessageId(); + const messageContent: PiContent[] = [{ type: "text", text: content }]; + + if (images && images.length > 0) { + for (const img of images) { + messageContent.push({ + type: "image", + data: img.data, + mimeType: img.mimeType, + }); + } + } + + const message: PiMessage = { + role: "user", + content: messageContent, + }; + + if (this.sessionManager) { + this.sessionManager.appendMessage(message); + } else { + this.pendingWrites.push({ + type: "message", + id, + message, + }); + } + + return id; + } + + async appendAssistantMessage(content: AssistantContent[], _usage?: UsageInfo): Promise { + const id = generateMessageId(); + const piContent = denormalizeAssistantContent(content); + + const message: PiMessage = { + role: "assistant", + content: piContent, + }; + + if (this.sessionManager) { + this.sessionManager.appendMessage(message); + } else { + this.pendingWrites.push({ + type: "message", + id, + message, + }); + } + + return id; + } + + async appendToolResult( + toolCallId: string, + result: NormalizedToolResultContent, + isError?: boolean, + ): Promise { + const id = generateMessageId(); + + const toolResultContent: PiContent[] = [ + { + type: "toolResult", + toolCallId, + content: result.content.map((c) => { + if (c.type === "text") { + return { type: "text" as const, text: c.text }; + } + return { + type: "image" as const, + data: c.data, + mimeType: c.mimeType, + }; + }), + isError: isError ?? result.isError, + }, + ]; + + const message: PiMessage = { + role: "toolResult", + content: toolResultContent, + }; + + if (this.sessionManager) { + this.sessionManager.appendMessage(message); + } else { + this.pendingWrites.push({ + type: "message", + id, + message, + }); + } + + return id; + } + + async flush(): Promise { + // SessionManager handles flushing internally + // Nothing to do here for the wrapped case + } + + async close(): Promise { + // SessionManager cleanup is handled externally + this.sessionManager = null; + this.pendingWrites = []; + } +} + +/** + * Create a Pi-Agent session adapter. + */ +export function createPiSessionAdapter( + sessionFile: string, + options: PiSessionAdapterOptions, +): PiSessionAdapter { + return new PiSessionAdapter(sessionFile, options); +} diff --git a/src/agents/sessions/session-adapter.ts b/src/agents/sessions/session-adapter.ts new file mode 100644 index 000000000..a921d4d9b --- /dev/null +++ b/src/agents/sessions/session-adapter.ts @@ -0,0 +1,107 @@ +/** + * Abstract session adapter interface. + * + * Provides a unified interface for reading and writing session history + * across different runtime formats (Pi-Agent JSONL, Claude Code SDK JSONL). + * + * Each runtime reads/writes its own JSONL format. No cross-format conversion. + */ + +import type { + AssistantContent, + NormalizedImageContent, + NormalizedMessage, + NormalizedToolResultContent, + SessionMetadata, + UsageInfo, +} from "./types.js"; + +/** + * Abstract interface for session persistence. + * + * Implementations handle reading/writing session history in their native format. + */ +export interface SessionAdapter { + /** Runtime format discriminant. */ + readonly format: "pi-agent" | "ccsdk"; + + /** Path to the session file. */ + readonly sessionFile: string; + + // ─── Reading ─────────────────────────────────────────────────────────────── + + /** + * Load conversation history from the session file. + * Returns messages in normalized format for internal processing. + */ + loadHistory(): Promise; + + /** + * Get session metadata (ID, cwd, version, etc.). + */ + getMetadata(): SessionMetadata; + + // ─── Writing ─────────────────────────────────────────────────────────────── + + /** + * Append a user message to the session. + * @param content Text content of the message. + * @param images Optional image attachments. + * @returns Message ID. + */ + appendUserMessage(content: string, images?: NormalizedImageContent[]): Promise; + + /** + * Append an assistant message to the session. + * @param content Array of content blocks (text, tool calls, thinking). + * @param usage Optional token usage information. + * @returns Message ID. + */ + appendAssistantMessage(content: AssistantContent[], usage?: UsageInfo): Promise; + + /** + * Append a tool result to the session. + * @param toolCallId ID of the tool call this is a result for. + * @param result Tool result content. + * @param isError Whether this is an error result. + * @returns Message ID. + */ + appendToolResult( + toolCallId: string, + result: NormalizedToolResultContent, + isError?: boolean, + ): Promise; + + // ─── CCSDK-specific (no-op for pi-agent) ─────────────────────────────────── + + /** + * Set the parent message ID for tree-structured sessions. + * CCSDK uses parentUuid/uuid for conversation threading. + * No-op for Pi-Agent adapter. + */ + setParentId?(parentId: string): void; + + // ─── Lifecycle ───────────────────────────────────────────────────────────── + + /** + * Flush any pending writes to disk. + */ + flush(): Promise; + + /** + * Close the session adapter and release resources. + */ + close(): Promise; +} + +/** + * Factory function to create the appropriate session adapter. + */ +export type SessionAdapterFactory = ( + runtime: "pi-agent" | "ccsdk", + sessionFile: string, + options?: { + sessionId?: string; + cwd?: string; + }, +) => SessionAdapter; diff --git a/src/agents/sessions/types.ts b/src/agents/sessions/types.ts new file mode 100644 index 000000000..a08d2a759 --- /dev/null +++ b/src/agents/sessions/types.ts @@ -0,0 +1,109 @@ +/** + * Shared types for session adapters. + * + * These types provide a normalized representation of conversation history + * that can be converted to/from runtime-specific formats. + */ + +/** + * Normalized text content block. + */ +export type NormalizedTextContent = { + type: "text"; + text: string; +}; + +/** + * Normalized image content block. + */ +export type NormalizedImageContent = { + type: "image"; + data: string; + mimeType: string; +}; + +/** + * Normalized tool call content block. + */ +export type NormalizedToolCall = { + type: "tool_call"; + id: string; + name: string; + arguments: Record; +}; + +/** + * Normalized thinking/reasoning content block. + */ +export type NormalizedThinking = { + type: "thinking"; + text: string; +}; + +/** + * Union of all normalized content block types. + */ +export type NormalizedContent = + | NormalizedTextContent + | NormalizedImageContent + | NormalizedToolCall + | NormalizedThinking; + +/** + * Normalized tool result content. + */ +export type NormalizedToolResultContent = { + type: "tool_result"; + toolCallId: string; + content: Array; + isError?: boolean; +}; + +/** + * Normalized message format for internal processing. + */ +export type NormalizedMessage = { + /** Unique identifier for the message. */ + id: string; + /** Message role. */ + role: "user" | "assistant" | "tool_result"; + /** Message content blocks. */ + content: NormalizedContent[] | NormalizedToolResultContent; + /** Timestamp of the message. */ + timestamp?: number; + /** Additional metadata. */ + metadata?: Record; +}; + +/** + * Session metadata. + */ +export type SessionMetadata = { + /** Session ID. */ + sessionId: string; + /** Working directory when the session was created. */ + cwd?: string; + /** Session format version. */ + version?: string; + /** Git branch (CCSDK specific). */ + gitBranch?: string; + /** Session slug (CCSDK specific). */ + slug?: string; + /** Runtime type. */ + runtime: "pi-agent" | "ccsdk"; +}; + +/** + * Usage information for a message. + */ +export type UsageInfo = { + inputTokens?: number; + outputTokens?: number; + cacheReadTokens?: number; + cacheWriteTokens?: number; +}; + +/** + * Assistant content types for appending messages. + */ +export type AssistantContent = NormalizedTextContent | NormalizedToolCall | NormalizedThinking; diff --git a/src/config/zod-schema.agent-defaults.ts b/src/config/zod-schema.agent-defaults.ts index aec9c6193..4222ad5e1 100644 --- a/src/config/zod-schema.agent-defaults.ts +++ b/src/config/zod-schema.agent-defaults.ts @@ -6,6 +6,7 @@ import { SandboxBrowserSchema, SandboxDockerSchema, SandboxPruneSchema, + validateCcsdkModelProvider, } from "./zod-schema.agent-runtime.js"; import { BlockStreamingChunkSchema, @@ -174,4 +175,17 @@ export const AgentDefaultsSchema = z .optional(), }) .strict() + .superRefine((value, ctx) => { + // Validate CCSDK runtime requires compatible model provider + if (value.runtime === "ccsdk") { + const error = validateCcsdkModelProvider(value.model); + if (error) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["model"], + message: error, + }); + } + } + }) .optional(); diff --git a/src/config/zod-schema.agent-runtime.ts b/src/config/zod-schema.agent-runtime.ts index a58bd0c9c..e3afbf80d 100644 --- a/src/config/zod-schema.agent-runtime.ts +++ b/src/config/zod-schema.agent-runtime.ts @@ -416,6 +416,58 @@ export const AgentModelSchema = z.union([ /** Agent runtime backend discriminant. */ export const AgentRuntimeKindSchema = z.union([z.literal("pi"), z.literal("ccsdk")]); +/** + * Providers compatible with the Claude Code SDK (ccsdk) runtime. + * + * These providers expose Anthropic-compatible API endpoints: + * - anthropic: Direct Anthropic API + * - zai: z.AI subscription (Anthropic-compatible, supports non-Claude models like GLM-4.7) + * - openrouter: OpenRouter (Anthropic-compatible proxy, supports many models) + * + * Note: Bedrock and Vertex AI are supported by CCSDK but via environment variables + * (CLAUDE_CODE_USE_BEDROCK=1, CLAUDE_CODE_USE_VERTEX=1), not as provider prefixes. + */ +export const CCSDK_COMPATIBLE_PROVIDERS = new Set(["anthropic", "zai", "openrouter"]); + +/** + * Extract provider from a model string (e.g., "anthropic/claude-sonnet-4" -> "anthropic"). + * Returns null if no provider prefix is found. + */ +function extractProviderFromModelString(model: string): string | null { + const trimmed = model.trim(); + if (!trimmed) return null; + const slash = trimmed.indexOf("/"); + if (slash === -1) return null; + return trimmed.slice(0, slash).trim().toLowerCase() || null; +} + +/** + * Validate that a model is compatible with the CCSDK runtime. + * Returns an error message if invalid, or null if valid. + */ +export function validateCcsdkModelProvider( + model: string | { primary?: string; fallbacks?: string[] } | undefined, +): string | null { + if (!model) return null; // No model specified, will use defaults + + const modelsToCheck: string[] = []; + if (typeof model === "string") { + modelsToCheck.push(model); + } else { + if (model.primary) modelsToCheck.push(model.primary); + if (model.fallbacks) modelsToCheck.push(...model.fallbacks); + } + + for (const m of modelsToCheck) { + const provider = extractProviderFromModelString(m); + if (provider && !CCSDK_COMPATIBLE_PROVIDERS.has(provider)) { + return `runtime "ccsdk" requires an Anthropic-compatible provider (anthropic, zai, or openrouter), but model "${m}" uses provider "${provider}"`; + } + } + + return null; +} + /** Model tier configuration for Claude Code SDK. */ export const CcSdkModelTiersSchema = z .object({ @@ -471,7 +523,20 @@ export const AgentEntrySchema = z sandbox: AgentSandboxSchema, tools: AgentToolsSchema, }) - .strict(); + .strict() + .superRefine((value, ctx) => { + // Validate CCSDK runtime requires compatible model provider + if (value.runtime === "ccsdk") { + const error = validateCcsdkModelProvider(value.model); + if (error) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["model"], + message: error, + }); + } + } + }); export const ToolsSchema = z .object({