fix Claude acquisition exit proof

This commit is contained in:
Merge Sim
2026-08-31 20:24:02 -07:00
parent b76c8125f6
commit a397cb8f01
7 changed files with 158 additions and 50 deletions
@@ -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)
})
})
@@ -45,7 +45,8 @@ export type ClaudeStreamJsonConnection = {
) => Promise<unknown>
respond: (requestId: string, response: unknown) => Promise<void>
respondWithError: (requestId: string, error: string) => Promise<void>
close: () => Promise<void>
/** Resolves true only after the child emitted exit/close; false is unproven. */
close: () => Promise<boolean>
}
export class ClaudeControlRequestError extends Error {
@@ -96,7 +97,7 @@ export async function openClaudeStreamJsonConnection(
let closing = false
let terminalError: Error | null = null
let writeChain: Promise<void> = Promise.resolve()
let closePromise: Promise<void> | null = null
let closePromise: Promise<boolean> | null = null
let settleExit = (): void => {}
const exitPromise = new Promise<void>((resolve) => {
@@ -251,7 +252,7 @@ export async function openClaudeStreamJsonConnection(
return promise
}
const close = (): Promise<void> => {
const close = (): Promise<boolean> => {
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
}
@@ -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<boolean>>()
.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()
@@ -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<boolean> {
return this.closeSession(input.sessionId)
}
releaseAcquisition = (input: { sessionId: string }): Promise<boolean> =>
this.closeSession(input.sessionId)
async closeSession(sessionId: string): Promise<boolean> {
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<void> {
await closeClaudePublishedSession({
sessions: this.sessions,
closeSession = (sessionId: string): Promise<boolean> =>
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<void> {
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<void> =>
closeAllClaudeSessions({
sessions: this.sessions,
acquisitions: this.acquisitions,
closeSession: this.closeSession
})
private session(sessionId: string): ClaudeSession {
const session = this.sessions.get(sessionId)
@@ -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<void>
onEvent?: (event: ClaudeStructuredSessionEvent) => void
}): Promise<void> {
}): Promise<boolean> {
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<string, ClaudeSession>,
sessionId: string,
deps: {
persistHandle?: (handle: {
sessionId: string
providerSessionId: string
leafUuid: string | null
fence: number
}) => Promise<void>
onEvent?: (event: ClaudeStructuredSessionEvent) => void
}
): Promise<boolean> {
return closeClaudePublishedSession({ sessions, sessionId, ...deps })
}
export async function closeClaudeSession(input: {
sessionId: string
sessions: Map<string, ClaudeSession>
acquisitions: ClaudeAcquisitionRegistry
persistHandle?: (handle: {
sessionId: string
providerSessionId: string
leafUuid: string | null
fence: number
}) => Promise<void>
onEvent?: (event: ClaudeStructuredSessionEvent) => void
}): Promise<boolean> {
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<string, ClaudeSession>
acquisitions: ClaudeAcquisitionRegistry
closeSession: (sessionId: string) => Promise<boolean>
}): Promise<void> {
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'
})
}
@@ -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<void>
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<string> {
return this.attempts.keys()
}
@@ -150,11 +163,16 @@ export class ClaudeAcquisitionRegistry {
export async function cancelClaudeAcquisitionAttempt(
attempt: ClaudeAcquisitionAttempt | undefined
): Promise<void> {
): Promise<boolean> {
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
})
}
@@ -101,6 +101,7 @@ function fakeClaude() {
respondWithError: async () => {},
close: async () => {
connection.closed = true
return true
}
}
connections.push(connection)