diff --git a/src/main/claude/claude-stream-json-connection.test.ts b/src/main/claude/claude-stream-json-connection.test.ts index 02d610bc226..b6de3a24680 100644 --- a/src/main/claude/claude-stream-json-connection.test.ts +++ b/src/main/claude/claude-stream-json-connection.test.ts @@ -193,9 +193,9 @@ describe('Claude stream-json connection', () => { process.child.emit('error', new Error('spawn missing-claude ENOENT')) await expect(Promise.all([connection.close(), connection.close()])).resolves.toEqual([ - undefined, - undefined + true, + true ]) - await expect(connection.close()).resolves.toBeUndefined() + await expect(connection.close()).resolves.toBe(true) }) }) diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts index fc47e43581e..0d5d7e01488 100644 --- a/src/main/claude/claude-stream-json-connection.ts +++ b/src/main/claude/claude-stream-json-connection.ts @@ -45,7 +45,8 @@ export type ClaudeStreamJsonConnection = { ) => Promise respond: (requestId: string, response: unknown) => Promise respondWithError: (requestId: string, error: string) => Promise - close: () => Promise + /** Resolves true only after the child emitted exit/close; false is unproven. */ + close: () => Promise } export class ClaudeControlRequestError extends Error { @@ -96,7 +97,7 @@ export async function openClaudeStreamJsonConnection( let closing = false let terminalError: Error | null = null let writeChain: Promise = Promise.resolve() - let closePromise: Promise | null = null + let closePromise: Promise | null = null let settleExit = (): void => {} const exitPromise = new Promise((resolve) => { @@ -251,7 +252,7 @@ export async function openClaudeStreamJsonConnection( return promise } - const close = (): Promise => { + const close = (): Promise => { closePromise ??= (async () => { closing = true try { @@ -267,6 +268,11 @@ export async function openClaudeStreamJsonConnection( } } failPending(new Error('claude stream-json connection closed')) + const proven = exited + if (!proven) { + closePromise = null + } + return proven })() return closePromise } diff --git a/src/main/claude/claude-structured-session-adapter.test.ts b/src/main/claude/claude-structured-session-adapter.test.ts index 4bd3fc6e652..41d59d1e189 100644 --- a/src/main/claude/claude-structured-session-adapter.test.ts +++ b/src/main/claude/claude-structured-session-adapter.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import type { AgentJournalMessageItem, AgentSessionJournalIdentity @@ -135,6 +135,7 @@ function fakeClaude( close: async () => { connection.closeCount += 1 connection.closed = true + return true } } connections.push(connection) @@ -540,6 +541,27 @@ describe('ClaudeStructuredSessionAdapter turns and controls', () => { }) }) +describe('ClaudeStructuredSessionAdapter acquisition cleanup', () => { + it('reports unproven published-session cleanup so callers can retry safely', async () => { + const claude = fakeClaude() + const adapter = await acquired(claude) + const connection = claude.connections[0] + connection.close = vi + .fn<() => Promise>() + .mockResolvedValueOnce(false) + .mockResolvedValueOnce(true) as unknown as FakeConnection['close'] + + await expect(adapter.releaseAcquisition({ sessionId: 'session-1' })).resolves.toBe(false) + expect(await adapter.readOptions({ sessionId: 'session-1', fence: 7 })).toMatchObject({ + current: { model: 'claude-sonnet-5' } + }) + await expect(adapter.releaseAcquisition({ sessionId: 'session-1' })).resolves.toBe(true) + expect(() => adapter.readOptions({ sessionId: 'session-1', fence: 7 })).toThrow( + 'no live claude stream-json session' + ) + }) +}) + describe('ClaudeStructuredSessionAdapter prompts', () => { it('turns can_use_tool into an addressable durable approval callback', async () => { const claude = fakeClaude() diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 711bf933c89..464e11e3a3c 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -1,3 +1,4 @@ +import { AgentSessionAcquisitionExitUnprovenError } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { AgentSessionAcquisition, StructuredAgentSessionAcquireInput, @@ -45,7 +46,9 @@ import { type ClaudeStructuredSessionEvent } from './claude-structured-session-state' import { - closeClaudePublishedSession, + closeAllClaudeSessions, + closeClaudePublishedSessionForDeps, + closeClaudeSession, settleClaudeExitedSession } from './claude-structured-session-close' @@ -110,9 +113,18 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda } try { - await cancelClaudeAcquisitionAttempt(previous) + if (previous && !(await cancelClaudeAcquisitionAttempt(previous))) { + this.acquisitions.restoreIfCurrent(sessionId, attempt, previous) + throw new AgentSessionAcquisitionExitUnprovenError( + new Error(`claude acquisition for session ${sessionId} could not be stopped`) + ) + } this.acquisitions.assertCurrent(sessionId, attempt) - await this.closePublishedSession(sessionId) + if (!(await closeClaudePublishedSessionForDeps(this.sessions, sessionId, this.deps))) { + throw new AgentSessionAcquisitionExitUnprovenError( + new Error(`claude session ${sessionId} could not be stopped`) + ) + } this.acquisitions.assertCurrent(sessionId, attempt) const launch = await this.deps.resolveLaunch({ identity: input.identity }) observedLeafUuid = launch.resumeLeafUuid @@ -213,12 +225,15 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda return acquired } catch (error) { initDeadline.clear() - this.acquisitions.deleteIfCurrent(sessionId, attempt) if (this.sessions.get(sessionId)?.connection !== attempt.connection) { translator?.dispose() prompts.clear() - await attempt.connection?.close() + const closed = (await attempt.connection?.close()) ?? true + if (!closed) { + throw new AgentSessionAcquisitionExitUnprovenError(error) + } } + this.acquisitions.deleteIfCurrent(sessionId, attempt) throw error } finally { attempt.finish() @@ -272,47 +287,31 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda 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) - } + releaseAcquisition = (input: { sessionId: string }): Promise => + 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) - return true - } - - private async closePublishedSession(sessionId: string): Promise { - await closeClaudePublishedSession({ - sessions: this.sessions, + closeSession = (sessionId: string): Promise => + closeClaudeSession({ sessionId, + sessions: this.sessions, + acquisitions: this.acquisitions, ...(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))) - } - } + closeAll = (): Promise => + closeAllClaudeSessions({ + sessions: this.sessions, + acquisitions: this.acquisitions, + closeSession: this.closeSession + }) private session(sessionId: string): ClaudeSession { const session = this.sessions.get(sessionId) diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts index 590e175a6d3..ee2bf51e1ef 100644 --- a/src/main/claude/claude-structured-session-close.ts +++ b/src/main/claude/claude-structured-session-close.ts @@ -1,4 +1,10 @@ -import type { ClaudeSession, ClaudeStructuredSessionEvent } from './claude-structured-session-state' +import type { + ClaudeAcquisitionRegistry, + ClaudeSession, + ClaudeStructuredSessionEvent +} from './claude-structured-session-state' +import { cancelClaudeAcquisitionAttempt } from './claude-structured-session-state' +import { closeProcessRegistry } from '../../shared/child-process/close-process-registry' export function settleClaudeDispatchWaiters(session: ClaudeSession): void { for (const waiter of session.dispatchWaiters.splice(0)) { @@ -23,12 +29,11 @@ export async function closeClaudePublishedSession(input: { fence: number }) => Promise onEvent?: (event: ClaudeStructuredSessionEvent) => void -}): Promise { +}): Promise { const session = input.sessions.get(input.sessionId) if (!session) { - return + return true } - input.sessions.delete(input.sessionId) settleClaudeDispatchWaiters(session) const pending = session.prompts.clear() await Promise.allSettled( @@ -41,6 +46,10 @@ export async function closeClaudePublishedSession(input: { }) ) ) + if ((await session.connection.close()) !== true) { + return false + } + input.sessions.delete(input.sessionId) let persistenceError: unknown try { await input.persistHandle?.({ @@ -67,9 +76,62 @@ export async function closeClaudePublishedSession(input: { session.translator?.handle(ended) input.onEvent?.(ended) session.translator?.dispose() - await session.connection.close() } if (persistenceError) { throw persistenceError } + return true +} + +export function closeClaudePublishedSessionForDeps( + sessions: Map, + sessionId: string, + deps: { + persistHandle?: (handle: { + sessionId: string + providerSessionId: string + leafUuid: string | null + fence: number + }) => Promise + onEvent?: (event: ClaudeStructuredSessionEvent) => void + } +): Promise { + return closeClaudePublishedSession({ sessions, sessionId, ...deps }) +} + +export async function closeClaudeSession(input: { + sessionId: string + sessions: Map + acquisitions: ClaudeAcquisitionRegistry + persistHandle?: (handle: { + sessionId: string + providerSessionId: string + leafUuid: string | null + fence: number + }) => Promise + onEvent?: (event: ClaudeStructuredSessionEvent) => void +}): Promise { + const attempt = input.acquisitions.get(input.sessionId) + if (!(await cancelClaudeAcquisitionAttempt(attempt))) { + return false + } + if (attempt) { + input.acquisitions.deleteIfCurrent(input.sessionId, attempt) + } + return closeClaudePublishedSession(input) +} + +export async function closeAllClaudeSessions(input: { + sessions: Map + acquisitions: ClaudeAcquisitionRegistry + closeSession: (sessionId: string) => Promise +}): Promise { + input.acquisitions.close() + await closeProcessRegistry({ + attempts: 3, + hasEntries: () => input.sessions.size > 0 || input.acquisitions.size > 0, + entryIds: () => new Set([...input.sessions.keys(), ...input.acquisitions.sessionIds()]), + closeEntry: input.closeSession, + failureMessage: 'claude structured session shutdown could not prove every child stopped' + }) } diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index 9a8462f6ac2..a766f86955e 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -7,6 +7,7 @@ import type { import type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' import type { ClaudeJournalTranslator } from './claude-structured-journal-translation' import type { ClaudePendingPrompt, ClaudePromptRegistry } from './claude-structured-prompt-replies' +import { cancelProcessAcquisition } from '../../shared/child-process/cancel-process-acquisition' export type ClaudeAuthDiagnostic = { apiKeySourceConfigured: boolean @@ -77,6 +78,7 @@ export type ClaudeAcquisitionAttempt = { buffered: (() => void)[] published: boolean cancelled: boolean + exitProven: boolean finished: Promise finish: () => void } @@ -94,6 +96,7 @@ export function createClaudeAcquisitionAttempt( buffered: [], published: false, cancelled: false, + exitProven: false, finished, finish } @@ -139,6 +142,16 @@ export class ClaudeAcquisitionRegistry { } } + restoreIfCurrent( + sessionId: string, + replacement: ClaudeAcquisitionAttempt, + previous: ClaudeAcquisitionAttempt + ): void { + if (this.attempts.get(sessionId) === replacement) { + this.attempts.set(sessionId, previous) + } + } + sessionIds(): IterableIterator { return this.attempts.keys() } @@ -150,11 +163,16 @@ export class ClaudeAcquisitionRegistry { export async function cancelClaudeAcquisitionAttempt( attempt: ClaudeAcquisitionAttempt | undefined -): Promise { +): Promise { if (!attempt) { - return + return true } - attempt.cancelled = true - await attempt.connection?.close() - await attempt.finished + return cancelProcessAcquisition({ + cancel: () => { + attempt.cancelled = true + }, + connection: () => attempt.connection, + exitProven: () => attempt.exitProven, + finished: attempt.finished + }) } diff --git a/src/main/runtime/claude-structured-session-integration.test.ts b/src/main/runtime/claude-structured-session-integration.test.ts index 60983506f49..ac2a99fa898 100644 --- a/src/main/runtime/claude-structured-session-integration.test.ts +++ b/src/main/runtime/claude-structured-session-integration.test.ts @@ -101,6 +101,7 @@ function fakeClaude() { respondWithError: async () => {}, close: async () => { connection.closed = true + return true } } connections.push(connection)