/** * ClaudeAdapterLive - Scoped live implementation for the Claude Agent provider adapter. * * Wraps `@anthropic-ai/claude-agent-sdk` query sessions behind the generic * provider adapter contract and emits canonical runtime events. * * @module ClaudeAdapterLive */ import { type CanUseTool, query, type Options as ClaudeQueryOptions, type PermissionMode, type PermissionResult, type PermissionUpdate, type SDKMessage, type SDKResultMessage, type SDKUserMessage, } from "@anthropic-ai/claude-agent-sdk"; import { ApprovalRequestId, type CanonicalItemType, type CanonicalRequestType, EventId, type ProviderApprovalDecision, ProviderItemId, type ProviderRuntimeEvent, type ProviderRuntimeTurnStatus, type ProviderSendTurnInput, type ProviderSession, type ProviderUserInputAnswers, type RuntimeContentStreamKind, RuntimeItemId, RuntimeRequestId, RuntimeTaskId, ThreadId, TurnId, type UserInputQuestion, } from "@t3tools/contracts"; import { applyClaudePromptEffortPrefix, getEffectiveClaudeCodeEffort, getReasoningEffortOptions, resolveReasoningEffortForProvider, supportsClaudeFastMode, supportsClaudeThinkingToggle, supportsClaudeUltrathinkKeyword, } from "@t3tools/shared/model"; import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Ref, Stream } from "effect"; import { ProviderAdapterProcessError, ProviderAdapterRequestError, ProviderAdapterSessionClosedError, ProviderAdapterSessionNotFoundError, ProviderAdapterValidationError, type ProviderAdapterError, } from "../Errors.ts"; import { ClaudeAdapter, type ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "claudeAgent" as const; type ClaudeTextStreamKind = Extract; type ClaudeToolResultStreamKind = Extract< RuntimeContentStreamKind, "command_output" | "file_change_output" >; type PromptQueueItem = | { readonly type: "message"; readonly message: SDKUserMessage; } | { readonly type: "terminate"; }; interface ClaudeResumeState { readonly threadId?: ThreadId; readonly resume?: string; readonly resumeSessionAt?: string; readonly turnCount?: number; } interface ClaudeTurnState { readonly turnId: TurnId; readonly startedAt: string; readonly items: Array; readonly assistantTextBlocks: Map; readonly assistantTextBlockOrder: Array; nextSyntheticAssistantBlockIndex: number; } interface AssistantTextBlockState { readonly itemId: string; emittedTextDelta: boolean; fallbackText: string; streamClosed: boolean; completionEmitted: boolean; } interface PendingApproval { readonly requestType: CanonicalRequestType; readonly detail?: string; readonly suggestions?: ReadonlyArray; readonly decision: Deferred.Deferred; } interface PendingUserInput { readonly questions: ReadonlyArray; readonly answers: Deferred.Deferred; } interface ToolInFlight { readonly itemId: string; readonly itemType: CanonicalItemType; readonly toolName: string; readonly title: string; readonly detail?: string; readonly input: Record; readonly partialInputJson: string; readonly lastEmittedInputFingerprint?: string; } interface ClaudeSessionContext { session: ProviderSession; readonly promptQueue: Queue.Queue; readonly query: ClaudeQueryRuntime; readonly startedAt: string; readonly basePermissionMode: PermissionMode | undefined; resumeSessionId: string | undefined; readonly pendingApprovals: Map; readonly pendingUserInputs: Map; readonly turns: Array<{ id: TurnId; items: Array; }>; readonly inFlightTools: Map; turnState: ClaudeTurnState | undefined; lastAssistantUuid: string | undefined; lastThreadStartedId: string | undefined; stopped: boolean; } interface ClaudeQueryRuntime extends AsyncIterable { readonly interrupt: () => Promise; readonly setModel: (model?: string) => Promise; readonly setPermissionMode: (mode: PermissionMode) => Promise; readonly setMaxThinkingTokens: (maxThinkingTokens: number | null) => Promise; readonly close: () => void; } export interface ClaudeAdapterLiveOptions { readonly createQuery?: (input: { readonly prompt: AsyncIterable; readonly options: ClaudeQueryOptions; }) => ClaudeQueryRuntime; readonly nativeEventLogPath?: string; readonly nativeEventLogger?: EventNdjsonLogger; } function isUuid(value: string): boolean { return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value); } function isSyntheticClaudeThreadId(value: string): boolean { return value.startsWith("claude-thread-"); } function toMessage(cause: unknown, fallback: string): string { if (cause instanceof Error && cause.message.length > 0) { return cause.message; } return fallback; } function asRuntimeItemId(value: string): RuntimeItemId { return RuntimeItemId.makeUnsafe(value); } function asCanonicalTurnId(value: TurnId): TurnId { return value; } function asRuntimeRequestId(value: ApprovalRequestId): RuntimeRequestId { return RuntimeRequestId.makeUnsafe(value); } function toPermissionMode(value: unknown): PermissionMode | undefined { switch (value) { case "default": case "acceptEdits": case "bypassPermissions": case "plan": case "dontAsk": return value; default: return undefined; } } function readClaudeResumeState(resumeCursor: unknown): ClaudeResumeState | undefined { if (!resumeCursor || typeof resumeCursor !== "object") { return undefined; } const cursor = resumeCursor as { threadId?: unknown; resume?: unknown; sessionId?: unknown; resumeSessionAt?: unknown; turnCount?: unknown; }; const threadIdCandidate = typeof cursor.threadId === "string" ? cursor.threadId : undefined; const threadId = threadIdCandidate && !isSyntheticClaudeThreadId(threadIdCandidate) ? ThreadId.makeUnsafe(threadIdCandidate) : undefined; const resumeCandidate = typeof cursor.resume === "string" ? cursor.resume : typeof cursor.sessionId === "string" ? cursor.sessionId : undefined; const resume = resumeCandidate && isUuid(resumeCandidate) ? resumeCandidate : undefined; const resumeSessionAt = typeof cursor.resumeSessionAt === "string" ? cursor.resumeSessionAt : undefined; const turnCountValue = typeof cursor.turnCount === "number" ? cursor.turnCount : undefined; return { ...(threadId ? { threadId } : {}), ...(resume ? { resume } : {}), ...(resumeSessionAt ? { resumeSessionAt } : {}), ...(turnCountValue !== undefined && Number.isInteger(turnCountValue) && turnCountValue >= 0 ? { turnCount: turnCountValue } : {}), }; } function classifyToolItemType(toolName: string): CanonicalItemType { const normalized = toolName.toLowerCase(); if (normalized.includes("agent")) { return "collab_agent_tool_call"; } if ( normalized === "task" || normalized === "agent" || normalized.includes("subagent") || normalized.includes("sub-agent") ) { return "collab_agent_tool_call"; } if ( normalized.includes("bash") || normalized.includes("command") || normalized.includes("shell") || normalized.includes("terminal") ) { return "command_execution"; } if ( normalized.includes("edit") || normalized.includes("write") || normalized.includes("file") || normalized.includes("patch") || normalized.includes("replace") || normalized.includes("create") || normalized.includes("delete") ) { return "file_change"; } if (normalized.includes("mcp")) { return "mcp_tool_call"; } if (normalized.includes("websearch") || normalized.includes("web search")) { return "web_search"; } if (normalized.includes("image")) { return "image_view"; } return "dynamic_tool_call"; } function isReadOnlyToolName(toolName: string): boolean { const normalized = toolName.toLowerCase(); return ( normalized === "read" || normalized.includes("read file") || normalized.includes("view") || normalized.includes("grep") || normalized.includes("glob") || normalized.includes("search") ); } function classifyRequestType(toolName: string): CanonicalRequestType { if (isReadOnlyToolName(toolName)) { return "file_read_approval"; } const itemType = classifyToolItemType(toolName); return itemType === "command_execution" ? "command_execution_approval" : itemType === "file_change" ? "file_change_approval" : "dynamic_tool_call"; } function summarizeToolRequest(toolName: string, input: Record): string { const commandValue = input.command ?? input.cmd; const command = typeof commandValue === "string" ? commandValue : undefined; if (command && command.trim().length > 0) { return `${toolName}: ${command.trim().slice(0, 400)}`; } const serialized = JSON.stringify(input); if (serialized.length <= 400) { return `${toolName}: ${serialized}`; } return `${toolName}: ${serialized.slice(0, 397)}...`; } function titleForTool(itemType: CanonicalItemType): string { switch (itemType) { case "command_execution": return "Command run"; case "file_change": return "File change"; case "mcp_tool_call": return "MCP tool call"; case "collab_agent_tool_call": return "Subagent task"; case "web_search": return "Web search"; case "image_view": return "Image view"; case "dynamic_tool_call": return "Tool call"; default: return "Item"; } } function buildUserMessage(input: ProviderSendTurnInput): SDKUserMessage { const fragments: string[] = []; if (input.input && input.input.trim().length > 0) { fragments.push(input.input.trim()); } for (const attachment of input.attachments ?? []) { if (attachment.type === "image") { fragments.push( `Attached image: ${attachment.name} (${attachment.mimeType}, ${attachment.sizeBytes} bytes).`, ); } } const requestedEffort = resolveReasoningEffortForProvider( "claudeAgent", input.modelOptions?.claudeAgent?.effort ?? null, ); const supportedEffortOptions = getReasoningEffortOptions("claudeAgent", input.model); const promptEffort = requestedEffort === "ultrathink" && supportsClaudeUltrathinkKeyword(input.model) ? "ultrathink" : requestedEffort && supportedEffortOptions.includes(requestedEffort) ? requestedEffort : null; const text = applyClaudePromptEffortPrefix(fragments.join("\n\n"), promptEffort); return { type: "user", session_id: "", parent_tool_use_id: null, message: { role: "user", content: [{ type: "text", text }], }, } as SDKUserMessage; } function turnStatusFromResult(result: SDKResultMessage): ProviderRuntimeTurnStatus { if (result.subtype === "success") { return "completed"; } const errors = result.errors.join(" ").toLowerCase(); if (errors.includes("interrupt")) { return "interrupted"; } if (errors.includes("cancel")) { return "cancelled"; } return "failed"; } function streamKindFromDeltaType(deltaType: string): ClaudeTextStreamKind { return deltaType.includes("thinking") ? "reasoning_text" : "assistant_text"; } function nativeProviderRefs( _context: ClaudeSessionContext, options?: { readonly providerItemId?: string | undefined; }, ): NonNullable { if (options?.providerItemId) { return { providerItemId: ProviderItemId.makeUnsafe(options.providerItemId), }; } return {}; } function extractAssistantTextBlocks(message: SDKMessage): Array { if (message.type !== "assistant") { return []; } const content = (message.message as { content?: unknown } | undefined)?.content; if (!Array.isArray(content)) { return []; } const fragments: string[] = []; for (const block of content) { if (!block || typeof block !== "object") { continue; } const candidate = block as { type?: unknown; text?: unknown }; if ( candidate.type === "text" && typeof candidate.text === "string" && candidate.text.length > 0 ) { fragments.push(candidate.text); } } return fragments; } function extractContentBlockText(block: unknown): string { if (!block || typeof block !== "object") { return ""; } const candidate = block as { type?: unknown; text?: unknown }; return candidate.type === "text" && typeof candidate.text === "string" ? candidate.text : ""; } function extractTextContent(value: unknown): string { if (typeof value === "string") { return value; } if (Array.isArray(value)) { return value.map((entry) => extractTextContent(entry)).join(""); } if (!value || typeof value !== "object") { return ""; } const record = value as { text?: unknown; content?: unknown; }; if (typeof record.text === "string") { return record.text; } return extractTextContent(record.content); } function tryParseJsonRecord(value: string): Record | undefined { try { const parsed = JSON.parse(value); return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? (parsed as Record) : undefined; } catch { return undefined; } } function toolInputFingerprint(input: Record): string | undefined { try { return JSON.stringify(input); } catch { return undefined; } } function toolResultStreamKind(itemType: CanonicalItemType): ClaudeToolResultStreamKind | undefined { switch (itemType) { case "command_execution": return "command_output"; case "file_change": return "file_change_output"; default: return undefined; } } function toolResultBlocksFromUserMessage(message: SDKMessage): Array<{ readonly toolUseId: string; readonly block: Record; readonly text: string; readonly isError: boolean; }> { if (message.type !== "user") { return []; } const content = (message.message as { content?: unknown } | undefined)?.content; if (!Array.isArray(content)) { return []; } const blocks: Array<{ readonly toolUseId: string; readonly block: Record; readonly text: string; readonly isError: boolean; }> = []; for (const entry of content) { if (!entry || typeof entry !== "object") { continue; } const block = entry as Record; if (block.type !== "tool_result") { continue; } const toolUseId = typeof block.tool_use_id === "string" ? block.tool_use_id : undefined; if (!toolUseId) { continue; } blocks.push({ toolUseId, block, text: extractTextContent(block.content), isError: block.is_error === true, }); } return blocks; } function toSessionError( threadId: ThreadId, cause: unknown, ): ProviderAdapterSessionNotFoundError | ProviderAdapterSessionClosedError | undefined { const normalized = toMessage(cause, "").toLowerCase(); if (normalized.includes("unknown session") || normalized.includes("not found")) { return new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId, cause, }); } if (normalized.includes("closed")) { return new ProviderAdapterSessionClosedError({ provider: PROVIDER, threadId, cause, }); } return undefined; } function toRequestError(threadId: ThreadId, method: string, cause: unknown): ProviderAdapterError { const sessionError = toSessionError(threadId, cause); if (sessionError) { return sessionError; } return new ProviderAdapterRequestError({ provider: PROVIDER, method, detail: toMessage(cause, `${method} failed`), cause, }); } function sdkMessageType(value: unknown): string | undefined { if (!value || typeof value !== "object") { return undefined; } const record = value as { type?: unknown }; return typeof record.type === "string" ? record.type : undefined; } function sdkMessageSubtype(value: unknown): string | undefined { if (!value || typeof value !== "object") { return undefined; } const record = value as { subtype?: unknown }; return typeof record.subtype === "string" ? record.subtype : undefined; } function sdkNativeMethod(message: SDKMessage): string { const subtype = sdkMessageSubtype(message); if (subtype) { return `claude/${message.type}/${subtype}`; } if (message.type === "stream_event") { const streamType = sdkMessageType(message.event); if (streamType) { const deltaType = streamType === "content_block_delta" ? sdkMessageType((message.event as { delta?: unknown }).delta) : undefined; if (deltaType) { return `claude/${message.type}/${streamType}/${deltaType}`; } return `claude/${message.type}/${streamType}`; } } return `claude/${message.type}`; } function sdkNativeItemId(message: SDKMessage): string | undefined { if (message.type === "assistant") { const maybeId = (message.message as { id?: unknown }).id; if (typeof maybeId === "string") { return maybeId; } return undefined; } if (message.type === "user") { return toolResultBlocksFromUserMessage(message)[0]?.toolUseId; } if (message.type === "stream_event") { const event = message.event as { type?: unknown; content_block?: { id?: unknown }; }; if (event.type === "content_block_start" && typeof event.content_block?.id === "string") { return event.content_block.id; } } return undefined; } function makeClaudeAdapter(options?: ClaudeAdapterLiveOptions) { return Effect.gen(function* () { const nativeEventLogger = options?.nativeEventLogger ?? (options?.nativeEventLogPath !== undefined ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { stream: "native", }) : undefined); const createQuery = options?.createQuery ?? ((input: { readonly prompt: AsyncIterable; readonly options: ClaudeQueryOptions; }) => query({ prompt: input.prompt, options: input.options }) as ClaudeQueryRuntime); const sessions = new Map(); const runtimeEventQueue = yield* Queue.unbounded(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const nextEventId = Effect.map(Random.nextUUIDv4, (id) => EventId.makeUnsafe(id)); const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso }); const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid); const logNativeSdkMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (!nativeEventLogger) { return; } const observedAt = new Date().toISOString(); const itemId = sdkNativeItemId(message); yield* nativeEventLogger.write( { observedAt, event: { id: "uuid" in message && typeof message.uuid === "string" ? message.uuid : crypto.randomUUID(), kind: "notification", provider: PROVIDER, createdAt: observedAt, method: sdkNativeMethod(message), ...(typeof message.session_id === "string" ? { providerThreadId: message.session_id } : {}), ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}), payload: message, }, }, context.session.threadId, ); }); const snapshotThread = ( context: ClaudeSessionContext, ): Effect.Effect< { threadId: ThreadId; turns: ReadonlyArray<{ id: TurnId; items: ReadonlyArray; }>; }, ProviderAdapterValidationError > => Effect.gen(function* () { const threadId = context.session.threadId; if (!threadId) { return yield* new ProviderAdapterValidationError({ provider: PROVIDER, operation: "readThread", issue: "Session thread id is not initialized yet.", }); } return { threadId, turns: context.turns.map((turn) => ({ id: turn.id, items: [...turn.items], })), }; }); const updateResumeCursor = (context: ClaudeSessionContext): Effect.Effect => Effect.gen(function* () { const threadId = context.session.threadId; if (!threadId) return; const resumeCursor = { threadId, ...(context.resumeSessionId ? { resume: context.resumeSessionId } : {}), ...(context.lastAssistantUuid ? { resumeSessionAt: context.lastAssistantUuid } : {}), turnCount: context.turns.length, }; context.session = { ...context.session, resumeCursor, updatedAt: yield* nowIso, }; }); const ensureAssistantTextBlock = ( context: ClaudeSessionContext, blockIndex: number, options?: { readonly fallbackText?: string; readonly streamClosed?: boolean; }, ): Effect.Effect< | { readonly blockIndex: number; readonly block: AssistantTextBlockState; } | undefined > => Effect.gen(function* () { const turnState = context.turnState; if (!turnState) { return undefined; } const existing = turnState.assistantTextBlocks.get(blockIndex); if (existing) { if (existing.fallbackText.length === 0 && options?.fallbackText) { existing.fallbackText = options.fallbackText; } if (options?.streamClosed) { existing.streamClosed = true; } return { blockIndex, block: existing }; } const block: AssistantTextBlockState = { itemId: yield* Random.nextUUIDv4, emittedTextDelta: false, fallbackText: options?.fallbackText ?? "", streamClosed: options?.streamClosed ?? false, completionEmitted: false, }; turnState.assistantTextBlocks.set(blockIndex, block); turnState.assistantTextBlockOrder.push(blockIndex); return { blockIndex, block }; }); const createSyntheticAssistantTextBlock = ( context: ClaudeSessionContext, fallbackText: string, ): Effect.Effect< | { readonly blockIndex: number; readonly block: AssistantTextBlockState; } | undefined > => Effect.gen(function* () { const turnState = context.turnState; if (!turnState) { return undefined; } const blockIndex = turnState.nextSyntheticAssistantBlockIndex; turnState.nextSyntheticAssistantBlockIndex -= 1; return yield* ensureAssistantTextBlock(context, blockIndex, { fallbackText, streamClosed: true, }); }); const completeAssistantTextBlock = ( context: ClaudeSessionContext, blockIndex: number, options?: { readonly force?: boolean; readonly rawMethod?: string; readonly rawPayload?: unknown; }, ): Effect.Effect => Effect.gen(function* () { const turnState = context.turnState; const block = turnState?.assistantTextBlocks.get(blockIndex); if (!turnState || !block || block.completionEmitted) { return; } if (!options?.force && !block.streamClosed) { return; } if (!block.emittedTextDelta && block.fallbackText.length > 0) { const deltaStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "content.delta", eventId: deltaStamp.eventId, provider: PROVIDER, createdAt: deltaStamp.createdAt, threadId: context.session.threadId, turnId: turnState.turnId, itemId: asRuntimeItemId(block.itemId), payload: { streamKind: "assistant_text", delta: block.fallbackText, }, providerRefs: nativeProviderRefs(context), ...(options?.rawMethod || options?.rawPayload ? { raw: { source: "claude.sdk.message" as const, ...(options.rawMethod ? { method: options.rawMethod } : {}), payload: options?.rawPayload, }, } : {}), }); } block.completionEmitted = true; const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.completed", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, itemId: asRuntimeItemId(block.itemId), threadId: context.session.threadId, turnId: turnState.turnId, payload: { itemType: "assistant_message", status: "completed", title: "Assistant message", ...(block.fallbackText.length > 0 ? { detail: block.fallbackText } : {}), }, providerRefs: nativeProviderRefs(context), ...(options?.rawMethod || options?.rawPayload ? { raw: { source: "claude.sdk.message" as const, ...(options.rawMethod ? { method: options.rawMethod } : {}), payload: options?.rawPayload, }, } : {}), }); }); const backfillAssistantTextBlocksFromSnapshot = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { const turnState = context.turnState; if (!turnState) { return; } const snapshotTextBlocks = extractAssistantTextBlocks(message); if (snapshotTextBlocks.length === 0) { return; } const orderedBlocks = turnState.assistantTextBlockOrder .map((blockIndex) => ({ blockIndex, block: turnState.assistantTextBlocks.get(blockIndex), })) .filter( ( entry, ): entry is { readonly blockIndex: number; readonly block: AssistantTextBlockState; } => entry.block !== undefined, ); for (const [position, text] of snapshotTextBlocks.entries()) { const existingEntry = orderedBlocks[position]; const entry = existingEntry ?? (yield* createSyntheticAssistantTextBlock(context, text).pipe( Effect.map((created) => { if (!created) { return undefined; } orderedBlocks.push(created); return created; }), )); if (!entry) { continue; } if (entry.block.fallbackText.length === 0) { entry.block.fallbackText = text; } if (entry.block.streamClosed && !entry.block.completionEmitted) { yield* completeAssistantTextBlock(context, entry.blockIndex, { rawMethod: "claude/assistant", rawPayload: message, }); } } }); const ensureThreadId = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (typeof message.session_id !== "string" || message.session_id.length === 0) { return; } const nextThreadId = message.session_id; context.resumeSessionId = message.session_id; yield* updateResumeCursor(context); if (context.lastThreadStartedId !== nextThreadId) { context.lastThreadStartedId = nextThreadId; const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "thread.started", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, payload: { providerThreadId: nextThreadId, }, providerRefs: {}, raw: { source: "claude.sdk.message", method: "claude/thread/started", payload: { session_id: message.session_id, }, }, }); } }); const emitRuntimeError = ( context: ClaudeSessionContext, message: string, cause?: unknown, ): Effect.Effect => Effect.gen(function* () { if (cause !== undefined) { void cause; } const turnState = context.turnState; const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "runtime.error", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(turnState ? { turnId: asCanonicalTurnId(turnState.turnId) } : {}), payload: { message, class: "provider_error", ...(cause !== undefined ? { detail: cause } : {}), }, providerRefs: nativeProviderRefs(context), }); }); const emitRuntimeWarning = ( context: ClaudeSessionContext, message: string, detail?: unknown, ): Effect.Effect => Effect.gen(function* () { const turnState = context.turnState; const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "runtime.warning", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(turnState ? { turnId: asCanonicalTurnId(turnState.turnId) } : {}), payload: { message, ...(detail !== undefined ? { detail } : {}), }, providerRefs: nativeProviderRefs(context), }); }); const completeTurn = ( context: ClaudeSessionContext, status: ProviderRuntimeTurnStatus, errorMessage?: string, result?: SDKResultMessage, ): Effect.Effect => Effect.gen(function* () { const turnState = context.turnState; if (!turnState) { const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "turn.completed", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, payload: { state: status, ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}), ...(result?.usage ? { usage: result.usage } : {}), ...(result?.modelUsage ? { modelUsage: result.modelUsage } : {}), ...(typeof result?.total_cost_usd === "number" ? { totalCostUsd: result.total_cost_usd } : {}), ...(errorMessage ? { errorMessage } : {}), }, providerRefs: {}, }); return; } for (const [index, tool] of context.inFlightTools.entries()) { const toolStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.completed", eventId: toolStamp.eventId, provider: PROVIDER, createdAt: toolStamp.createdAt, threadId: context.session.threadId, turnId: turnState.turnId, itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: status === "completed" ? "completed" : "failed", title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), data: { toolName: tool.toolName, input: tool.input, }, }, providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), raw: { source: "claude.sdk.message", method: "claude/result", payload: result ?? { status }, }, }); context.inFlightTools.delete(index); } // Clear any remaining stale entries (e.g. from interrupted content blocks) context.inFlightTools.clear(); for (const blockIndex of turnState.assistantTextBlockOrder) { yield* completeAssistantTextBlock(context, blockIndex, { force: true, rawMethod: "claude/result", rawPayload: result ?? { status }, }); } context.turns.push({ id: turnState.turnId, items: [...turnState.items], }); const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "turn.completed", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, turnId: turnState.turnId, payload: { state: status, ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}), ...(result?.usage ? { usage: result.usage } : {}), ...(result?.modelUsage ? { modelUsage: result.modelUsage } : {}), ...(typeof result?.total_cost_usd === "number" ? { totalCostUsd: result.total_cost_usd } : {}), ...(errorMessage ? { errorMessage } : {}), }, providerRefs: nativeProviderRefs(context), }); const updatedAt = yield* nowIso; context.turnState = undefined; context.session = { ...context.session, status: "ready", activeTurnId: undefined, updatedAt, ...(status === "failed" && errorMessage ? { lastError: errorMessage } : {}), }; yield* updateResumeCursor(context); }); const handleStreamEvent = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (message.type !== "stream_event") { return; } const { event } = message; if (event.type === "content_block_delta") { if ( (event.delta.type === "text_delta" || event.delta.type === "thinking_delta") && context.turnState ) { const deltaText = event.delta.type === "text_delta" ? event.delta.text : typeof event.delta.thinking === "string" ? event.delta.thinking : ""; if (deltaText.length === 0) { return; } const streamKind = streamKindFromDeltaType(event.delta.type); const assistantBlockEntry = event.delta.type === "text_delta" ? yield* ensureAssistantTextBlock(context, event.index) : context.turnState.assistantTextBlocks.get(event.index) ? { blockIndex: event.index, block: context.turnState.assistantTextBlocks.get( event.index, ) as AssistantTextBlockState, } : undefined; if (assistantBlockEntry?.block && event.delta.type === "text_delta") { assistantBlockEntry.block.emittedTextDelta = true; } const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "content.delta", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, turnId: context.turnState.turnId, ...(assistantBlockEntry?.block ? { itemId: asRuntimeItemId(assistantBlockEntry.block.itemId) } : {}), payload: { streamKind, delta: deltaText, }, providerRefs: nativeProviderRefs(context), raw: { source: "claude.sdk.message", method: "claude/stream_event/content_block_delta", payload: message, }, }); return; } if (event.delta.type === "input_json_delta") { const tool = context.inFlightTools.get(event.index); if (!tool || typeof event.delta.partial_json !== "string") { return; } const partialInputJson = tool.partialInputJson + event.delta.partial_json; const parsedInput = tryParseJsonRecord(partialInputJson); const detail = parsedInput ? summarizeToolRequest(tool.toolName, parsedInput) : tool.detail; let nextTool: ToolInFlight = { ...tool, partialInputJson, ...(parsedInput ? { input: parsedInput } : {}), ...(detail ? { detail } : {}), }; const nextFingerprint = parsedInput && Object.keys(parsedInput).length > 0 ? toolInputFingerprint(parsedInput) : undefined; context.inFlightTools.set(event.index, nextTool); if ( !parsedInput || !nextFingerprint || tool.lastEmittedInputFingerprint === nextFingerprint ) { return; } nextTool = { ...nextTool, lastEmittedInputFingerprint: nextFingerprint, }; context.inFlightTools.set(event.index, nextTool); const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.updated", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), itemId: asRuntimeItemId(nextTool.itemId), payload: { itemType: nextTool.itemType, status: "inProgress", title: nextTool.title, ...(nextTool.detail ? { detail: nextTool.detail } : {}), data: { toolName: nextTool.toolName, input: nextTool.input, }, }, providerRefs: nativeProviderRefs(context, { providerItemId: nextTool.itemId }), raw: { source: "claude.sdk.message", method: "claude/stream_event/content_block_delta/input_json_delta", payload: message, }, }); } return; } if (event.type === "content_block_start") { const { index, content_block: block } = event; if (block.type === "text") { yield* ensureAssistantTextBlock(context, index, { fallbackText: extractContentBlockText(block), }); return; } if ( block.type !== "tool_use" && block.type !== "server_tool_use" && block.type !== "mcp_tool_use" ) { return; } const toolName = block.name; const itemType = classifyToolItemType(toolName); const toolInput = typeof block.input === "object" && block.input !== null ? (block.input as Record) : {}; const itemId = block.id; const detail = summarizeToolRequest(toolName, toolInput); const inputFingerprint = Object.keys(toolInput).length > 0 ? toolInputFingerprint(toolInput) : undefined; const tool: ToolInFlight = { itemId, itemType, toolName, title: titleForTool(itemType), detail, input: toolInput, partialInputJson: "", ...(inputFingerprint ? { lastEmittedInputFingerprint: inputFingerprint } : {}), }; context.inFlightTools.set(index, tool); const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.started", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: "inProgress", title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), data: { toolName: tool.toolName, input: toolInput, }, }, providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), raw: { source: "claude.sdk.message", method: "claude/stream_event/content_block_start", payload: message, }, }); return; } if (event.type === "content_block_stop") { const { index } = event; const assistantBlock = context.turnState?.assistantTextBlocks.get(index); if (assistantBlock) { assistantBlock.streamClosed = true; yield* completeAssistantTextBlock(context, index, { rawMethod: "claude/stream_event/content_block_stop", rawPayload: message, }); return; } const tool = context.inFlightTools.get(index); if (!tool) { return; } } }); const handleUserMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (message.type !== "user") { return; } if (context.turnState) { context.turnState.items.push(message.message); } for (const toolResult of toolResultBlocksFromUserMessage(message)) { const toolEntry = Array.from(context.inFlightTools.entries()).find( ([, tool]) => tool.itemId === toolResult.toolUseId, ); if (!toolEntry) { continue; } const [index, tool] = toolEntry; const itemStatus = toolResult.isError ? "failed" : "completed"; const toolData = { toolName: tool.toolName, input: tool.input, result: toolResult.block, }; const updatedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.updated", eventId: updatedStamp.eventId, provider: PROVIDER, createdAt: updatedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: toolResult.isError ? "failed" : "inProgress", title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), data: toolData, }, providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), raw: { source: "claude.sdk.message", method: "claude/user", payload: message, }, }); const streamKind = toolResultStreamKind(tool.itemType); if (streamKind && toolResult.text.length > 0 && context.turnState) { const deltaStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "content.delta", eventId: deltaStamp.eventId, provider: PROVIDER, createdAt: deltaStamp.createdAt, threadId: context.session.threadId, turnId: context.turnState.turnId, itemId: asRuntimeItemId(tool.itemId), payload: { streamKind, delta: toolResult.text, }, providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), raw: { source: "claude.sdk.message", method: "claude/user", payload: message, }, }); } const completedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "item.completed", eventId: completedStamp.eventId, provider: PROVIDER, createdAt: completedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: itemStatus, title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), data: toolData, }, providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), raw: { source: "claude.sdk.message", method: "claude/user", payload: message, }, }); context.inFlightTools.delete(index); } }); const handleAssistantMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (message.type !== "assistant") { return; } // Auto-start a synthetic turn for assistant messages that arrive without // an active turn (e.g., background agent/subagent responses between user prompts). if (!context.turnState) { const turnId = TurnId.makeUnsafe(yield* Random.nextUUIDv4); const startedAt = yield* nowIso; context.turnState = { turnId, startedAt, items: [], assistantTextBlocks: new Map(), assistantTextBlockOrder: [], nextSyntheticAssistantBlockIndex: -1, }; context.session = { ...context.session, status: "running", activeTurnId: turnId, updatedAt: startedAt, }; const turnStartedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "turn.started", eventId: turnStartedStamp.eventId, provider: PROVIDER, createdAt: turnStartedStamp.createdAt, threadId: context.session.threadId, turnId, payload: {}, providerRefs: { ...nativeProviderRefs(context), providerTurnId: turnId, }, raw: { source: "claude.sdk.message", method: "claude/synthetic-turn-start", payload: {}, }, }); } if (context.turnState) { context.turnState.items.push(message.message); yield* backfillAssistantTextBlocksFromSnapshot(context, message); } context.lastAssistantUuid = message.uuid; yield* updateResumeCursor(context); }); const handleResultMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (message.type !== "result") { return; } const status = turnStatusFromResult(message); const errorMessage = message.subtype === "success" ? undefined : message.errors[0]; if (status === "failed") { yield* emitRuntimeError(context, errorMessage ?? "Claude turn failed."); } yield* completeTurn(context, status, errorMessage, message); }); const handleSystemMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { if (message.type !== "system") { return; } const stamp = yield* makeEventStamp(); const base = { eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), providerRefs: nativeProviderRefs(context), raw: { source: "claude.sdk.message" as const, method: sdkNativeMethod(message), messageType: `${message.type}:${message.subtype}`, payload: message, }, }; switch (message.subtype) { case "init": yield* offerRuntimeEvent({ ...base, type: "session.configured", payload: { config: message as Record, }, }); return; case "status": yield* offerRuntimeEvent({ ...base, type: "session.state.changed", payload: { state: message.status === "compacting" ? "waiting" : "running", reason: `status:${message.status ?? "active"}`, detail: message, }, }); return; case "compact_boundary": yield* offerRuntimeEvent({ ...base, type: "thread.state.changed", payload: { state: "compacted", detail: message, }, }); return; case "hook_started": yield* offerRuntimeEvent({ ...base, type: "hook.started", payload: { hookId: message.hook_id, hookName: message.hook_name, hookEvent: message.hook_event, }, }); return; case "hook_progress": yield* offerRuntimeEvent({ ...base, type: "hook.progress", payload: { hookId: message.hook_id, output: message.output, stdout: message.stdout, stderr: message.stderr, }, }); return; case "hook_response": yield* offerRuntimeEvent({ ...base, type: "hook.completed", payload: { hookId: message.hook_id, outcome: message.outcome, output: message.output, stdout: message.stdout, stderr: message.stderr, ...(typeof message.exit_code === "number" ? { exitCode: message.exit_code } : {}), }, }); return; case "task_started": yield* offerRuntimeEvent({ ...base, type: "task.started", payload: { taskId: RuntimeTaskId.makeUnsafe(message.task_id), description: message.description, ...(message.task_type ? { taskType: message.task_type } : {}), }, }); return; case "task_progress": yield* offerRuntimeEvent({ ...base, type: "task.progress", payload: { taskId: RuntimeTaskId.makeUnsafe(message.task_id), description: message.description, ...(message.summary ? { summary: message.summary } : {}), ...(message.usage ? { usage: message.usage } : {}), ...(message.last_tool_name ? { lastToolName: message.last_tool_name } : {}), }, }); return; case "task_notification": yield* offerRuntimeEvent({ ...base, type: "task.completed", payload: { taskId: RuntimeTaskId.makeUnsafe(message.task_id), status: message.status, ...(message.summary ? { summary: message.summary } : {}), ...(message.usage ? { usage: message.usage } : {}), }, }); return; case "files_persisted": yield* offerRuntimeEvent({ ...base, type: "files.persisted", payload: { files: Array.isArray(message.files) ? message.files.map((file: { filename: string; file_id: string }) => ({ filename: file.filename, fileId: file.file_id, })) : [], ...(Array.isArray(message.failed) ? { failed: message.failed.map((entry: { filename: string; error: string }) => ({ filename: entry.filename, error: entry.error, })), } : {}), }, }); return; default: yield* emitRuntimeWarning( context, `Unhandled Claude system message subtype '${message.subtype}'.`, message, ); return; } }); const handleSdkTelemetryMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { const stamp = yield* makeEventStamp(); const base = { eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), providerRefs: nativeProviderRefs(context), raw: { source: "claude.sdk.message" as const, method: sdkNativeMethod(message), messageType: message.type, payload: message, }, }; if (message.type === "tool_progress") { yield* offerRuntimeEvent({ ...base, type: "tool.progress", payload: { toolUseId: message.tool_use_id, toolName: message.tool_name, elapsedSeconds: message.elapsed_time_seconds, ...(message.task_id ? { summary: `task:${message.task_id}` } : {}), }, }); return; } if (message.type === "tool_use_summary") { yield* offerRuntimeEvent({ ...base, type: "tool.summary", payload: { summary: message.summary, ...(message.preceding_tool_use_ids.length > 0 ? { precedingToolUseIds: message.preceding_tool_use_ids } : {}), }, }); return; } if (message.type === "auth_status") { yield* offerRuntimeEvent({ ...base, type: "auth.status", payload: { isAuthenticating: message.isAuthenticating, output: message.output, ...(message.error ? { error: message.error } : {}), }, }); return; } if (message.type === "rate_limit_event") { yield* offerRuntimeEvent({ ...base, type: "account.rate-limits.updated", payload: { rateLimits: message, }, }); return; } }); const handleSdkMessage = ( context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { yield* logNativeSdkMessage(context, message); yield* ensureThreadId(context, message); switch (message.type) { case "stream_event": yield* handleStreamEvent(context, message); return; case "user": yield* handleUserMessage(context, message); return; case "assistant": yield* handleAssistantMessage(context, message); return; case "result": yield* handleResultMessage(context, message); return; case "system": yield* handleSystemMessage(context, message); return; case "tool_progress": case "tool_use_summary": case "auth_status": case "rate_limit_event": yield* handleSdkTelemetryMessage(context, message); return; default: yield* emitRuntimeWarning( context, `Unhandled Claude SDK message type '${message.type}'.`, message, ); return; } }); const runSdkStream = (context: ClaudeSessionContext): Effect.Effect => Stream.fromAsyncIterable(context.query, (cause) => cause).pipe( Stream.takeWhile(() => !context.stopped), Stream.runForEach((message) => handleSdkMessage(context, message)), Effect.catchCause((cause) => Effect.gen(function* () { if (Cause.hasInterruptsOnly(cause) || context.stopped) { return; } const message = toMessage(Cause.squash(cause), "Claude runtime stream failed."); yield* emitRuntimeError(context, message, cause); yield* completeTurn(context, "failed", message); }), ), ); const stopSessionInternal = ( context: ClaudeSessionContext, options?: { readonly emitExitEvent?: boolean }, ): Effect.Effect => Effect.gen(function* () { if (context.stopped) return; context.stopped = true; for (const [requestId, pending] of context.pendingApprovals) { yield* Deferred.succeed(pending.decision, "cancel"); const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "request.resolved", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), requestId: asRuntimeRequestId(requestId), payload: { requestType: pending.requestType, decision: "cancel", }, providerRefs: nativeProviderRefs(context), }); } context.pendingApprovals.clear(); if (context.turnState) { yield* completeTurn(context, "interrupted", "Session stopped."); } yield* Queue.shutdown(context.promptQueue); context.query.close(); const updatedAt = yield* nowIso; context.session = { ...context.session, status: "closed", activeTurnId: undefined, updatedAt, }; if (options?.emitExitEvent !== false) { const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.exited", eventId: stamp.eventId, provider: PROVIDER, createdAt: stamp.createdAt, threadId: context.session.threadId, payload: { reason: "Session stopped", exitKind: "graceful", }, providerRefs: {}, }); } sessions.delete(context.session.threadId); }); const requireSession = ( threadId: ThreadId, ): Effect.Effect => { const context = sessions.get(threadId); if (!context) { return Effect.fail( new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId, }), ); } if (context.stopped || context.session.status === "closed") { return Effect.fail( new ProviderAdapterSessionClosedError({ provider: PROVIDER, threadId, }), ); } return Effect.succeed(context); }; const startSession: ClaudeAdapterShape["startSession"] = (input) => Effect.gen(function* () { if (input.provider !== undefined && input.provider !== PROVIDER) { return yield* new ProviderAdapterValidationError({ provider: PROVIDER, operation: "startSession", issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, }); } const startedAt = yield* nowIso; const resumeState = readClaudeResumeState(input.resumeCursor); const threadId = input.threadId; const promptQueue = yield* Queue.unbounded(); const prompt = Stream.fromQueue(promptQueue).pipe( Stream.filter((item) => item.type === "message"), Stream.map((item) => item.message), Stream.toAsyncIterable, ); const pendingApprovals = new Map(); const pendingUserInputs = new Map(); const inFlightTools = new Map(); const contextRef = yield* Ref.make(undefined); /** * Handle AskUserQuestion tool calls by emitting a `user-input.requested` * runtime event and waiting for the user to respond via `respondToUserInput`. */ const handleAskUserQuestion = ( context: ClaudeSessionContext, toolInput: Record, callbackOptions: { readonly signal: AbortSignal; readonly toolUseID?: string }, ) => Effect.gen(function* () { const requestId = ApprovalRequestId.makeUnsafe(yield* Random.nextUUIDv4); // Parse questions from the SDK's AskUserQuestion input. const rawQuestions = Array.isArray(toolInput.questions) ? toolInput.questions : []; const questions: Array = rawQuestions.map( (q: Record, idx: number) => ({ id: typeof q.header === "string" ? q.header : `q-${idx}`, header: typeof q.header === "string" ? q.header : `Question ${idx + 1}`, question: typeof q.question === "string" ? q.question : "", options: Array.isArray(q.options) ? q.options.map((opt: Record) => ({ label: typeof opt.label === "string" ? opt.label : "", description: typeof opt.description === "string" ? opt.description : "", })) : [], multiSelect: typeof q.multiSelect === "boolean" ? q.multiSelect : false, }), ); const answersDeferred = yield* Deferred.make(); const pendingInput: PendingUserInput = { questions, answers: answersDeferred, }; // Emit user-input.requested so the UI can present the questions. const requestedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "user-input.requested", eventId: requestedStamp.eventId, provider: PROVIDER, createdAt: requestedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), requestId: asRuntimeRequestId(requestId), payload: { questions }, providerRefs: nativeProviderRefs(context, { providerItemId: callbackOptions.toolUseID, }), raw: { source: "claude.sdk.permission", method: "canUseTool/AskUserQuestion", payload: { toolName: "AskUserQuestion", input: toolInput }, }, }); pendingUserInputs.set(requestId, pendingInput); // Handle abort (e.g. turn interrupted while waiting for user input). const onAbort = () => { if (!pendingUserInputs.has(requestId)) { return; } pendingUserInputs.delete(requestId); Effect.runFork(Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers)); }; callbackOptions.signal.addEventListener("abort", onAbort, { once: true }); // Block until the user provides answers. const answers = yield* Deferred.await(answersDeferred); pendingUserInputs.delete(requestId); // Emit user-input.resolved so the UI knows the interaction completed. const resolvedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "user-input.resolved", eventId: resolvedStamp.eventId, provider: PROVIDER, createdAt: resolvedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), requestId: asRuntimeRequestId(requestId), payload: { answers }, providerRefs: nativeProviderRefs(context, { providerItemId: callbackOptions.toolUseID, }), raw: { source: "claude.sdk.permission", method: "canUseTool/AskUserQuestion/resolved", payload: { answers }, }, }); // Return the answers to the SDK in the expected format: // { questions: [...], answers: { questionText: selectedLabel } } return { behavior: "allow", updatedInput: { questions: toolInput.questions, answers, }, } satisfies PermissionResult; }); const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) => Effect.runPromise( Effect.gen(function* () { const context = yield* Ref.get(contextRef); if (!context) { return { behavior: "deny", message: "Claude session context is unavailable.", } satisfies PermissionResult; } // Handle AskUserQuestion: surface clarifying questions to the // user via the user-input runtime event channel, regardless of // runtime mode (plan mode relies on this heavily). if (toolName === "AskUserQuestion") { return yield* handleAskUserQuestion(context, toolInput, callbackOptions); } const runtimeMode = input.runtimeMode ?? "full-access"; if (runtimeMode === "full-access") { return { behavior: "allow", updatedInput: toolInput, } satisfies PermissionResult; } const requestId = ApprovalRequestId.makeUnsafe(yield* Random.nextUUIDv4); const requestType = classifyRequestType(toolName); const detail = summarizeToolRequest(toolName, toolInput); const decisionDeferred = yield* Deferred.make(); const pendingApproval: PendingApproval = { requestType, detail, decision: decisionDeferred, ...(callbackOptions.suggestions ? { suggestions: callbackOptions.suggestions } : {}), }; const requestedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "request.opened", eventId: requestedStamp.eventId, provider: PROVIDER, createdAt: requestedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), requestId: asRuntimeRequestId(requestId), payload: { requestType, detail, args: { toolName, input: toolInput, ...(callbackOptions.toolUseID ? { toolUseId: callbackOptions.toolUseID } : {}), }, }, providerRefs: nativeProviderRefs(context, { providerItemId: callbackOptions.toolUseID, }), raw: { source: "claude.sdk.permission", method: "canUseTool/request", payload: { toolName, input: toolInput, }, }, }); pendingApprovals.set(requestId, pendingApproval); const onAbort = () => { if (!pendingApprovals.has(requestId)) { return; } pendingApprovals.delete(requestId); Effect.runFork(Deferred.succeed(decisionDeferred, "cancel")); }; callbackOptions.signal.addEventListener("abort", onAbort, { once: true, }); const decision = yield* Deferred.await(decisionDeferred); pendingApprovals.delete(requestId); const resolvedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "request.resolved", eventId: resolvedStamp.eventId, provider: PROVIDER, createdAt: resolvedStamp.createdAt, threadId: context.session.threadId, ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), requestId: asRuntimeRequestId(requestId), payload: { requestType, decision, }, providerRefs: nativeProviderRefs(context, { providerItemId: callbackOptions.toolUseID, }), raw: { source: "claude.sdk.permission", method: "canUseTool/decision", payload: { decision, }, }, }); if (decision === "accept" || decision === "acceptForSession") { return { behavior: "allow", updatedInput: toolInput, ...(decision === "acceptForSession" && pendingApproval.suggestions ? { updatedPermissions: [...pendingApproval.suggestions] } : {}), } satisfies PermissionResult; } return { behavior: "deny", message: decision === "cancel" ? "User cancelled tool execution." : "User declined tool execution.", } satisfies PermissionResult; }), ); const providerOptions = input.providerOptions?.claudeAgent; const requestedEffort = resolveReasoningEffortForProvider( "claudeAgent", input.modelOptions?.claudeAgent?.effort ?? null, ); const supportedEffortOptions = getReasoningEffortOptions("claudeAgent", input.model); const effort = requestedEffort && supportedEffortOptions.includes(requestedEffort) ? requestedEffort : null; const fastMode = input.modelOptions?.claudeAgent?.fastMode === true && supportsClaudeFastMode(input.model); const thinking = typeof input.modelOptions?.claudeAgent?.thinking === "boolean" && supportsClaudeThinkingToggle(input.model) ? input.modelOptions.claudeAgent.thinking : undefined; const effectiveEffort = getEffectiveClaudeCodeEffort(effort); const permissionMode = toPermissionMode(providerOptions?.permissionMode) ?? (input.runtimeMode === "full-access" ? "bypassPermissions" : undefined); const settings = { ...(typeof thinking === "boolean" ? { alwaysThinkingEnabled: thinking } : {}), ...(fastMode ? { fastMode: true } : {}), }; const queryOptions: ClaudeQueryOptions = { ...(input.cwd ? { cwd: input.cwd } : {}), ...(input.model ? { model: input.model } : {}), ...(providerOptions?.binaryPath ? { pathToClaudeCodeExecutable: providerOptions.binaryPath } : {}), ...(effectiveEffort ? { effort: effectiveEffort } : {}), ...(permissionMode ? { permissionMode } : {}), ...(permissionMode === "bypassPermissions" ? { allowDangerouslySkipPermissions: true } : {}), ...(providerOptions?.maxThinkingTokens !== undefined ? { maxThinkingTokens: providerOptions.maxThinkingTokens } : {}), ...(Object.keys(settings).length > 0 ? { settings } : {}), ...(resumeState?.resume ? { resume: resumeState.resume } : {}), ...(resumeState?.resumeSessionAt ? { resumeSessionAt: resumeState.resumeSessionAt } : {}), includePartialMessages: true, canUseTool, env: process.env, ...(input.cwd ? { additionalDirectories: [input.cwd] } : {}), }; const queryRuntime = yield* Effect.try({ try: () => createQuery({ prompt, options: queryOptions, }), catch: (cause) => new ProviderAdapterProcessError({ provider: PROVIDER, threadId, detail: toMessage(cause, "Failed to start Claude runtime session."), cause, }), }); const session: ProviderSession = { threadId, provider: PROVIDER, status: "ready", runtimeMode: input.runtimeMode, ...(input.cwd ? { cwd: input.cwd } : {}), ...(input.model ? { model: input.model } : {}), ...(threadId ? { threadId } : {}), resumeCursor: { ...(threadId ? { threadId } : {}), ...(resumeState?.resume ? { resume: resumeState.resume } : {}), ...(resumeState?.resumeSessionAt ? { resumeSessionAt: resumeState.resumeSessionAt } : {}), turnCount: resumeState?.turnCount ?? 0, }, createdAt: startedAt, updatedAt: startedAt, }; const context: ClaudeSessionContext = { session, promptQueue, query: queryRuntime, startedAt, basePermissionMode: permissionMode, resumeSessionId: resumeState?.resume, pendingApprovals, pendingUserInputs, turns: [], inFlightTools, turnState: undefined, lastAssistantUuid: resumeState?.resumeSessionAt, lastThreadStartedId: undefined, stopped: false, }; yield* Ref.set(contextRef, context); sessions.set(threadId, context); const sessionStartedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.started", eventId: sessionStartedStamp.eventId, provider: PROVIDER, createdAt: sessionStartedStamp.createdAt, threadId, payload: input.resumeCursor !== undefined ? { resume: input.resumeCursor } : {}, providerRefs: {}, }); const configuredStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.configured", eventId: configuredStamp.eventId, provider: PROVIDER, createdAt: configuredStamp.createdAt, threadId, payload: { config: { ...(input.model ? { model: input.model } : {}), ...(input.cwd ? { cwd: input.cwd } : {}), ...(effectiveEffort ? { effort: effectiveEffort } : {}), ...(permissionMode ? { permissionMode } : {}), ...(providerOptions?.maxThinkingTokens !== undefined ? { maxThinkingTokens: providerOptions.maxThinkingTokens } : {}), ...(fastMode ? { fastMode: true } : {}), }, }, providerRefs: {}, }); const readyStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.state.changed", eventId: readyStamp.eventId, provider: PROVIDER, createdAt: readyStamp.createdAt, threadId, payload: { state: "ready", }, providerRefs: {}, }); Effect.runFork(runSdkStream(context)); return { ...session, }; }); const sendTurn: ClaudeAdapterShape["sendTurn"] = (input) => Effect.gen(function* () { const context = yield* requireSession(input.threadId); if (context.turnState) { // Auto-close a stale synthetic turn (from background agent responses // between user prompts) to prevent blocking the user's next turn. yield* completeTurn(context, "completed"); } if (input.model) { yield* Effect.tryPromise({ try: () => context.query.setModel(input.model), catch: (cause) => toRequestError(input.threadId, "turn/setModel", cause), }); } // Apply interaction mode by switching the SDK's permission mode. // "plan" maps directly to the SDK's "plan" permission mode; // "default" restores the session's original permission mode. // When interactionMode is absent we leave the current mode unchanged. if (input.interactionMode === "plan") { yield* Effect.tryPromise({ try: () => context.query.setPermissionMode("plan"), catch: (cause) => toRequestError(input.threadId, "turn/setPermissionMode", cause), }); } else if (input.interactionMode === "default") { yield* Effect.tryPromise({ try: () => context.query.setPermissionMode(context.basePermissionMode ?? "bypassPermissions"), catch: (cause) => toRequestError(input.threadId, "turn/setPermissionMode", cause), }); } const turnId = TurnId.makeUnsafe(yield* Random.nextUUIDv4); const turnState: ClaudeTurnState = { turnId, startedAt: yield* nowIso, items: [], assistantTextBlocks: new Map(), assistantTextBlockOrder: [], nextSyntheticAssistantBlockIndex: -1, }; const updatedAt = yield* nowIso; context.turnState = turnState; context.session = { ...context.session, status: "running", activeTurnId: turnId, updatedAt, }; const turnStartedStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "turn.started", eventId: turnStartedStamp.eventId, provider: PROVIDER, createdAt: turnStartedStamp.createdAt, threadId: context.session.threadId, turnId, payload: input.model ? { model: input.model } : {}, providerRefs: {}, }); const message = buildUserMessage(input); yield* Queue.offer(context.promptQueue, { type: "message", message, }).pipe(Effect.mapError((cause) => toRequestError(input.threadId, "turn/start", cause))); return { threadId: context.session.threadId, turnId, ...(context.session.resumeCursor !== undefined ? { resumeCursor: context.session.resumeCursor } : {}), }; }); const interruptTurn: ClaudeAdapterShape["interruptTurn"] = (threadId, _turnId) => Effect.gen(function* () { const context = yield* requireSession(threadId); yield* Effect.tryPromise({ try: () => context.query.interrupt(), catch: (cause) => toRequestError(threadId, "turn/interrupt", cause), }); }); const readThread: ClaudeAdapterShape["readThread"] = (threadId) => Effect.gen(function* () { const context = yield* requireSession(threadId); return yield* snapshotThread(context); }); const rollbackThread: ClaudeAdapterShape["rollbackThread"] = (threadId, numTurns) => Effect.gen(function* () { const context = yield* requireSession(threadId); const nextLength = Math.max(0, context.turns.length - numTurns); context.turns.splice(nextLength); yield* updateResumeCursor(context); return yield* snapshotThread(context); }); const respondToRequest: ClaudeAdapterShape["respondToRequest"] = ( threadId, requestId, decision, ) => Effect.gen(function* () { const context = yield* requireSession(threadId); const pending = context.pendingApprovals.get(requestId); if (!pending) { return yield* new ProviderAdapterRequestError({ provider: PROVIDER, method: "item/requestApproval/decision", detail: `Unknown pending approval request: ${requestId}`, }); } context.pendingApprovals.delete(requestId); yield* Deferred.succeed(pending.decision, decision); }); const respondToUserInput: ClaudeAdapterShape["respondToUserInput"] = ( threadId, requestId, answers, ) => Effect.gen(function* () { const context = yield* requireSession(threadId); const pending = context.pendingUserInputs.get(requestId); if (!pending) { return yield* new ProviderAdapterRequestError({ provider: PROVIDER, method: "item/tool/respondToUserInput", detail: `Unknown pending user-input request: ${requestId}`, }); } context.pendingUserInputs.delete(requestId); yield* Deferred.succeed(pending.answers, answers); }); const stopSession: ClaudeAdapterShape["stopSession"] = (threadId) => Effect.gen(function* () { const context = yield* requireSession(threadId); yield* stopSessionInternal(context, { emitExitEvent: true, }); }); const listSessions: ClaudeAdapterShape["listSessions"] = () => Effect.sync(() => Array.from(sessions.values(), ({ session }) => ({ ...session }))); const hasSession: ClaudeAdapterShape["hasSession"] = (threadId) => Effect.sync(() => { const context = sessions.get(threadId); return context !== undefined && !context.stopped; }); const stopAll: ClaudeAdapterShape["stopAll"] = () => Effect.forEach( sessions, ([, context]) => stopSessionInternal(context, { emitExitEvent: true, }), { discard: true }, ); yield* Effect.addFinalizer(() => Effect.forEach( sessions, ([, context]) => stopSessionInternal(context, { emitExitEvent: false, }), { discard: true }, ).pipe(Effect.tap(() => Queue.shutdown(runtimeEventQueue))), ); return { provider: PROVIDER, capabilities: { sessionModelSwitch: "in-session", }, startSession, sendTurn, interruptTurn, readThread, rollbackThread, respondToRequest, respondToUserInput, stopSession, listSessions, hasSession, stopAll, streamEvents: Stream.fromQueue(runtimeEventQueue), } satisfies ClaudeAdapterShape; }); } export const ClaudeAdapterLive = Layer.effect(ClaudeAdapter, makeClaudeAdapter()); export function makeClaudeAdapterLive(options?: ClaudeAdapterLiveOptions) { return Layer.effect(ClaudeAdapter, makeClaudeAdapter(options)); }