diff --git a/src/main/claude/claude-stream-json-connection.test.ts b/src/main/claude/claude-stream-json-connection.test.ts new file mode 100644 index 00000000000..defa3b10ab1 --- /dev/null +++ b/src/main/claude/claude-stream-json-connection.test.ts @@ -0,0 +1,128 @@ +import { EventEmitter } from 'node:events' +import { PassThrough } from 'node:stream' +import { describe, expect, it, vi } from 'vitest' +import type { spawn } from 'node:child_process' +import { + openClaudeStreamJsonConnection, + type ClaudeControlRequest +} from './claude-stream-json-connection' + +type FakeChild = EventEmitter & { + pid: number + stdin: PassThrough + stdout: PassThrough + stderr: PassThrough + kill: ReturnType +} + +function fakeSpawn() { + const child = new EventEmitter() as FakeChild + child.pid = 4321 + child.stdin = new PassThrough() + child.stdout = new PassThrough() + child.stderr = new PassThrough() + child.kill = vi.fn(() => true) + const spawnImpl = vi.fn(() => child) as unknown as typeof spawn + return { child, spawnImpl } +} + +function writtenFrames(child: FakeChild): Promise[]> { + return new Promise((resolve) => { + setImmediate(() => { + const text = child.stdin.read()?.toString('utf8') ?? '' + resolve( + text + .trim() + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as Record) + ) + }) + }) +} + +describe('Claude stream-json connection', () => { + it('spawns in the pinned workspace and routes acknowledged control requests', async () => { + const process = fakeSpawn() + const connection = await openClaudeStreamJsonConnection( + { + command: 'claude', + args: ['-p'], + cwd: '/work/repo', + env: { CLAUDE_CONFIG_DIR: '/accounts/one' } + }, + {}, + process.spawnImpl + ) + const listing = connection.request('list_models') + const outbound = await writtenFrames(process.child) + const requestId = (outbound[0] as { request_id: string }).request_id + process.child.stdout.write( + `${JSON.stringify({ + type: 'control_response', + response: { + subtype: 'success', + request_id: requestId, + response: { models: [{ value: 'sonnet' }] } + } + })}\n` + ) + + await expect(listing).resolves.toEqual({ models: [{ value: 'sonnet' }] }) + expect(process.spawnImpl).toHaveBeenCalledWith( + 'claude', + ['-p'], + expect.objectContaining({ + cwd: '/work/repo', + env: expect.objectContaining({ CLAUDE_CONFIG_DIR: '/accounts/one' }), + windowsHide: true + }) + ) + }) + + it('routes provider permission controls and writes their response envelope', async () => { + const process = fakeSpawn() + let inbound: ClaudeControlRequest | null = null + const connection = await openClaudeStreamJsonConnection( + { command: 'claude', args: [], cwd: '/work' }, + { onControlRequest: (request) => (inbound = request) }, + process.spawnImpl + ) + process.child.stdout.write( + `${JSON.stringify({ + type: 'control_request', + request_id: 'permission-1', + request: { subtype: 'can_use_tool', tool_name: 'Bash' } + })}\n` + ) + await new Promise((resolve) => setImmediate(resolve)) + expect(inbound).toMatchObject({ request_id: 'permission-1' }) + + await connection.respond('permission-1', { behavior: 'deny', message: 'No' }) + expect(await writtenFrames(process.child)).toEqual([ + { + type: 'control_response', + response: { + subtype: 'success', + request_id: 'permission-1', + response: { behavior: 'deny', message: 'No' } + } + } + ]) + }) + + it('fails closed on malformed provider output', async () => { + const process = fakeSpawn() + const onExit = vi.fn() + await openClaudeStreamJsonConnection( + { command: 'claude', args: [], cwd: '/work' }, + { onExit }, + process.spawnImpl + ) + process.child.stdout.write('{not-json}\n') + await new Promise((resolve) => setImmediate(resolve)) + + expect(process.child.kill).toHaveBeenCalled() + expect(onExit).toHaveBeenCalledWith(expect.objectContaining({ message: expect.any(String) })) + }) +}) diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts new file mode 100644 index 00000000000..21497bb0471 --- /dev/null +++ b/src/main/claude/claude-stream-json-connection.ts @@ -0,0 +1,306 @@ +import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process' +import { waitForProcessExitUntil } from '../codex/codex-process-exit-deadline' +import { killCodexAppServerProcessTree } from '../codex/codex-app-server-session' + +export type ClaudeStreamJsonLaunch = { + command: string + args: string[] + cwd: string + env?: Record +} + +export type ClaudeControlRequest = { + type: 'control_request' + request_id: string + request: Record & { subtype: string } +} + +export type ClaudeControlCancelRequest = { + type: 'control_cancel_request' + request_id: string +} + +export type ClaudeStreamJsonConnectionHandlers = { + onMessage?: (message: Record) => void + onControlRequest?: (request: ClaudeControlRequest) => void + onControlCancelRequest?: (request: ClaudeControlCancelRequest) => void + onExit?: (error: Error) => void +} + +export type ClaudeStreamJsonConnection = { + readonly pid: number | undefined + readonly closed: boolean + send: (message: Record) => Promise + request: ( + subtype: string, + params?: Record, + options?: { timeoutMs?: number } + ) => Promise + respond: (requestId: string, response: unknown) => Promise + respondWithError: (requestId: string, error: string) => Promise + close: () => Promise +} + +export class ClaudeControlRequestError extends Error { + constructor( + readonly subtype: string, + message: string + ) { + super(message) + this.name = 'ClaudeControlRequestError' + } +} + +const DEFAULT_REQUEST_TIMEOUT_MS = 30_000 +const GRACEFUL_EXIT_MS = 1_500 +const FORCED_EXIT_MS = 1_000 +const STDERR_TAIL_MAX_BYTES = 8192 +const STDOUT_LINE_MAX_BYTES = 8 * 1024 * 1024 + +type PendingRequest = { + subtype: string + resolve: (result: unknown) => void + reject: (error: Error) => void + timer: ReturnType +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function exitError(stderrTail: string, cause?: Error): Error { + const detail = stderrTail.trim() + const message = detail ? `claude stream-json exited: ${detail}` : 'claude stream-json exited' + return cause ? new Error(message, { cause }) : new Error(message) +} + +export async function openClaudeStreamJsonConnection( + launch: ClaudeStreamJsonLaunch, + handlers: ClaudeStreamJsonConnectionHandlers = {}, + spawnImpl: typeof spawn = spawn +): Promise { + const child = spawnImpl(launch.command, launch.args, { + cwd: launch.cwd, + env: { ...process.env, ...launch.env }, + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true + }) as ChildProcessWithoutNullStreams + const pending = new Map() + let nextRequestId = 1 + let stderrTail = '' + let stdoutBuffer = '' + let exited = false + let closing = false + let terminalError: Error | null = null + let writeChain: Promise = Promise.resolve() + + const exitPromise = new Promise((resolve) => { + child.on('exit', () => { + exited = true + resolve() + }) + }) + + const failPending = (error: Error): void => { + for (const waiter of pending.values()) { + clearTimeout(waiter.timer) + waiter.reject(error) + } + pending.clear() + } + + const handleUnexpectedEnd = (cause?: Error): void => { + if (terminalError) { + return + } + terminalError = exitError(stderrTail, cause) + failPending(terminalError) + if (!closing) { + handlers.onExit?.(terminalError) + } + } + + const writeLine = (payload: Record): Promise => { + const write = async (): Promise => { + if (closing || exited || terminalError || child.stdin.destroyed || !child.stdin.writable) { + throw terminalError ?? new Error('claude stream-json connection is closed') + } + const line = `${JSON.stringify(payload)}\n` + await new Promise((resolve, reject) => { + child.stdin.write(line, (error) => (error ? reject(error) : resolve())) + }) + } + const queued = writeChain.then(write) + writeChain = queued.catch(() => {}) + return queued + } + + const dispatchMessage = (message: Record): void => { + if (message.type === 'control_response') { + const response = isRecord(message.response) ? message.response : null + const requestId = typeof response?.request_id === 'string' ? response.request_id : null + const waiter = requestId ? pending.get(requestId) : undefined + if (!waiter || !requestId || !response) { + return + } + pending.delete(requestId) + clearTimeout(waiter.timer) + if (response.subtype === 'success') { + waiter.resolve(response.response) + } else { + waiter.reject( + new ClaudeControlRequestError( + waiter.subtype, + typeof response.error === 'string' + ? response.error + : `claude ${waiter.subtype} request failed` + ) + ) + } + return + } + if ( + message.type === 'control_request' && + typeof message.request_id === 'string' && + isRecord(message.request) && + typeof message.request.subtype === 'string' + ) { + handlers.onControlRequest?.(message as ClaudeControlRequest) + return + } + if (message.type === 'control_cancel_request' && typeof message.request_id === 'string') { + handlers.onControlCancelRequest?.(message as ClaudeControlCancelRequest) + return + } + handlers.onMessage?.(message) + } + + const parseLine = (line: string): void => { + if (!line.trim()) { + return + } + let parsed: unknown + try { + parsed = JSON.parse(line) + } catch (error) { + killCodexAppServerProcessTree(child) + handleUnexpectedEnd(new Error('claude emitted invalid stream-json', { cause: error })) + return + } + if (isRecord(parsed)) { + dispatchMessage(parsed) + } + } + + child.stdout.setEncoding('utf8').on('data', (chunk: string) => { + stdoutBuffer += chunk + let newline = stdoutBuffer.indexOf('\n') + while (newline !== -1) { + const line = stdoutBuffer.slice(0, newline) + stdoutBuffer = stdoutBuffer.slice(newline + 1) + if (Buffer.byteLength(line, 'utf8') > STDOUT_LINE_MAX_BYTES) { + killCodexAppServerProcessTree(child) + handleUnexpectedEnd(new Error('claude emitted an oversized stream-json line')) + return + } + parseLine(line) + newline = stdoutBuffer.indexOf('\n') + } + if (Buffer.byteLength(stdoutBuffer, 'utf8') > STDOUT_LINE_MAX_BYTES) { + killCodexAppServerProcessTree(child) + handleUnexpectedEnd(new Error('claude emitted an oversized stream-json line')) + } + }) + child.stderr.setEncoding('utf8').on('data', (chunk: string) => { + stderrTail = (stderrTail + chunk).slice(-STDERR_TAIL_MAX_BYTES) + }) + child.on('error', (error) => { + exited = true + handleUnexpectedEnd(error) + }) + child.on('close', () => { + if (stdoutBuffer.trim()) { + parseLine(stdoutBuffer) + stdoutBuffer = '' + } + handleUnexpectedEnd() + }) + child.stdin.on('error', (error) => { + if (!closing) { + killCodexAppServerProcessTree(child) + handleUnexpectedEnd(error) + } + }) + + const request = ( + subtype: string, + params: Record = {}, + options: { timeoutMs?: number } = {} + ): Promise => { + const requestId = `orca-${nextRequestId++}` + const promise = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + pending.delete(requestId) + reject(new Error(`claude ${subtype} request timed out`)) + }, options.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS) + timer.unref?.() + pending.set(requestId, { subtype, resolve, reject, timer }) + }) + void writeLine({ + type: 'control_request', + request_id: requestId, + request: { subtype, ...params } + }).catch((error) => { + const waiter = pending.get(requestId) + if (waiter) { + pending.delete(requestId) + clearTimeout(waiter.timer) + waiter.reject(error as Error) + } + }) + return promise + } + + const close = async (): Promise => { + if (closing) { + await exitPromise + return + } + closing = true + try { + child.stdin.end() + } catch { + // The reap below still owns the process. + } + if (!exited) { + await waitForProcessExitUntil(exitPromise, GRACEFUL_EXIT_MS) + if (!exited) { + killCodexAppServerProcessTree(child) + await waitForProcessExitUntil(exitPromise, FORCED_EXIT_MS) + } + } + failPending(new Error('claude stream-json connection closed')) + } + + return { + get pid() { + return child.pid + }, + get closed() { + return closing || exited || terminalError !== null + }, + send: writeLine, + request, + respond: (requestId, response) => + writeLine({ + type: 'control_response', + response: { subtype: 'success', request_id: requestId, response } + }), + respondWithError: (requestId, error) => + writeLine({ + type: 'control_response', + response: { subtype: 'error', request_id: requestId, error } + }), + close + } +} diff --git a/src/main/claude/claude-structured-control-actions.ts b/src/main/claude/claude-structured-control-actions.ts new file mode 100644 index 00000000000..1b098fb75f7 --- /dev/null +++ b/src/main/claude/claude-structured-control-actions.ts @@ -0,0 +1,30 @@ +import { applyClaudePromptAnswer } from './claude-structured-prompt-replies' +import type { ClaudeSession } from './claude-structured-session-state' + +export async function cancelClaudeTurn( + session: ClaudeSession, + timeoutMs: number | undefined +): Promise<{ cancelled: boolean }> { + try { + await session.connection.request('interrupt', {}, { timeoutMs }) + return { cancelled: true } + } catch { + return { cancelled: false } + } +} + +export async function answerClaudePrompt( + session: ClaudeSession, + input: { itemId: string; kind: 'approval' | 'question'; optionId: string } +): Promise { + const found = session.prompts.find(input.itemId) + if (!found || found.prompt.kind !== input.kind) { + throw new Error(`claude is no longer waiting on ${input.itemId}`) + } + const response = applyClaudePromptAnswer(found, input.optionId) + if (response === null) { + return + } + session.prompts.forget(found.prompt) + await session.connection.respond(found.prompt.requestId, response) +} diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts new file mode 100644 index 00000000000..ba97cd7662f --- /dev/null +++ b/src/main/claude/claude-structured-dispatch.ts @@ -0,0 +1,134 @@ +import { extname } from 'node:path' +import { readFile, stat } from 'node:fs/promises' +import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' +import type { NativeChatBlock } from '../../shared/native-chat-types' +import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { ClaudeSession } from './claude-structured-session-state' +import { readClaudeFrameString } from './claude-structured-init-proof' + +const MAX_IMAGE_BYTES = 5 * 1024 * 1024 + +const IMAGE_MIME_BY_EXTENSION: Record = { + '.gif': 'image/gif', + '.jpeg': 'image/jpeg', + '.jpg': 'image/jpeg', + '.png': 'image/png', + '.webp': 'image/webp' +} + +export function resolveClaudeReplayWaiter( + session: ClaudeSession, + message: Record +): void { + if ( + message.type !== 'user' || + message.parent_tool_use_id !== null || + readClaudeFrameString(message, 'session_id') !== session.providerSessionId + ) { + return + } + const uuid = readClaudeFrameString(message, 'uuid') + const waiter = uuid ? session.dispatchWaiters.shift() : undefined + if (waiter && uuid) { + clearTimeout(waiter.timer) + waiter.resolve(uuid) + } +} + +async function imageContent( + block: Extract +): Promise { + if (block.url) { + return { type: 'image', source: { type: 'url', url: block.url } } + } + if (!block.path) { + throw new Error('image reference has neither a path nor a URL') + } + const info = await stat(block.path) + if (!info.isFile() || info.size > MAX_IMAGE_BYTES) { + throw new Error(`Claude image must be a file no larger than ${MAX_IMAGE_BYTES} bytes`) + } + const mediaType = IMAGE_MIME_BY_EXTENSION[extname(block.path).toLowerCase()] + if (!mediaType) { + throw new Error(`Claude does not support the image type ${extname(block.path)}`) + } + return { + type: 'image', + source: { + type: 'base64', + media_type: mediaType, + data: (await readFile(block.path)).toString('base64') + } + } +} + +async function messageContent(body: AgentJournalMessageItem): Promise { + if (body.role !== 'user') { + throw new Error('Claude dispatch accepts only user messages') + } + const content: unknown[] = [] + for (const block of body.blocks as NativeChatBlock[]) { + if (block.type === 'text' && block.text.length > 0) { + content.push({ type: 'text', text: block.text }) + } else if (block.type === 'image-ref') { + content.push(await imageContent(block)) + } + } + if (content.length === 0) { + throw new Error('Claude dispatch requires text or an image') + } + return content +} + +function waitForReplay(session: ClaudeSession, timeoutMs: number): Promise { + return new Promise((resolve) => { + const waiter = { + resolve, + timer: setTimeout(() => { + const index = session.dispatchWaiters.indexOf(waiter) + if (index !== -1) { + session.dispatchWaiters.splice(index, 1) + } + resolve(null) + }, timeoutMs) + } + waiter.timer.unref?.() + session.dispatchWaiters.push(waiter) + }) +} + +export async function dispatchClaudeTurn( + session: ClaudeSession, + input: { clientMessageId: string; body: AgentJournalMessageItem }, + timeoutMs: number +): Promise { + let content: unknown[] + try { + content = await messageContent(input.body) + } catch (error) { + return { state: 'rejected', reason: (error as Error).message } + } + const replayed = waitForReplay(session, timeoutMs) + try { + await session.connection.send({ + type: 'user', + message: { role: 'user', content }, + parent_tool_use_id: null, + session_id: session.providerSessionId + }) + } catch (error) { + const waiter = session.dispatchWaiters.shift() + if (waiter) { + clearTimeout(waiter.timer) + waiter.resolve(null) + } + return { state: 'unknown', reason: (error as Error).message } + } + const uuid = await replayed + return uuid + ? { + state: 'accepted', + providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid } + } + : { state: 'unknown', reason: 'claude accepted a message but did not replay its uuid in time' } +} diff --git a/src/main/claude/claude-structured-inbound-control.ts b/src/main/claude/claude-structured-inbound-control.ts new file mode 100644 index 00000000000..26f7e1d8355 --- /dev/null +++ b/src/main/claude/claude-structured-inbound-control.ts @@ -0,0 +1,27 @@ +import type { ClaudeControlRequest } from './claude-stream-json-connection' +import type { + ClaudeAcquisitionAttempt, + ClaudeStructuredSessionEvent +} from './claude-structured-session-state' + +export function handleClaudeInboundControl(input: { + sessionId: string + attempt: ClaudeAcquisitionAttempt + request: ClaudeControlRequest + emit: (event: ClaudeStructuredSessionEvent) => void +}): void { + const connection = input.attempt.connection + if (input.request.request.subtype === 'request_user_dialog') { + void connection?.respond(input.request.request_id, { behavior: 'cancelled' }) + return + } + const prompt = input.attempt.prompts.register(input.request) + if (!prompt) { + void connection?.respondWithError( + input.request.request_id, + `Orca does not handle ${input.request.request.subtype}` + ) + return + } + input.emit({ type: 'prompt', sessionId: input.sessionId, prompt }) +} diff --git a/src/main/claude/claude-structured-init-proof.ts b/src/main/claude/claude-structured-init-proof.ts new file mode 100644 index 00000000000..f426c0ce346 --- /dev/null +++ b/src/main/claude/claude-structured-init-proof.ts @@ -0,0 +1,55 @@ +import { CLAUDE_DEFAULT_SETTING_SOURCES } from './claude-structured-launch-resolution' +import type { ClaudeAuthDiagnostic } from './claude-structured-session-state' + +export type ClaudeInitObservation = { + providerSessionId: string + uuid: string | null + message: Record +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +export function readClaudeFrameString(source: Record, key: string): string | null { + const value = source[key] + return typeof value === 'string' && value.length > 0 ? value : null +} + +export function readClaudeInit(message: Record): ClaudeInitObservation | null { + if (message.type !== 'system' || message.subtype !== 'init') { + return null + } + const providerSessionId = readClaudeFrameString(message, 'session_id') + return providerSessionId + ? { + providerSessionId, + uuid: readClaudeFrameString(message, 'uuid'), + message + } + : null +} + +export function readClaudeModels(initialization: unknown): unknown[] { + return isRecord(initialization) && Array.isArray(initialization.models) + ? initialization.models + : [] +} + +export function claudeAuthDiagnostic( + init: ClaudeInitObservation, + settings: unknown +): ClaudeAuthDiagnostic { + const env = isRecord(settings) && isRecord(settings.env) ? settings.env : {} + const apiKeySource = readClaudeFrameString(init.message, 'apiKeySource') + const configured = (key: string): boolean => + (typeof env[key] === 'string' && (env[key] as string).trim().length > 0) || + Boolean(process.env[key]?.trim()) + return { + apiKeySourceConfigured: apiKeySource !== null && apiKeySource !== 'none', + baseUrlConfigured: configured('ANTHROPIC_BASE_URL'), + authTokenConfigured: configured('ANTHROPIC_AUTH_TOKEN'), + apiKeyConfigured: configured('ANTHROPIC_API_KEY'), + settingSources: CLAUDE_DEFAULT_SETTING_SOURCES + } +} diff --git a/src/main/claude/claude-structured-launch-resolution.test.ts b/src/main/claude/claude-structured-launch-resolution.test.ts new file mode 100644 index 00000000000..4604e308752 --- /dev/null +++ b/src/main/claude/claude-structured-launch-resolution.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' +import type { AgentSessionRecordStore } from '../runtime/agent-session-record-store' +import { + CLAUDE_DEFAULT_SETTING_SOURCES, + claudeSessionIdForOrcaSession, + createClaudeStructuredLaunchResolver +} from './claude-structured-launch-resolution' + +const SESSION_ID = 'orca-session-1' +const IDENTITY = { sessionId: SESSION_ID } as Parameters< + ReturnType +>[0]['identity'] + +function record(overrides: Partial = {}): AgentSessionRecord { + return { + sessionId: SESSION_ID, + provider: 'claude', + location: { + executionHostId: LOCAL_EXECUTION_HOST_ID, + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/work/.claude' }, + providerHandleChain: [], + ...overrides + } as AgentSessionRecord +} + +function resolverFor(value: AgentSessionRecord | null) { + return createClaudeStructuredLaunchResolver({ + store: { getRecord: () => value } as unknown as AgentSessionRecordStore, + resolveWorkspacePath: async (id) => `/repos/${id}`, + resolveCommand: () => '/usr/local/bin/claude' + }) +} + +describe('claude structured launch resolution', () => { + it('pre-mints a stable provider id and pins interactive setting sources', async () => { + const first = await resolverFor(record())({ identity: IDENTITY }) + const second = await resolverFor(record())({ identity: IDENTITY }) + + expect(first.providerSessionId).toBe(claudeSessionIdForOrcaSession(SESSION_ID)) + expect(second.providerSessionId).toBe(first.providerSessionId) + expect(first).toMatchObject({ + command: '/usr/local/bin/claude', + cwd: '/repos/workspace-1', + claudeConfigDir: '/home/work/.claude', + resumeLeafUuid: null, + resumed: false + }) + expect(first.args).toContain('--session-id') + expect(first.args).toContain(first.providerSessionId) + expect(first.args).toContain('--permission-prompt-tool') + expect(first.args).toContain('stdio') + expect(first.args).toContain('--setting-sources') + expect(first.args).toContain(CLAUDE_DEFAULT_SETTING_SOURCES.join(',')) + }) + + it('resumes the session and leaf at the durable chain head', async () => { + const launch = await resolverFor( + record({ + providerHandleChain: [ + { handle: { provider: 'claude', sessionId: 'provider-old', leafUuid: 'leaf-old' } }, + { + handle: { + provider: 'claude', + sessionId: 'provider-current', + leafUuid: 'leaf-current' + } + } + ] as AgentSessionRecord['providerHandleChain'] + }) + )({ identity: IDENTITY }) + + expect(launch).toMatchObject({ + providerSessionId: 'provider-current', + resumeLeafUuid: 'leaf-current', + resumed: true + }) + expect(launch.args.slice(-2)).toEqual(['--resume', 'provider-current']) + }) + + it('refuses other hosts, WSL, providers, and account-home variables', async () => { + await expect( + resolverFor(record({ location: { ...record().location, executionHostId: 'ssh:build' } }))({ + identity: IDENTITY + }) + ).rejects.toThrow(/local host/) + await expect( + resolverFor(record({ location: { ...record().location, wslDistro: 'Ubuntu' } }))({ + identity: IDENTITY + }) + ).rejects.toThrow(/local host/) + await expect( + resolverFor(record({ provider: 'codex' } as Partial))({ + identity: IDENTITY + }) + ).rejects.toThrow(/codex session/) + await expect( + resolverFor(record({ accountHome: { variable: 'CODEX_HOME', path: '/tmp/codex' } }))({ + identity: IDENTITY + }) + ).rejects.toThrow(/CLAUDE_CONFIG_DIR/) + }) +}) diff --git a/src/main/claude/claude-structured-launch-resolution.ts b/src/main/claude/claude-structured-launch-resolution.ts new file mode 100644 index 00000000000..6103e7c4156 --- /dev/null +++ b/src/main/claude/claude-structured-launch-resolution.ts @@ -0,0 +1,93 @@ +import { createHash } from 'node:crypto' +import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' +import { agentSessionProviderHandleChainHead } from '../../shared/agent-session-provider-handle' +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' +import { resolveClaudeCommand } from '../codex-cli/command' +import type { AgentSessionRecordStore } from '../runtime/agent-session-record-store' +import { getSpawnArgsForWindows } from '../win32-utils' + +export const CLAUDE_DEFAULT_SETTING_SOURCES = ['user', 'project', 'local'] as const + +const BASE_ARGS = [ + '-p', + '--input-format', + 'stream-json', + '--output-format', + 'stream-json', + '--include-partial-messages', + '--verbose', + '--replay-user-messages', + '--permission-prompt-tool', + 'stdio', + '--setting-sources', + CLAUDE_DEFAULT_SETTING_SOURCES.join(',') +] + +export type ClaudeStructuredLaunch = { + command: string + args: string[] + cwd: string + claudeConfigDir: string + providerSessionId: string + resumeLeafUuid: string | null + resumed: boolean +} + +export type ClaudeStructuredLaunchResolverDeps = { + store: AgentSessionRecordStore + resolveWorkspacePath: (workspaceId: string) => Promise + resolveCommand?: () => string +} + +export function claudeSessionIdForOrcaSession(sessionId: string): string { + const bytes = createHash('sha256').update(`orca-claude:${sessionId}`).digest().subarray(0, 16) + bytes[6] = ((bytes[6] ?? 0) & 0x0f) | 0x40 + bytes[8] = ((bytes[8] ?? 0) & 0x3f) | 0x80 + const hex = bytes.toString('hex') + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}` +} + +export function createClaudeStructuredLaunchResolver( + deps: ClaudeStructuredLaunchResolverDeps +): (input: { identity: AgentSessionJournalIdentity }) => Promise { + return async ({ identity }) => { + const record = deps.store.getRecord(identity.sessionId) + if (!record) { + throw new Error(`no durable agent-session record for ${identity.sessionId}`) + } + if (record.provider !== 'claude') { + throw new Error(`session ${identity.sessionId} is a ${record.provider} session`) + } + if ( + record.location.executionHostId !== LOCAL_EXECUTION_HOST_ID || + record.location.wslDistro !== null + ) { + throw new Error( + `claude structured sessions run on the local host, not ${record.location.executionHostId}` + ) + } + if (record.accountHome.variable !== 'CLAUDE_CONFIG_DIR') { + throw new Error(`claude sessions pin CLAUDE_CONFIG_DIR, not ${record.accountHome.variable}`) + } + const head = agentSessionProviderHandleChainHead(record.providerHandleChain) + const providerSessionId = + head?.handle.provider === 'claude' + ? head.handle.sessionId + : claudeSessionIdForOrcaSession(identity.sessionId) + const providerArgs = + head?.handle.provider === 'claude' + ? ['--resume', providerSessionId] + : ['--session-id', providerSessionId] + const command = (deps.resolveCommand ?? resolveClaudeCommand)() + const { spawnCmd, spawnArgs } = getSpawnArgsForWindows(command, [...BASE_ARGS, ...providerArgs]) + return { + command: spawnCmd, + args: spawnArgs, + cwd: await deps.resolveWorkspacePath(record.location.workspaceId), + claudeConfigDir: record.accountHome.path, + providerSessionId, + resumeLeafUuid: head?.handle.provider === 'claude' ? head.handle.leafUuid : null, + resumed: head?.handle.provider === 'claude' + } + } +} diff --git a/src/main/claude/claude-structured-location-support.ts b/src/main/claude/claude-structured-location-support.ts new file mode 100644 index 00000000000..41b45683546 --- /dev/null +++ b/src/main/claude/claude-structured-location-support.ts @@ -0,0 +1,6 @@ +import type { AgentSessionExecutionLocation } from '../../shared/agent-session-record' +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' + +export function supportsClaudeStructuredLocation(location: AgentSessionExecutionLocation): boolean { + return location.executionHostId === LOCAL_EXECUTION_HOST_ID && location.wslDistro === null +} diff --git a/src/main/claude/claude-structured-options.ts b/src/main/claude/claude-structured-options.ts new file mode 100644 index 00000000000..937825e7666 --- /dev/null +++ b/src/main/claude/claude-structured-options.ts @@ -0,0 +1,21 @@ +import type { ClaudeSession } from './claude-structured-session-state' + +export async function setClaudeStructuredOption( + session: ClaudeSession, + input: { key: string; value: string }, + timeoutMs: number | undefined +): Promise { + const request = + input.key === 'model' + ? { subtype: 'set_model', params: { model: input.value } } + : input.key === 'permissionMode' + ? { subtype: 'set_permission_mode', params: { mode: input.value } } + : input.key === 'effort' + ? { subtype: 'apply_flag_settings', params: { settings: { effortLevel: input.value } } } + : null + if (!request) { + throw new Error(`claude stream-json has no session option named ${input.key}`) + } + await session.connection.request(request.subtype, request.params, { timeoutMs }) + session.options.set(input.key, input.value) +} diff --git a/src/main/claude/claude-structured-owner-identity.ts b/src/main/claude/claude-structured-owner-identity.ts new file mode 100644 index 00000000000..d7c54bf1d15 --- /dev/null +++ b/src/main/claude/claude-structured-owner-identity.ts @@ -0,0 +1,44 @@ +import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' +import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' +import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' +import { readProcessStartTimeMs } from '../runtime/agent-session-process-identity-probe' + +export const CLAUDE_SPAWN_TOKEN_ENV = 'ORCA_AGENT_SESSION_SPAWN_TOKEN' + +export async function claudeProcessIdentity( + input: { + identity: AgentSessionJournalIdentity + spawnToken: string + pid: number | undefined + }, + readStartTime: (pid: number) => Promise = readProcessStartTimeMs +): Promise { + if (input.pid === undefined) { + throw new Error('claude stream-json started without a pid') + } + return { + hostId: input.identity.hostId, + pid: input.pid, + processStartTimeMs: await readStartTime(input.pid), + spawnToken: input.spawnToken + } +} + +export function claudeProviderHandleLink(input: { + sessionId: string + leafUuid: string | null + resumed: boolean + fence: number + linkId?: string + observedAt: number +}): AgentSessionProviderHandleLink { + return { + linkId: + input.linkId ?? + `claude-${input.fence}-${input.sessionId}-${input.leafUuid ?? 'empty'}`.slice(0, 128), + handle: { provider: 'claude', sessionId: input.sessionId, leafUuid: input.leafUuid }, + origin: input.resumed ? 'resumed' : 'created', + mintedAtFence: input.fence, + observedAt: input.observedAt + } +} diff --git a/src/main/claude/claude-structured-prompt-replies.ts b/src/main/claude/claude-structured-prompt-replies.ts new file mode 100644 index 00000000000..564e02a8f63 --- /dev/null +++ b/src/main/claude/claude-structured-prompt-replies.ts @@ -0,0 +1,193 @@ +import type { ClaudeControlRequest } from './claude-stream-json-connection' + +export const CLAUDE_APPROVAL_DECISIONS = ['allow', 'allowForSession', 'deny', 'cancel'] as const +export type ClaudeApprovalDecision = (typeof CLAUDE_APPROVAL_DECISIONS)[number] + +export type ClaudePendingPrompt = { + requestId: string + promptKey: string + toolUseId: string + toolName: string + kind: 'approval' | 'question' + input: Record + suggestions: unknown[] + questionIds: readonly string[] + answers: Map + request: ClaudeControlRequest['request'] +} + +type PromptBinding = { + address: string + questionId?: string +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function readString(value: unknown): string | null { + return typeof value === 'string' && value.trim().length > 0 ? value : null +} + +function questionsFrom(input: Record): Record[] { + return Array.isArray(input.questions) ? input.questions.filter(isRecord) : [] +} + +function questionId(question: Record, index: number): string { + return readString(question.question) ?? readString(question.header) ?? `question-${index + 1}` +} + +export function encodeClaudeQuestionOptionId(questionId: string, answer: string): string { + return `${encodeURIComponent(questionId)}:${encodeURIComponent(answer)}` +} + +export function decodeClaudeQuestionOptionId( + optionId: string +): { questionId: string; answer: string } | null { + const separator = optionId.indexOf(':') + if (separator <= 0) { + return null + } + try { + return { + questionId: decodeURIComponent(optionId.slice(0, separator)), + answer: decodeURIComponent(optionId.slice(separator + 1)) + } + } catch { + return null + } +} + +export class ClaudePromptRegistry { + private readonly prompts = new Map() + private readonly journalBindings = new Map() + + register(control: ClaudeControlRequest): ClaudePendingPrompt | null { + if (control.request.subtype !== 'can_use_tool') { + return null + } + const toolUseId = readString(control.request.tool_use_id) + const toolName = readString(control.request.tool_name) + const input = isRecord(control.request.input) ? control.request.input : null + if (!toolUseId || !toolName || !input) { + return null + } + const questions = toolName === 'AskUserQuestion' ? questionsFrom(input) : [] + const prompt: ClaudePendingPrompt = { + requestId: control.request_id, + promptKey: control.request_id, + toolUseId, + toolName, + kind: questions.length > 0 ? 'question' : 'approval', + input, + suggestions: Array.isArray(control.request.permission_suggestions) + ? control.request.permission_suggestions + : [], + questionIds: questions.map(questionId), + answers: new Map(), + request: control.request + } + this.prompts.set(prompt.promptKey, prompt) + return prompt + } + + bindJournalItemId(journalItemId: string, promptKey: string, questionIdForItem?: string): void { + this.journalBindings.set(journalItemId, { + address: promptKey, + ...(questionIdForItem ? { questionId: questionIdForItem } : {}) + }) + } + + find(itemId: string): { prompt: ClaudePendingPrompt; questionId?: string } | null { + const binding = this.journalBindings.get(itemId) + const prompt = this.prompts.get(binding?.address ?? itemId) + return prompt + ? { prompt, ...(binding?.questionId ? { questionId: binding.questionId } : {}) } + : null + } + + cancel(requestId: string): ClaudePendingPrompt | null { + const prompt = this.prompts.get(requestId) ?? null + if (prompt) { + this.forget(prompt) + } + return prompt + } + + forget(prompt: ClaudePendingPrompt): void { + this.prompts.delete(prompt.promptKey) + for (const [itemId, binding] of this.journalBindings) { + if (binding.address === prompt.promptKey) { + this.journalBindings.delete(itemId) + } + } + } + + clear(): ClaudePendingPrompt[] { + const pending = [...this.prompts.values()] + this.prompts.clear() + this.journalBindings.clear() + return pending + } +} + +function approvalResponse(prompt: ClaudePendingPrompt, optionId: string): Record { + if (!(CLAUDE_APPROVAL_DECISIONS as readonly string[]).includes(optionId)) { + throw new Error(`${optionId} is not a Claude approval decision`) + } + const decision = optionId as ClaudeApprovalDecision + if (decision === 'allow' || decision === 'allowForSession') { + return { + behavior: 'allow', + updatedInput: prompt.input, + ...(decision === 'allowForSession' && prompt.suggestions.length > 0 + ? { updatedPermissions: prompt.suggestions } + : {}), + toolUseID: prompt.toolUseId + } + } + return { + behavior: 'deny', + message: decision === 'cancel' ? 'User stopped this turn.' : 'User denied this action.', + ...(decision === 'cancel' ? { interrupt: true } : {}), + toolUseID: prompt.toolUseId + } +} + +function questionResponse( + prompt: ClaudePendingPrompt, + optionId: string, + boundQuestionId?: string +): Record | null { + const decoded = decodeClaudeQuestionOptionId(optionId) + const selectedQuestionId = + decoded?.questionId ?? + boundQuestionId ?? + (prompt.questionIds.length === 1 ? prompt.questionIds[0] : null) + const answer = decoded?.answer ?? optionId + if (!selectedQuestionId || !prompt.questionIds.includes(selectedQuestionId)) { + throw new Error(`${optionId} does not name a question on Claude prompt ${prompt.promptKey}`) + } + prompt.answers.set(selectedQuestionId, answer) + if (prompt.questionIds.some((id) => !prompt.answers.has(id))) { + return null + } + const answers: Record = {} + for (const id of prompt.questionIds) { + answers[id] = prompt.answers.get(id) as string + } + return { + behavior: 'allow', + updatedInput: { ...prompt.input, answers }, + toolUseID: prompt.toolUseId + } +} + +export function applyClaudePromptAnswer( + found: { prompt: ClaudePendingPrompt; questionId?: string }, + optionId: string +): Record | null { + return found.prompt.kind === 'approval' + ? approvalResponse(found.prompt, optionId) + : questionResponse(found.prompt, optionId, found.questionId) +} diff --git a/src/main/claude/claude-structured-session-adapter.test.ts b/src/main/claude/claude-structured-session-adapter.test.ts new file mode 100644 index 00000000000..9b5cb797efe --- /dev/null +++ b/src/main/claude/claude-structured-session-adapter.test.ts @@ -0,0 +1,513 @@ +import { describe, expect, it } from 'vitest' +import type { + AgentJournalMessageItem, + AgentSessionJournalIdentity +} from '../../shared/agent-session-journal-types' +import type { + ClaudeStreamJsonConnection, + ClaudeStreamJsonConnectionHandlers, + ClaudeStreamJsonLaunch, + openClaudeStreamJsonConnection +} from './claude-stream-json-connection' +import { CLAUDE_SPAWN_TOKEN_ENV } from './claude-structured-owner-identity' +import { encodeClaudeQuestionOptionId } from './claude-structured-prompt-replies' +import { + ClaudeStructuredSessionAdapter, + type ClaudeStructuredLaunch, + type ClaudeStructuredSessionEvent +} from './claude-structured-session-adapter' + +const PROVIDER_SESSION_ID = '819cf9f8-e43c-4ad7-b50f-54aa158a726a' + +const USER_MESSAGE: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'ship it' }] +} + +function identityFor(sessionId = 'session-1'): AgentSessionJournalIdentity { + return { + sessionId, + workspaceId: 'workspace-1', + hostId: 'host-1', + agent: 'claude', + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: null } + } +} + +type Route = (params: Record | undefined) => unknown + +type FakeConnection = Omit & { + closed: boolean + launch: ClaudeStreamJsonLaunch + handlers: ClaudeStreamJsonConnectionHandlers + calls: { subtype: string; params?: Record }[] + sent: Record[] + replies: { requestId: string; response?: unknown; error?: string }[] + closeCount: number +} + +function fakeClaude( + options: { + initSessionId?: string + initUuid?: string + initModel?: string + initEffort?: string + exitBeforeInit?: string + settings?: unknown + replayUuid?: string | null + routes?: Record + } = {} +): { + connections: FakeConnection[] + openConnection: typeof openClaudeStreamJsonConnection + routes: Record +} { + const connections: FakeConnection[] = [] + const routes = options.routes ?? {} + const openConnection = (async (launch, handlers = {}) => { + const connection: FakeConnection = { + launch, + handlers, + calls: [], + sent: [], + replies: [], + closeCount: 0, + pid: 4321, + closed: false, + request: async (subtype, params) => { + connection.calls.push({ subtype, params }) + if (subtype === 'initialize') { + if (options.exitBeforeInit) { + handlers.onExit?.(new Error(options.exitBeforeInit)) + return { models: [] } + } + handlers.onMessage?.({ + type: 'system', + subtype: 'init', + session_id: options.initSessionId ?? PROVIDER_SESSION_ID, + uuid: options.initUuid ?? 'init-uuid', + model: options.initModel ?? 'claude-sonnet-5', + effortLevel: options.initEffort ?? 'high', + apiKeySource: 'none' + }) + return { models: [{ value: 'claude-sonnet', displayName: 'Sonnet' }] } + } + if (subtype === 'get_settings') { + return options.settings ?? { env: {} } + } + const route = routes[subtype] + return route ? route(params) : {} + }, + send: async (message) => { + connection.sent.push(message) + if (message.type === 'user' && options.replayUuid !== null) { + handlers.onMessage?.({ + ...message, + uuid: options.replayUuid ?? 'user-uuid' + }) + } + }, + respond: async (requestId, response) => { + connection.replies.push({ requestId, response }) + }, + respondWithError: async (requestId, error) => { + connection.replies.push({ requestId, error }) + }, + close: async () => { + connection.closeCount += 1 + connection.closed = true + } + } + connections.push(connection) + return connection + }) as typeof openClaudeStreamJsonConnection + return { connections, openConnection, routes } +} + +function adapterFor( + claude: ReturnType, + launch: Partial = {}, + events: ClaudeStructuredSessionEvent[] = [], + persistedHandles: unknown[] = [] +): ClaudeStructuredSessionAdapter { + return new ClaudeStructuredSessionAdapter({ + resolveLaunch: async () => ({ + command: 'claude', + args: ['-p'], + cwd: '/work/repo', + claudeConfigDir: '/accounts/claude', + providerSessionId: PROVIDER_SESSION_ID, + resumeLeafUuid: null, + resumed: false, + ...launch + }), + onEvent: (event) => events.push(event), + openConnection: claude.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => 1_700_000_000_500, + dispatchAckTimeoutMs: 10, + persistHandle: async (handle) => { + persistedHandles.push(handle) + } + }) +} + +async function acquired( + claude: ReturnType, + launch: Partial = {}, + events: ClaudeStructuredSessionEvent[] = [] +): Promise { + const adapter = adapterFor(claude, launch, events) + await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) + return adapter +} + +describe('ClaudeStructuredSessionAdapter.acquire', () => { + it('pins the account, proves init, and reports the process and chain leaf', async () => { + const claude = fakeClaude() + const events: ClaudeStructuredSessionEvent[] = [] + const adapter = adapterFor(claude, {}, events) + + const acquisition = await adapter.acquire({ + identity: identityFor(), + fence: 7, + spawnToken: 'spawn-9' + }) + + expect(claude.connections[0].launch).toMatchObject({ + cwd: '/work/repo', + env: { + [CLAUDE_SPAWN_TOKEN_ENV]: 'spawn-9', + CLAUDE_CONFIG_DIR: '/accounts/claude' + } + }) + expect(claude.connections[0].calls.slice(0, 2)).toEqual([ + { subtype: 'initialize', params: { supportedDialogKinds: [] } }, + { subtype: 'get_settings', params: {} } + ]) + expect(acquisition.process).toEqual({ + hostId: 'host-1', + pid: 4321, + processStartTimeMs: 1_700_000_000_000, + spawnToken: 'spawn-9' + }) + expect(acquisition.link).toEqual({ + linkId: `claude-7-${PROVIDER_SESSION_ID}-init-uuid`, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: 'init-uuid' }, + origin: 'created', + mintedAtFence: 7, + observedAt: 1_700_000_000_500 + }) + expect(events[0]).toMatchObject({ type: 'message', message: { subtype: 'init' } }) + }) + + it('records only non-secret effective auth-lane diagnostics', async () => { + const claude = fakeClaude({ + settings: { + env: { + ANTHROPIC_BASE_URL: 'https://gateway.example.test', + ANTHROPIC_AUTH_TOKEN: 'secret' + } + } + }) + const events: ClaudeStructuredSessionEvent[] = [] + await acquired(claude, {}, events) + + const diagnostic = events.find((event) => event.type === 'auth-diagnostic') + expect(diagnostic).toEqual({ + type: 'auth-diagnostic', + sessionId: 'session-1', + diagnostic: { + apiKeySourceConfigured: false, + baseUrlConfigured: true, + authTokenConfigured: true, + apiKeyConfigured: false, + settingSources: ['user', 'project', 'local'] + } + }) + expect(JSON.stringify(diagnostic)).not.toContain('secret') + expect(JSON.stringify(diagnostic)).not.toContain('gateway.example.test') + }) + + it('resumes the same provider id and refuses an init proof for another session', async () => { + const resumedClaude = fakeClaude() + const resumed = adapterFor(resumedClaude, { + resumed: true, + resumeLeafUuid: 'leaf-before' + }) + const acquisition = await resumed.acquire({ + identity: identityFor(), + fence: 9, + spawnToken: 'spawn-9' + }) + expect(acquisition.link.origin).toBe('resumed') + expect(acquisition.link.handle).toEqual({ + provider: 'claude', + sessionId: PROVIDER_SESSION_ID, + leafUuid: 'init-uuid' + }) + + const wrongClaude = fakeClaude({ initSessionId: 'different-session' }) + const wrong = adapterFor(wrongClaude) + await expect( + wrong.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) + ).rejects.toThrow(/expected/) + expect(wrongClaude.connections[0].closeCount).toBe(1) + }) + + it('surfaces a CLI startup failure instead of waiting for the init deadline', async () => { + const claude = fakeClaude({ exitBeforeInit: 'Claude login required' }) + const adapter = adapterFor(claude) + + await expect( + adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) + ).rejects.toThrow('Claude login required') + expect(claude.connections[0].closeCount).toBe(1) + }) +}) + +describe('ClaudeStructuredSessionAdapter turns and controls', () => { + it('accepts a dispatch only after Claude replays its provider uuid', async () => { + const claude = fakeClaude({ replayUuid: 'user-provider-uuid' }) + const adapter = await acquired(claude) + + const result = await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + + expect(result).toEqual({ + state: 'accepted', + providerIdentity: { + provider: 'claude', + sessionId: PROVIDER_SESSION_ID, + uuid: 'user-provider-uuid' + } + }) + expect(claude.connections[0].sent[0]).toMatchObject({ + type: 'user', + message: { role: 'user', content: [{ type: 'text', text: 'ship it' }] }, + session_id: PROVIDER_SESSION_ID + }) + }) + + it('leaves delivery unconfirmed when no replay uuid arrives', async () => { + const adapter = await acquired(fakeClaude({ replayUuid: null })) + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + ).resolves.toMatchObject({ state: 'unknown' }) + }) + + it('requires an acknowledged interrupt and supports controlled options', async () => { + const claude = fakeClaude() + const adapter = await acquired(claude) + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + ).resolves.toEqual({ cancelled: true }) + await adapter.setOption({ sessionId: 'session-1', key: 'model', value: 'sonnet', fence: 7 }) + expect(claude.connections[0].calls.slice(-2)).toEqual([ + { subtype: 'interrupt', params: {} }, + { subtype: 'set_model', params: { model: 'sonnet' } } + ]) + + claude.routes.interrupt = () => { + throw new Error('not running') + } + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-2', fence: 7 }) + ).resolves.toEqual({ cancelled: false }) + }) + + it('hydrates live model choices and maps the resolved current model to its CLI id', async () => { + const claude = fakeClaude({ + initModel: 'claude-sonnet-5', + routes: { + list_models: () => ({ + models: [ + { value: 'default', resolvedModel: 'claude-opus-5', displayName: 'Default' }, + { + value: 'opus', + resolvedModel: 'claude-opus-5', + displayName: 'Opus', + supportsEffort: true, + supportedEffortLevels: ['low', 'high'] + }, + { + value: 'sonnet', + resolvedModel: 'claude-sonnet-5', + displayName: 'Sonnet' + } + ] + }) + } + }) + const adapter = await acquired(claude) + + await expect(adapter.readOptions({ sessionId: 'session-1', fence: 7 })).resolves.toEqual({ + models: [ + { + id: 'opus', + label: 'Opus', + isDefault: true, + efforts: [ + { value: 'low', label: 'Low' }, + { value: 'high', label: 'High' } + ] + }, + { id: 'sonnet', label: 'Sonnet', isDefault: false, efforts: [] } + ], + current: { model: 'sonnet', effort: 'high' } + }) + }) + + it('keeps the shared Claude seed when live model discovery is unavailable', async () => { + const claude = fakeClaude({ + initModel: 'custom-model', + routes: { + list_models: () => { + throw new Error('unsupported') + } + } + }) + const adapter = await acquired(claude) + const result = await adapter.readOptions({ sessionId: 'session-1', fence: 7 }) + + expect(result.models.map((model) => model.id)).toEqual([ + 'fable', + 'opus', + 'sonnet', + 'haiku', + 'custom-model' + ]) + expect(result.current).toEqual({ model: 'custom-model', effort: 'high' }) + }) +}) + +describe('ClaudeStructuredSessionAdapter prompts', () => { + it('turns can_use_tool into an addressable durable approval callback', async () => { + const claude = fakeClaude() + const events: ClaudeStructuredSessionEvent[] = [] + const adapter = await acquired(claude, {}, events) + claude.connections[0].handlers.onControlRequest?.({ + type: 'control_request', + request_id: 'permission-1', + request: { + subtype: 'can_use_tool', + tool_name: 'Bash', + tool_use_id: 'tool-1', + input: { command: 'git status' }, + permission_suggestions: [{ type: 'addRules' }] + } + }) + expect(events.at(-1)).toMatchObject({ + type: 'prompt', + prompt: { kind: 'approval', toolName: 'Bash', promptKey: 'permission-1' } + }) + + adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1') + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'journal-approval', + kind: 'approval', + optionId: 'allowForSession', + fence: 7 + }) + expect(claude.connections[0].replies).toEqual([ + { + requestId: 'permission-1', + response: { + behavior: 'allow', + updatedInput: { command: 'git status' }, + updatedPermissions: [{ type: 'addRules' }], + toolUseID: 'tool-1' + } + } + ]) + }) + + it('collects every AskUserQuestion card before answering the one callback', async () => { + const claude = fakeClaude() + const adapter = await acquired(claude) + claude.connections[0].handlers.onControlRequest?.({ + type: 'control_request', + request_id: 'question-1', + request: { + subtype: 'can_use_tool', + tool_name: 'AskUserQuestion', + tool_use_id: 'tool-question', + input: { + questions: [ + { question: 'Library?', options: [{ label: 'Luxon' }] }, + { question: 'Ship now?', options: [{ label: 'Yes' }] } + ] + } + } + }) + adapter.bindPromptItemId('session-1', 'journal-q1', 'question-1', 'Library?') + adapter.bindPromptItemId('session-1', 'journal-q2', 'question-1', 'Ship now?') + + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'journal-q1', + kind: 'question', + optionId: encodeClaudeQuestionOptionId('Library?', 'Luxon'), + fence: 7 + }) + expect(claude.connections[0].replies).toEqual([]) + await adapter.answerPrompt({ + sessionId: 'session-1', + itemId: 'journal-q2', + kind: 'question', + optionId: encodeClaudeQuestionOptionId('Ship now?', 'Yes'), + fence: 7 + }) + expect(claude.connections[0].replies[0]).toMatchObject({ + requestId: 'question-1', + response: { + behavior: 'allow', + updatedInput: { answers: { 'Library?': 'Luxon', 'Ship now?': 'Yes' } }, + toolUseID: 'tool-question' + } + }) + }) + + it('persists the last observed leaf before graceful close', async () => { + const claude = fakeClaude() + const events: ClaudeStructuredSessionEvent[] = [] + const persistedHandles: unknown[] = [] + const adapter = adapterFor(claude, {}, events, persistedHandles) + await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) + claude.connections[0].handlers.onMessage?.({ + type: 'result', + session_id: PROVIDER_SESSION_ID, + uuid: 'final-leaf' + }) + + await adapter.closeSession('session-1') + + expect(persistedHandles).toEqual([ + { + sessionId: 'session-1', + providerSessionId: PROVIDER_SESSION_ID, + leafUuid: 'final-leaf', + fence: 7 + } + ]) + expect(events.at(-2)).toEqual({ + type: 'handle', + sessionId: 'session-1', + providerSessionId: PROVIDER_SESSION_ID, + leafUuid: 'final-leaf', + fence: 7 + }) + expect(claude.connections[0].closeCount).toBe(1) + }) +}) diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts new file mode 100644 index 00000000000..7ecc4210abc --- /dev/null +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -0,0 +1,317 @@ +import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' +import type { + AgentSessionAcquisition, + StructuredAgentSessionAdapter +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + openClaudeStreamJsonConnection, + type ClaudeControlRequest +} from './claude-stream-json-connection' +import { answerClaudePrompt, cancelClaudeTurn } from './claude-structured-control-actions' +import { dispatchClaudeTurn, resolveClaudeReplayWaiter } from './claude-structured-dispatch' +import { handleClaudeInboundControl } from './claude-structured-inbound-control' +import { + claudeAuthDiagnostic, + readClaudeFrameString, + readClaudeInit, + readClaudeModels, + type ClaudeInitObservation +} from './claude-structured-init-proof' +import { supportsClaudeStructuredLocation } from './claude-structured-location-support' +import { CLAUDE_SPAWN_TOKEN_ENV, claudeProcessIdentity } from './claude-structured-owner-identity' +import { setClaudeStructuredOption } from './claude-structured-options' +import { ClaudePromptRegistry } from './claude-structured-prompt-replies' +import { readClaudeStructuredSessionOptions } from './claude-structured-session-options' +import { createClaudeSessionPublication } from './claude-structured-session-publication' +import { + cancelClaudeAcquisitionAttempt, + ClaudeAcquisitionRegistry, + type ClaudeAcquisitionAttempt, + type ClaudeSession, + type ClaudeStructuredSessionAdapterDeps, + type ClaudeStructuredSessionEvent +} from './claude-structured-session-state' +import { + closeClaudePublishedSession, + settleClaudeDispatchWaiters +} from './claude-structured-session-close' + +export type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' +export type { + ClaudeAuthDiagnostic, + ClaudeStructuredSessionAdapterDeps, + ClaudeStructuredSessionEvent +} from './claude-structured-session-state' + +const INIT_TIMEOUT_MS = 30_000 +const DISPATCH_ACK_TIMEOUT_MS = 10_000 + +export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAdapter { + private readonly sessions = new Map() + private readonly acquisitions = new ClaudeAcquisitionRegistry() + + constructor(private readonly deps: ClaudeStructuredSessionAdapterDeps) {} + + supportsLocation = supportsClaudeStructuredLocation + + async acquire(input: { + identity: AgentSessionJournalIdentity + fence: number + spawnToken: string + events?: StructuredAgentSessionEventSink + }): Promise { + const sessionId = input.identity.sessionId + const prompts = new ClaudePromptRegistry() + const { previous, attempt } = this.acquisitions.start(sessionId, prompts) + let liveSession: ClaudeSession | null = null + let observedLeafUuid: string | null = null + let settleInit = (_init: ClaudeInitObservation): void => {} + let rejectInit = (_error: Error): void => {} + const initialized = new Promise((resolve, reject) => { + settleInit = resolve + rejectInit = reject + }) + const initTimer = setTimeout( + () => rejectInit(new Error(`claude session ${sessionId} did not emit system/init`)), + INIT_TIMEOUT_MS + ) + initTimer.unref?.() + + const onMessage = (message: Record): void => { + const init = readClaudeInit(message) + if (init) { + settleInit(init) + } + observedLeafUuid = readClaudeFrameString(message, 'uuid') ?? observedLeafUuid + if (liveSession) { + liveSession.leafUuid = observedLeafUuid + resolveClaudeReplayWaiter(liveSession, message) + } + this.deliver(attempt, sessionId, () => + this.emit(liveSession, input.events, { type: 'message', sessionId, message }) + ) + } + const onControlRequest = (request: ClaudeControlRequest): void => { + handleClaudeInboundControl({ + sessionId, + attempt, + request, + emit: (event) => + this.deliver(attempt, sessionId, () => this.emit(liveSession, input.events, event)) + }) + } + + try { + await cancelClaudeAcquisitionAttempt(previous) + this.acquisitions.assertCurrent(sessionId, attempt) + await this.closePublishedSession(sessionId) + this.acquisitions.assertCurrent(sessionId, attempt) + const launch = await this.deps.resolveLaunch({ identity: input.identity }) + observedLeafUuid = launch.resumeLeafUuid + this.acquisitions.assertCurrent(sessionId, attempt) + const open = this.deps.openConnection ?? openClaudeStreamJsonConnection + const connection = await open( + { + command: launch.command, + args: launch.args, + cwd: launch.cwd, + env: { + [CLAUDE_SPAWN_TOKEN_ENV]: input.spawnToken, + CLAUDE_CONFIG_DIR: launch.claudeConfigDir + } + }, + { + onMessage, + onControlRequest, + onControlCancelRequest: ({ request_id: requestId }) => { + const prompt = prompts.cancel(requestId) + if (prompt) { + this.deliver(attempt, sessionId, () => + this.emit(liveSession, input.events, { + type: 'prompt-cancelled', + sessionId, + promptKey: prompt.promptKey + }) + ) + } + }, + onExit: (error) => { + if (!attempt.published) { + rejectInit(error) + } + this.handleExit(sessionId, attempt, error) + } + } + ) + attempt.connection = connection + this.acquisitions.assertCurrent(sessionId, attempt) + const [initialization, init] = await Promise.all([ + connection.request( + 'initialize', + { supportedDialogKinds: [] }, + { timeoutMs: this.deps.requestTimeoutMs } + ), + initialized + ]) + const models = readClaudeModels(initialization) + this.deliver(attempt, sessionId, () => + this.emit(liveSession, input.events, { type: 'options', sessionId, models }) + ) + clearTimeout(initTimer) + this.acquisitions.assertCurrent(sessionId, attempt) + if (init.providerSessionId !== launch.providerSessionId) { + throw new Error( + `claude proved session ${init.providerSessionId}, expected ${launch.providerSessionId}` + ) + } + const settings = await connection + .request('get_settings', {}, { timeoutMs: this.deps.requestTimeoutMs }) + .catch(() => null) + this.deliver(attempt, sessionId, () => + this.emit(liveSession, input.events, { + type: 'auth-diagnostic', + sessionId, + diagnostic: claudeAuthDiagnostic(init, settings) + }) + ) + observedLeafUuid = init.uuid ?? observedLeafUuid + const process = await claudeProcessIdentity( + { ...input, pid: connection.pid }, + this.deps.readProcessStartTime + ) + this.acquisitions.assertCurrent(sessionId, attempt) + if (connection.closed) { + throw new Error(`claude stream-json for session ${sessionId} exited while being acquired`) + } + const publication = createClaudeSessionPublication({ + connection, + init, + leafUuid: observedLeafUuid, + fence: input.fence, + resumed: launch.resumed, + prompts, + events: input.events, + process, + ...(this.deps.mintLinkId ? { linkId: this.deps.mintLinkId() } : {}), + observedAt: this.deps.now?.() ?? Date.now() + }) + const acquired: AgentSessionAcquisition = publication.acquisition + liveSession = publication.session + this.acquisitions.assertCurrent(sessionId, attempt) + this.acquisitions.deleteIfCurrent(sessionId, attempt) + this.sessions.set(sessionId, liveSession) + attempt.published = true + for (const event of attempt.buffered.splice(0)) { + event() + } + return acquired + } catch (error) { + clearTimeout(initTimer) + this.acquisitions.deleteIfCurrent(sessionId, attempt) + if (this.sessions.get(sessionId)?.connection !== attempt.connection) { + prompts.clear() + await attempt.connection?.close() + } + throw error + } finally { + attempt.finish() + } + } + + private deliver(attempt: ClaudeAcquisitionAttempt, sessionId: string, event: () => void): void { + if (!attempt.published) { + attempt.buffered.push(event) + return + } + if (this.sessions.get(sessionId)?.connection === attempt.connection) { + event() + } + } + + private handleExit(sessionId: string, attempt: ClaudeAcquisitionAttempt, error: Error): void { + const session = this.sessions.get(sessionId) + if (!session || session.connection !== attempt.connection) { + return + } + this.sessions.delete(sessionId) + settleClaudeDispatchWaiters(session) + session.prompts.clear() + this.emit(session, session.events, { type: 'ended', sessionId, reason: error.message }) + } + + private emit( + _session: ClaudeSession | null, + _events: StructuredAgentSessionEventSink | undefined, + event: ClaudeStructuredSessionEvent + ): void { + this.deps.onEvent?.(event) + } + + bindPromptItemId( + sessionId: string, + journalItemId: string, + promptKey: string, + questionId?: string + ): void { + this.sessions.get(sessionId)?.prompts.bindJournalItemId(journalItemId, promptKey, questionId) + } + + dispatch: StructuredAgentSessionAdapter['dispatch'] = (input) => + dispatchClaudeTurn( + this.session(input.sessionId), + input, + this.deps.dispatchAckTimeoutMs ?? DISPATCH_ACK_TIMEOUT_MS + ) + + cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) => + cancelClaudeTurn(this.session(input.sessionId), this.deps.requestTimeoutMs) + + answerPrompt: StructuredAgentSessionAdapter['answerPrompt'] = (input) => + answerClaudePrompt(this.session(input.sessionId), input) + + setOption: StructuredAgentSessionAdapter['setOption'] = (input) => + setClaudeStructuredOption(this.session(input.sessionId), input, this.deps.requestTimeoutMs) + + readOptions = (input: { sessionId: string; fence: number }) => + readClaudeStructuredSessionOptions(this.session(input.sessionId), this.deps.requestTimeoutMs) + + releaseAcquisition(input: { sessionId: string }): Promise { + return this.closeSession(input.sessionId) + } + + async closeSession(sessionId: string): Promise { + const attempt = this.acquisitions.get(sessionId) + if (attempt) { + attempt.cancelled = true + await attempt.connection?.close() + await attempt.finished + } + await this.closePublishedSession(sessionId) + } + + private async closePublishedSession(sessionId: string): Promise { + await closeClaudePublishedSession({ + sessions: this.sessions, + sessionId, + ...(this.deps.persistHandle ? { persistHandle: this.deps.persistHandle } : {}), + ...(this.deps.onEvent ? { onEvent: this.deps.onEvent } : {}) + }) + } + + async closeAll(): Promise { + this.acquisitions.close() + while (this.sessions.size > 0 || this.acquisitions.size > 0) { + const ids = new Set([...this.sessions.keys(), ...this.acquisitions.sessionIds()]) + await Promise.all([...ids].map((sessionId) => this.closeSession(sessionId))) + } + } + + private session(sessionId: string): ClaudeSession { + const session = this.sessions.get(sessionId) + if (!session) { + throw new Error(`no live claude stream-json session for ${sessionId}`) + } + return session + } +} diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts new file mode 100644 index 00000000000..c486923dbc9 --- /dev/null +++ b/src/main/claude/claude-structured-session-close.ts @@ -0,0 +1,66 @@ +import type { ClaudeSession, ClaudeStructuredSessionEvent } from './claude-structured-session-state' + +export function settleClaudeDispatchWaiters(session: ClaudeSession): void { + for (const waiter of session.dispatchWaiters.splice(0)) { + clearTimeout(waiter.timer) + waiter.resolve(null) + } +} + +export async function closeClaudePublishedSession(input: { + sessions: Map + sessionId: string + persistHandle?: (handle: { + sessionId: string + providerSessionId: string + leafUuid: string | null + fence: number + }) => Promise + onEvent?: (event: ClaudeStructuredSessionEvent) => void +}): Promise { + const session = input.sessions.get(input.sessionId) + if (!session) { + return + } + input.sessions.delete(input.sessionId) + settleClaudeDispatchWaiters(session) + const pending = session.prompts.clear() + await Promise.allSettled( + pending.map((prompt) => + session.connection.respond(prompt.requestId, { + behavior: 'deny', + message: 'Structured Claude session closed.', + interrupt: true, + toolUseID: prompt.toolUseId + }) + ) + ) + let persistenceError: unknown + try { + await input.persistHandle?.({ + sessionId: input.sessionId, + providerSessionId: session.providerSessionId, + leafUuid: session.leafUuid, + fence: session.fence + }) + input.onEvent?.({ + type: 'handle', + sessionId: input.sessionId, + providerSessionId: session.providerSessionId, + leafUuid: session.leafUuid, + fence: session.fence + }) + } catch (error) { + persistenceError = error + } finally { + input.onEvent?.({ + type: 'ended', + sessionId: input.sessionId, + reason: 'claude session closed' + }) + await session.connection.close() + } + if (persistenceError) { + throw persistenceError + } +} diff --git a/src/main/claude/claude-structured-session-options.ts b/src/main/claude/claude-structured-session-options.ts new file mode 100644 index 00000000000..72a86479eac --- /dev/null +++ b/src/main/claude/claude-structured-session-options.ts @@ -0,0 +1,108 @@ +import type { + AgentSessionModelOption, + AgentSessionOptionChoice, + AgentSessionOptionsResult +} from '../../shared/agent-session-wire' +import { CLAUDE_SESSION_OPTION_CATALOG } from '../../shared/agent-session-option-catalog-claude-codex' +import type { CatalogModel } from '../../shared/agent-session-option-catalog-types' +import type { ClaudeSession } from './claude-structured-session-state' + +type ListedModel = AgentSessionModelOption & { resolvedModel: string | null } + +function record(value: unknown): Record | null { + return typeof value === 'object' && value !== null && !Array.isArray(value) + ? (value as Record) + : null +} + +function text(value: unknown): string | null { + return typeof value === 'string' && value.trim() ? value : null +} + +function effortLabel(value: string): string { + return value === 'xhigh' ? 'Extra high' : `${value.charAt(0).toUpperCase()}${value.slice(1)}` +} + +function listedEfforts(row: Record): AgentSessionOptionChoice[] { + return row.supportsEffort === true && Array.isArray(row.supportedEffortLevels) + ? row.supportedEffortLevels.flatMap((value) => { + const effort = text(value) + return effort ? [{ value: effort, label: effortLabel(effort) }] : [] + }) + : [] +} + +function listedModels(value: unknown): ListedModel[] { + const response = record(value) + const rows = Array.isArray(response?.models) + ? response.models.map(record).filter((row): row is Record => row !== null) + : [] + const defaultRow = rows.find((row) => text(row.value) === 'default') + const defaultResolvedModel = text(defaultRow?.resolvedModel) + const seen = new Set() + return rows.flatMap((row) => { + const id = text(row.value) + if (!id || id === 'default' || seen.has(id)) { + return [] + } + seen.add(id) + const resolvedModel = text(row.resolvedModel) + const description = text(row.description) + return [ + { + id, + label: text(row.displayName) ?? id, + ...(description ? { description } : {}), + isDefault: resolvedModel !== null && resolvedModel === defaultResolvedModel, + efforts: listedEfforts(row), + resolvedModel + } + ] + }) +} + +function seedEfforts(model: CatalogModel): AgentSessionOptionChoice[] { + const effort = model.options.find((option) => option.id === 'effort') + return effort?.kind.type === 'select' ? effort.kind.choices : [] +} + +function seedModels(): ListedModel[] { + return CLAUDE_SESSION_OPTION_CATALOG.models.map((model) => ({ + id: model.id, + label: model.label, + ...(model.description ? { description: model.description } : {}), + isDefault: model.isDefault === true, + efforts: seedEfforts(model), + resolvedModel: null + })) +} + +function currentModelId(models: ListedModel[], reportedModel: string | undefined): string { + const matched = reportedModel + ? models.find((model) => model.id === reportedModel || model.resolvedModel === reportedModel) + : undefined + return ( + matched?.id ?? reportedModel ?? models.find((model) => model.isDefault)?.id ?? models[0]!.id + ) +} + +export async function readClaudeStructuredSessionOptions( + session: ClaudeSession, + timeoutMs: number | undefined +): Promise { + const response = await session.connection + .request('list_models', {}, { timeoutMs }) + .catch(() => null) + const discovered = listedModels(response) + const models = discovered.length > 0 ? discovered : seedModels() + const reportedModel = session.options.get('model') ?? session.reportedOptions.model + const model = currentModelId(models, reportedModel) + if (!models.some((entry) => entry.id === model)) { + models.push({ id: model, label: model, isDefault: false, efforts: [], resolvedModel: null }) + } + const effort = session.options.get('effort') ?? session.reportedOptions.effort + return { + models: models.map(({ resolvedModel: _, ...entry }) => entry), + current: { model, ...(effort ? { effort } : {}) } + } +} diff --git a/src/main/claude/claude-structured-session-publication.ts b/src/main/claude/claude-structured-session-publication.ts new file mode 100644 index 00000000000..521a08dd839 --- /dev/null +++ b/src/main/claude/claude-structured-session-publication.ts @@ -0,0 +1,48 @@ +import type { AgentSessionAcquisition } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import { readClaudeFrameString, type ClaudeInitObservation } from './claude-structured-init-proof' +import { claudeProviderHandleLink } from './claude-structured-owner-identity' +import type { ClaudePromptRegistry } from './claude-structured-prompt-replies' +import type { ClaudeSession } from './claude-structured-session-state' + +export function createClaudeSessionPublication(input: { + connection: ClaudeSession['connection'] + init: ClaudeInitObservation + leafUuid: string | null + fence: number + resumed: boolean + prompts: ClaudePromptRegistry + events: ClaudeSession['events'] + process: AgentSessionAcquisition['process'] + linkId?: string + observedAt: number +}): { acquisition: AgentSessionAcquisition; session: ClaudeSession } { + const model = readClaudeFrameString(input.init.message, 'model') + const effort = readClaudeFrameString(input.init.message, 'effortLevel') + return { + acquisition: { + process: input.process, + link: claudeProviderHandleLink({ + sessionId: input.init.providerSessionId, + leafUuid: input.leafUuid, + resumed: input.resumed, + fence: input.fence, + ...(input.linkId ? { linkId: input.linkId } : {}), + observedAt: input.observedAt + }) + }, + session: { + connection: input.connection, + providerSessionId: input.init.providerSessionId, + leafUuid: input.leafUuid, + fence: input.fence, + prompts: input.prompts, + dispatchWaiters: [], + options: new Map(), + reportedOptions: { + ...(model ? { model } : {}), + ...(effort ? { effort } : {}) + }, + events: input.events + } + } +} diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts new file mode 100644 index 00000000000..ad2277a8b76 --- /dev/null +++ b/src/main/claude/claude-structured-session-state.ts @@ -0,0 +1,155 @@ +import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { + ClaudeStreamJsonConnection, + openClaudeStreamJsonConnection +} from './claude-stream-json-connection' +import type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' +import type { ClaudePendingPrompt, ClaudePromptRegistry } from './claude-structured-prompt-replies' + +export type ClaudeAuthDiagnostic = { + apiKeySourceConfigured: boolean + baseUrlConfigured: boolean + authTokenConfigured: boolean + apiKeyConfigured: boolean + settingSources: readonly string[] +} + +export type ClaudeStructuredSessionEvent = + | { type: 'message'; sessionId: string; message: Record } + | { type: 'prompt'; sessionId: string; prompt: ClaudePendingPrompt } + | { type: 'prompt-cancelled'; sessionId: string; promptKey: string } + | { type: 'options'; sessionId: string; models: unknown[] } + | { + type: 'handle' + sessionId: string + providerSessionId: string + leafUuid: string | null + fence: number + } + | { type: 'auth-diagnostic'; sessionId: string; diagnostic: ClaudeAuthDiagnostic } + | { type: 'ended'; sessionId: string; reason: string } + +export type ClaudeStructuredSessionAdapterDeps = { + resolveLaunch: (input: { + identity: AgentSessionJournalIdentity + }) => Promise + onEvent?: (event: ClaudeStructuredSessionEvent) => void + openConnection?: typeof openClaudeStreamJsonConnection + readProcessStartTime?: (pid: number) => Promise + mintLinkId?: () => string + now?: () => number + requestTimeoutMs?: number + dispatchAckTimeoutMs?: number + persistHandle?: (input: { + sessionId: string + providerSessionId: string + leafUuid: string | null + fence: number + }) => Promise +} + +export type ClaudeDispatchWaiter = { + resolve: (uuid: string | null) => void + timer: ReturnType +} + +export type ClaudeSession = { + connection: ClaudeStreamJsonConnection + providerSessionId: string + leafUuid: string | null + fence: number + prompts: ClaudePromptRegistry + dispatchWaiters: ClaudeDispatchWaiter[] + options: Map + reportedOptions: { model?: string; effort?: string } + events: StructuredAgentSessionEventSink | undefined +} + +export type ClaudeAcquisitionAttempt = { + connection: ClaudeStreamJsonConnection | null + prompts: ClaudePromptRegistry + buffered: (() => void)[] + published: boolean + cancelled: boolean + finished: Promise + finish: () => void +} + +export function createClaudeAcquisitionAttempt( + prompts: ClaudePromptRegistry +): ClaudeAcquisitionAttempt { + let finish = (): void => {} + const finished = new Promise((resolve) => { + finish = resolve + }) + return { + connection: null, + prompts, + buffered: [], + published: false, + cancelled: false, + finished, + finish + } +} + +export class ClaudeAcquisitionRegistry { + private readonly attempts = new Map() + private closing = false + + get size(): number { + return this.attempts.size + } + + start( + sessionId: string, + prompts: ClaudePromptRegistry + ): { + previous: ClaudeAcquisitionAttempt | undefined + attempt: ClaudeAcquisitionAttempt + } { + if (this.closing) { + throw new Error('claude structured session adapter is closing') + } + const previous = this.attempts.get(sessionId) + const attempt = createClaudeAcquisitionAttempt(prompts) + this.attempts.set(sessionId, attempt) + return { previous, attempt } + } + + assertCurrent(sessionId: string, attempt: ClaudeAcquisitionAttempt): void { + if (this.closing || attempt.cancelled || this.attempts.get(sessionId) !== attempt) { + throw new Error(`claude session ${sessionId} was superseded while being acquired`) + } + } + + get(sessionId: string): ClaudeAcquisitionAttempt | undefined { + return this.attempts.get(sessionId) + } + + deleteIfCurrent(sessionId: string, attempt: ClaudeAcquisitionAttempt): void { + if (this.attempts.get(sessionId) === attempt) { + this.attempts.delete(sessionId) + } + } + + sessionIds(): IterableIterator { + return this.attempts.keys() + } + + close(): void { + this.closing = true + } +} + +export async function cancelClaudeAcquisitionAttempt( + attempt: ClaudeAcquisitionAttempt | undefined +): Promise { + if (!attempt) { + return + } + attempt.cancelled = true + await attempt.connection?.close() + await attempt.finished +}