diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts index 274aaf76b72..f4bfde6f818 100644 --- a/src/main/claude/claude-stream-json-connection.ts +++ b/src/main/claude/claude-stream-json-connection.ts @@ -69,7 +69,7 @@ export type ClaudeStreamJsonConnectionHandlers = { canUseTool?: CanUseTool /** `request_user_dialog` control; the CLI only emits kinds declared in `supportedDialogKinds`. */ onUserDialog?: OnUserDialog - /** A transport/process fault; a physical root exit arrives marked observed and is followed by onExit. */ + /** A transport/process fault; proven root death arrives marked observed and is followed by onExit. */ onFault?: (error: Error) => void /** Report observed root exit once; expected stays true even after an unproven close attempt. */ onExit?: (error: Error, exit?: { expected: boolean }) => void @@ -178,7 +178,7 @@ export async function openClaudeStreamJsonConnection( const error = claudeStreamJsonExitError(managed.stderrTail(), managed.exitStatus, cause) terminalError = withProviderSpawnAbsence(error, managed.spawnAbsence) } - // One physical exit reaches both callbacks; mark it first so onFault cannot treat it as live-root. + // Proven root death reaches both callbacks; mark it first so onFault cannot treat it as live-root. if (managed.rootExitObserved && !closing) { withObservedProviderExit(terminalError) } diff --git a/src/main/claude/sdk-transport/claude-stream-json-connection.test.ts b/src/main/claude/sdk-transport/claude-stream-json-connection.test.ts index 333fbd25cb8..c473d6c322c 100644 --- a/src/main/claude/sdk-transport/claude-stream-json-connection.test.ts +++ b/src/main/claude/sdk-transport/claude-stream-json-connection.test.ts @@ -12,6 +12,12 @@ import { type ClaudeStreamJsonLaunch } from '../claude-stream-json-connection' import { CLAUDE_STRUCTURED_BASE_OPTIONS } from '../structured/launch/claude-structured-launch-resolution' +import { ClaudeStructuredSessionAdapter } from '../structured/claude-structured-session-adapter' +import type { ClaudeStructuredSessionEvent } from '../structured/claude-structured-session-state' +import { DaemonServer } from '../../daemon/server/daemon-server' +import { DaemonClient } from '../../daemon/transport/client/daemon-client' +import { installProviderProcessHost } from '../../provider-process/provider-process-spawn' +import * as identityProbe from '../../runtime/agent-session/ownership/agent-session-process-identity-probe' // Exercise the real SDK against a scripted child CLI. const FAKE_CLI = join(__dirname, '../__fixtures__', 'claude-agent-sdk-scripted-cli.mjs') @@ -94,6 +100,161 @@ async function open( } describe('Claude stream-json connection', () => { + it.skipIf(process.platform === 'win32')( + 'retires hosted active and idle sessions after proven root absence and permits a fresh send', + async () => { + const scenario = scriptScenario([ + { emit: { type: 'system', subtype: 'init', session_id: SESSION_ID, uuid: 'init' } }, + { awaitUserMessage: true }, + { + emit: { + type: 'user', + session_id: SESSION_ID, + uuid: 'echo', + parent_tool_use_id: null, + message: { role: 'user', content: [{ type: 'text', text: 'first' }] } + } + }, + { delayMs: 60_000 } + ]) + const server = new DaemonServer({ + socketPath: join(scenario.cwd, 'daemon.sock'), + tokenPath: join(scenario.cwd, 'token'), + providerSpawn: true, + spawnSubprocess: () => { + throw new Error('No PTY in hosted Claude fixture') + } + }) + const client = new DaemonClient({ + socketPath: join(scenario.cwd, 'daemon.sock'), + tokenPath: join(scenario.cwd, 'token') + }) + const nonces: string[] = [] + let liveObserved = false + const disconnected = new Set<() => void>() + vi.spyOn(client, 'onDisconnected').mockImplementation((listener) => { + disconnected.add(listener) + return () => disconnected.delete(listener) + }) + const events: ClaudeStructuredSessionEvent[] = [] + const probe = identityProbe.probeAgentSessionProcessIdentity + vi.spyOn(identityProbe, 'probeAgentSessionProcessIdentity').mockImplementation( + async (input) => { + const proof = await probe(input) + if (proof.outcome !== 'pid-absent') { + liveObserved = true + } + return proof + } + ) + const request = client.request.bind(client) + const querySpy = vi.spyOn(client, 'request').mockImplementation(function ( + ...args: Parameters + ): Promise { + // A restarted host has no retained outcome; use a real unknown-nonce query while keeping held signal authority. + return args[0] === 'queryProvider' + ? request('queryProvider', { nonce: 'f'.repeat(64) }) + : request(...args) + }) + const exitSpy = vi.spyOn(client, 'onProviderExit').mockImplementation((nonce) => { + nonces.push(nonce) + return () => {} + }) + const adapter = new ClaudeStructuredSessionAdapter({ + resolveLaunch: async () => ({ + ...launchFor(scenario, { ORCA_SDK_CONTRACT_TIMEOUT_MS: '60000' }), + claudeConfigDir: scenario.cwd, + providerSessionId: SESSION_ID, + resumeLeafUuid: null, + resumesTranscript: false, + continuesChain: false, + cliFlag: async () => 'supported' + }), + onEvent: (event) => { + events.push(event) + } + }) + const acquire = (sessionId: string, fence = 1) => + adapter.acquire({ + identity: { + sessionId, + workspaceId: 'hosted-fixture', + hostId: 'local', + agent: 'claude', + providerHandle: null + }, + fence, + spawnToken: `spawn-${sessionId}-${fence}` + }) + const dispatch = (sessionId: string, clientMessageId: string) => + adapter.dispatch({ + sessionId, + clientMessageId, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'first' }] } + }) + try { + await server.start() + await client.ensureConnected() + installProviderProcessHost(async () => client) + await acquire('active') + await acquire('idle') + await dispatch('active', 'before-loss') + client.disconnect() + for (const listener of disconnected) { + listener() + } + await vi.waitFor(() => expect(liveObserved).toBe(true), { timeout: 5_000 }) + expect(events.filter((event) => event.type === 'ended')).toEqual([]) + await client.ensureConnected() + for (const nonce of nonces) { + await expect( + client.request('signalProvider', { nonce, signal: 'SIGTERM' }) + ).resolves.toEqual({ + delivered: true + }) + } + await vi.waitFor( + () => expect(events.filter((event) => event.type === 'ended')).toHaveLength(2), + { timeout: 5_000 } + ) + await adapter.drainObservedExits() + expect( + events + .filter((event) => event.type === 'ended') + .map((event) => event.sessionId) + .sort() + ).toEqual(['active', 'idle']) + querySpy.mockRestore() + exitSpy.mockRestore() + events.length = 0 + await acquire('active', 2) + await expect(dispatch('active', 'after-loss')).resolves.toMatchObject({ state: 'admitted' }) + await vi.waitFor( + () => + expect(events).toContainEqual( + expect.objectContaining({ + type: 'message', + sessionId: 'active', + message: expect.objectContaining({ type: 'user', uuid: 'echo' }) + }) + ), + { timeout: 5_000 } + ) + } finally { + querySpy.mockRestore() + exitSpy.mockRestore() + try { + await adapter.closeAll() + } finally { + installProviderProcessHost(async () => null) + await server.shutdown() + client.disconnect() + vi.restoreAllMocks() + } + } + } + ) + it('completes task Stop through the real SDK while semantic reading is paused', async () => { const scenario = scriptScenario([HOLD_OPEN]) const connection = await open(launchFor(scenario)) diff --git a/src/main/provider-process/hosted-provider-child.integration.test.ts b/src/main/provider-process/hosted-provider-child.integration.test.ts index f20e7ab1868..e0747952488 100644 --- a/src/main/provider-process/hosted-provider-child.integration.test.ts +++ b/src/main/provider-process/hosted-provider-child.integration.test.ts @@ -121,10 +121,18 @@ describe.skipIf(process.platform === 'win32')('hosted provider production transp child.stderr.resume() child.stdin.end() await gate.pending - const closed = once(child, 'close') + const errors: Error[] = [] + child.on('error', (error) => errors.push(error)) + // Forced Stop can reset pending output before the child reports its actual close. + const closed = new Promise((resolve) => + child.once('close', (code, signal) => resolve([code, signal])) + ) expect(await child.signalRoot!('SIGTERM')).toBe(true) expect(await closed).toEqual([null, 'SIGTERM']) expect(child.rootAbsent).toBe(true) + for (const error of errors) { + expect(error).toMatchObject({ code: 'ECONNRESET' }) + } } finally { await gate.close() } diff --git a/src/main/provider-process/managed-provider-process.ts b/src/main/provider-process/managed-provider-process.ts index df2e9891372..b891b341d15 100644 --- a/src/main/provider-process/managed-provider-process.ts +++ b/src/main/provider-process/managed-provider-process.ts @@ -63,7 +63,7 @@ export type ManagedProviderProcess = { readonly processless: boolean /** `exited` covers a processless child too; use `rootExitObserved` for "a real process exited". */ readonly rootVerdict: DescendantTreeVerdict - /** A process that existed was seen to exit; never true for a processless child. */ + /** A process that existed exited or was proven absent; never true for a processless child. */ readonly rootExitObserved: boolean /** Retained close evidence on this handle; a late root event keeps its earlier diagnostic. */ readonly lastCloseResult: ProviderProcessCloseResult | null @@ -118,7 +118,6 @@ export function spawnManagedProviderProcess( guardSupervisedProviderGroup(child, { site: options.site, supervised: spec.supervised }) const listeners = new Set<(exit: ProviderProcessExit) => void>() let observed: ProviderProcessExit | null = null - let realExitObserved = false let spawnFailed = false let spawnError: unknown = null let lastCloseResult: ProviderProcessCloseResult | null = null @@ -167,7 +166,6 @@ export function spawnManagedProviderProcess( listeners.clear() } child.on('exit', (code, signal) => { - realExitObserved = true observeExit({ code, signal, processless: false }) }) child.on('providerAbsent', () => observeExit({ code: null, signal: null, processless: false })) @@ -264,7 +262,7 @@ export function spawnManagedProviderProcess( return rootVerdict() }, get rootExitObserved() { - return realExitObserved + return observed !== null && !observed.processless }, get executableMissing() { return spawnAbsence() === 'executable'