mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 18:37:24 +00:00
Publish Claude exits after hosted process absence is proven
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 <T>(
|
||||
...args: Parameters<typeof request>
|
||||
): Promise<T> {
|
||||
// A restarted host has no retained outcome; use a real unknown-nonce query while keeping held signal authority.
|
||||
return args[0] === 'queryProvider'
|
||||
? request<T>('queryProvider', { nonce: 'f'.repeat(64) })
|
||||
: request<T>(...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))
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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'
|
||||
|
||||
Reference in New Issue
Block a user