Settle refused agent launches and preserve exit observation wakes

This commit is contained in:
Brennan Benson
2026-11-13 15:24:40 -08:00
parent 9e22856f4b
commit b567fdbcc8
25 changed files with 583 additions and 166 deletions
@@ -58,9 +58,11 @@ folder prompts naming it, and a new Full Disk Access row for every user.
The helper's bundle, executable and display name are `Orca Agents`; its identifier, signing team
and designated requirement still match Orca.app. This names the existing terminal-agent process
without changing its privacy client. Native chat providers still run under the GUI process. An
adopted older daemon keeps its original name until it exits; the fork fallback and Windows names
are unchanged.
without changing its privacy client. Supported local macOS native chat providers also use the
independent host when it supports provider spawning and its executable can be resolved. Older or
unsupported hosts, an unavailable host executable, fork fallback and remote/server routes keep
direct launching under their existing owner. An adopted older daemon keeps its original name until
it exits; the fork fallback and Windows names are unchanged.
## macOS floor
@@ -62,6 +62,7 @@ function withoutKeys(env: Record<string, string>, keys: readonly string[]): Reco
}
export type ClaudeStreamJsonConnectionHandlers = {
onConnection?: (connection: ClaudeStreamJsonConnection) => void
onMessage?: (message: Record<string, unknown>) => void
/** Raw stdout before the SDK reads it, for state that must keep wire order with permission requests. */
onStdout?: (chunk: Buffer | string) => void
@@ -297,12 +298,8 @@ export async function openClaudeStreamJsonConnection(
return closePromise
}
await child.ready?.catch(async (error: unknown) => {
await close()
throw error
})
const controls = createClaudeControlSurface(session)
return {
const connection: ClaudeStreamJsonConnection = {
...controls,
startupFailure: startup.failure,
exited: exited.promise,
@@ -330,4 +327,10 @@ export async function openClaudeStreamJsonConnection(
send,
close
}
handlers.onConnection?.(connection)
await child.ready?.catch(async (error: unknown) => {
await close()
throw error
})
return connection
}
@@ -107,6 +107,30 @@ async function open(
}
describe('Claude stream-json connection', () => {
it('hands acquisition the actual connection before a pending startup can reject', async () => {
const scenario = scriptScenario([HOLD_OPEN])
const ready = Promise.withResolvers<void>()
const retained = Promise.withResolvers<ClaudeStreamJsonConnection>()
const opening = openClaudeStreamJsonConnection(
launchFor(scenario),
{
onConnection: (connection) => retained.resolve(connection)
},
(spec) => {
const child = spawnProcess(spec)
spawnedChildren.push(child)
childClosures.push(new Promise<void>((resolve) => child.once('close', () => resolve())))
return Object.assign(child, { ready: ready.promise })
}
)
const rejected = expect(opening).rejects.toThrow('pending startup failed')
const connection = await retained.promise
openConnections.push(connection)
ready.reject(new Error('pending startup failed'))
await rejected
await expect(connection.close()).resolves.toBe(true)
})
it.skipIf(process.platform === 'win32')(
'retires hosted active and idle sessions after proven root absence and permits a fresh send',
async () => {
@@ -1,3 +1,4 @@
import { AgentSessionAcquisitionExitUnprovenError } from '../../native-chat/agent-session-wire/acquisition/structured-agent-session-acquisition-errors'
import { describe, expect, it, vi } from 'vitest'
import type { AgentSessionJournalIdentity } from '../../../shared/agent-session/journal/agent-session-journal-identity'
import { AgentSessionPreSpawnError } from '../../native-chat/agent-session-wire/providers/structured-agent-session-adapter-errors'
@@ -17,60 +18,86 @@ const IDENTITY: AgentSessionJournalIdentity = {
}
describe('Claude structured processless acquisition', () => {
it('classifies pre-pid error and close as processless with idempotent cleanup', async () => {
const fault = new Error('spawn claude ENOENT')
const close = vi.fn(async () => true)
const openConnection: typeof openClaudeStreamJsonConnection = async (
_launch,
handlers = {}
) => {
// A failed spawn reports its error on a later tick, as child_process does.
setTimeout(() => handlers.onFault?.(fault), 0)
const connection: ClaudeStreamJsonConnection = {
pid: undefined,
closed: true,
exitVerdict: { root: 'exited', tree: 'exited', processless: true },
initializationResult: async () => {
it.each(['processless', 'unproven-open'] as const)(
'retains the failed-open owner until %s cleanup is proven',
async (mode) => {
const fault = new Error('spawn claude ENOENT')
let cleanupProven = mode === 'processless'
const close = vi.fn(async () => cleanupProven)
const openConnection: typeof openClaudeStreamJsonConnection = async (
_launch,
handlers = {}
) => {
// A failed spawn reports its error on a later tick, as child_process does.
if (mode === 'processless') {
setTimeout(() => handlers.onFault?.(fault), 0)
}
const connection: ClaudeStreamJsonConnection = {
pid: undefined,
closed: true,
exitVerdict:
mode === 'processless'
? { root: 'exited', tree: 'exited', processless: true }
: { root: 'live', tree: 'unverifiable' },
initializationResult: async () => {
throw fault
},
getSettings: async () => ({}),
getContextUsage: async () => ({}),
renameSession: async () => {},
supportedModels: async () => [],
interrupt: async () => undefined,
cancelAsyncMessage: async () => false,
setModel: async () => {},
setPermissionMode: async () => {},
applyFlagSettings: async () => {},
send: async () => {},
stopTask: async () => {},
close
}
if (mode === 'unproven-open') {
handlers.onConnection?.(connection)
throw fault
},
getSettings: async () => ({}),
getContextUsage: async () => ({}),
renameSession: async () => {},
supportedModels: async () => [],
interrupt: async () => undefined,
cancelAsyncMessage: async () => false,
setModel: async () => {},
setPermissionMode: async () => {},
applyFlagSettings: async () => {},
send: async () => {},
stopTask: async () => {},
close
}
return connection
}
const adapter = new ClaudeStructuredSessionAdapter({
resolveLaunch: async () => ({
pathToClaudeCodeExecutable: 'claude',
options: {},
cwd: '/work/repo',
claudeConfigDir: '/accounts/claude',
providerSessionId: PROVIDER_SESSION_ID,
resumeLeafUuid: null,
resumesTranscript: false,
cliFlag: async () => 'supported' as const,
continuesChain: false
}),
openConnection
})
const error = await adapter
.acquire({ identity: IDENTITY, fence: 7, spawnToken: 'spawn-9' })
.catch((cause: unknown) => cause)
if (mode === 'unproven-open') {
expect(error).toBeInstanceOf(AgentSessionAcquisitionExitUnprovenError)
await expect(adapter.releaseAcquisition({ sessionId: IDENTITY.sessionId })).resolves.toBe(
false
)
cleanupProven = true
await expect(adapter.releaseAcquisition({ sessionId: IDENTITY.sessionId })).resolves.toBe(
true
)
} else {
expect(error).toBeInstanceOf(AgentSessionPreSpawnError)
expect(error).toMatchObject({ message: fault.message })
expect(close).toHaveBeenCalledOnce()
await expect(adapter.releaseAcquisition({ sessionId: IDENTITY.sessionId })).resolves.toBe(
true
)
expect(close).toHaveBeenCalledOnce()
}
return connection
}
const adapter = new ClaudeStructuredSessionAdapter({
resolveLaunch: async () => ({
pathToClaudeCodeExecutable: 'claude',
options: {},
cwd: '/work/repo',
claudeConfigDir: '/accounts/claude',
providerSessionId: PROVIDER_SESSION_ID,
resumeLeafUuid: null,
resumesTranscript: false,
cliFlag: async () => 'supported' as const,
continuesChain: false
}),
openConnection
})
const error = await adapter
.acquire({ identity: IDENTITY, fence: 7, spawnToken: 'spawn-9' })
.catch((cause: unknown) => cause)
expect(error).toBeInstanceOf(AgentSessionPreSpawnError)
expect(error).toMatchObject({ message: fault.message })
expect(close).toHaveBeenCalledOnce()
await expect(adapter.releaseAcquisition({ sessionId: IDENTITY.sessionId })).resolves.toBe(true)
expect(close).toHaveBeenCalledOnce()
})
)
})
@@ -209,6 +209,7 @@ export async function acquireClaudeSession({
{
onMessage,
...permissions.handlers,
onConnection: (connection) => void (attempt.connection = connection),
onFault: childFailed,
onExit: permissions.closeThen((error, exit) => {
if (exit?.expected) {
@@ -1,4 +1,5 @@
// Why: daemons survive app updates, so wire behavior must be version-gated.
// v58 records authoritative provider refusals and fences retired hosts.
// v57 adds capability-gated provider child ownership on the signed macOS host.
// v56 accepts `reportFocusLost`; v55 daemons from 1.4.251 predate it and reject the request.
// v55 observes the foreground group at the physical PTY owner.
@@ -15,7 +16,7 @@
// v43 dropped it with the revert); v41 stages long startup commands as sourced scripts; v40 rolled
// the #25130/#24636 shell-wrapper changes and the wider agent list into a fresh daemon; older owners
// stay attachable.
export const PROTOCOL_VERSION = 57
export const PROTOCOL_VERSION = 58
export const FOCUS_LOSS_REPORT_PROTOCOL_VERSION = 56
export const PTY_FOREGROUND_OBSERVATION_PROTOCOL_VERSION = 55
export const FENCED_TERMINATION_PROTOCOL_VERSION = 51
@@ -53,7 +54,7 @@ export const MODE_2031_UNSUBSCRIBE_FACT_PROTOCOL_VERSION = 29
export const PREVIOUS_DAEMON_PROTOCOL_VERSIONS = [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27,
28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51,
52, 53, 54, 55, 56
52, 53, 54, 55, 56, 57
] as const
export function supportsColorQueryReplyColors(protocolVersion: number): boolean {
@@ -34,4 +34,4 @@ export type HostedProviderRequest =
}
| { id: string; type: 'queryProvider'; payload: { nonce: string; acknowledge?: boolean } }
export const PROVIDER_SPAWN_PROTOCOL_VERSION = 57
export const PROVIDER_SPAWN_PROTOCOL_VERSION = 58
@@ -1,3 +1,4 @@
import { PROVIDER_SPAWN_PROTOCOL_VERSION } from '../../protocol/daemon-provider-process-protocol'
import type { DaemonClient } from '../../transport/client/daemon-client'
import { createPtyExitPayload, emitPtyListeners } from './daemon-pty-listener-emission'
import { DaemonPtyDaemonRecovery } from './daemon-pty-daemon-recovery'
@@ -27,7 +28,7 @@ export abstract class DaemonPtyPreservedHostLoss extends DaemonPtyDaemonRecovery
}
getProviderProcessClient(): DaemonClient | null {
return this.protocolVersion >= 57 ? this.client : null
return this.protocolVersion >= PROVIDER_SPAWN_PROTOCOL_VERSION ? this.client : null
}
protected recoverAfterEndpointLoss(): void {
@@ -69,6 +69,24 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
rmSync(dir, { recursive: true, force: true })
})
it.each([57, 58])(
'preserves v%i terminals and only gives corrected hosts new provider work',
async (protocolVersion) => {
const harness = await startDaemonAdapterHarness(() => createMockSubprocess(), {
protocolVersion
})
try {
const terminal = await harness.adapter.spawn({ cols: 80, rows: 24 })
expect(await harness.adapter.probePtyLiveness(terminal.id)).toBe(true)
expect(harness.adapter.getProviderProcessClient() !== null).toBe(protocolVersion === 58)
} finally {
harness.adapter.dispose()
await harness.server.shutdown()
rmSync(harness.dir, { recursive: true, force: true })
}
}
)
describe('background stream thinning compatibility', () => {
// Provider fallback can route unknown IDs here; false authority would let the renderer discard their panes.
it('refuses an authoritative snapshot claim for a session it does not own', async () => {
@@ -42,3 +42,38 @@ it.each([10, 100])('drains %i live roots without rescanning retained outcomes',
await providers.dispose()
owner.destroy()
})
it('retains known no-starts before publishing and never replaces a live nonce on refusal', async () => {
const owner = new Socket()
const providers = new DaemonProviderProcesses(true, () => {})
const published: unknown[] = []
vi.spyOn(owner, 'write').mockImplementation((record) => {
const event = JSON.parse(String(record))
expect(providers.query(event.nonce, owner)).toEqual({ state: 'exited', outcome: event.outcome })
published.push(event.outcome)
return true
})
const spec = { nonce: 'a'.repeat(64), source: '', args: [], cwd: '/tmp', env: {} }
try {
await expect(providers.spawn({ ...spec, cwd: undefined! }, owner)).rejects.toThrow(
'Invalid provider launch'
)
expect(published).toEqual([{ pid: null, code: null, signal: null, processless: true }])
const child = Object.assign(new EventEmitter(), { pid: 43, kill: vi.fn(() => true) })
vi.mocked(spawnProcess).mockImplementation(() => {
queueMicrotask(() => child.emit('spawn'))
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Registry fixture supplies its held-child PID/events/kill.
return child as never
})
const liveSpec = { ...spec, nonce: 'b'.repeat(64) }
await providers.spawn(liveSpec, owner)
await expect(providers.spawn(liveSpec, owner, false)).rejects.toThrow('already dispatched')
expect(providers.query(liveSpec.nonce, owner)).toEqual({ state: 'live', pid: 43 })
expect(published).toHaveLength(1)
child.emit('exit', 0, null)
child.emit('close')
} finally {
await providers.dispose()
owner.destroy()
}
})
@@ -10,6 +10,8 @@ import type {
const OUTCOME_RETENTION_MS = 60_000
const STOP_TIMEOUT_MS = 7_500
const validLaunchString = (value: unknown): value is string =>
typeof value === 'string' && !value.includes('\0')
const SIGNALS = new Set(['SIGTERM', 'SIGINT', 'SIGHUP', 'SIGUSR2', 'SIGSTOP', 'SIGCONT', 'SIGKILL'])
type ProviderEntry = {
@@ -36,25 +38,45 @@ export class DaemonProviderProcesses {
return this.live.size > 0
}
async spawn(spec: HostedProviderSpawn, owner: Socket): Promise<{ pid: number }> {
if (!this.enabled || this.stopping) {
throw new Error('Provider host unavailable')
}
if (
!/^[a-f0-9]{64}$/.test(spec.nonce) ||
typeof spec.source !== 'string' ||
!Array.isArray(spec.args) ||
!spec.args.every((arg) => typeof arg === 'string') ||
typeof spec.cwd !== 'string' ||
!spec.env ||
typeof spec.env !== 'object' ||
!Object.values(spec.env).every((value) => typeof value === 'string')
) {
async spawn(
spec: HostedProviderSpawn,
owner: Socket,
accepting = true
): Promise<{ pid: number }> {
if (!spec || typeof spec.nonce !== 'string' || !/^[a-f0-9]{64}$/.test(spec.nonce)) {
throw new Error('Invalid provider launch')
}
if (this.live.has(spec.nonce) || this.outcomes.has(spec.nonce)) {
const retained = this.outcomes.get(spec.nonce)
if (retained) {
this.publishOutcome(spec.nonce, owner, retained.outcome)
}
throw new Error('Provider launch already dispatched')
}
const refuse = (message: string): never => {
this.retainOutcome(spec.nonce, owner, {
pid: null,
code: null,
signal: null,
processless: true
})
throw new Error(message)
}
if (!this.enabled || this.stopping || !accepting) {
refuse('Provider host unavailable')
}
if (
!validLaunchString(spec.source) ||
!Array.isArray(spec.args) ||
!spec.args.every(validLaunchString) ||
!validLaunchString(spec.cwd) ||
!spec.env ||
typeof spec.env !== 'object' ||
!Object.keys(spec.env).every(validLaunchString) ||
!Object.values(spec.env).every(validLaunchString)
) {
refuse('Invalid provider launch')
}
// The current GUI supplies its matching supervisor; the host supplies only its pinned runtime.
const child = spawnProcess({
program: process.execPath,
@@ -79,12 +101,7 @@ export class DaemonProviderProcesses {
return
}
this.live.delete(spec.nonce)
if (!entry.owner.destroyed) {
entry.owner.write(encodeNdjson({ type: 'providerExit', nonce: spec.nonce, outcome }))
}
const expiry = setTimeout(() => this.removeOutcome(spec.nonce), OUTCOME_RETENTION_MS)
expiry.unref()
this.outcomes.set(spec.nonce, { outcome, expiry })
this.retainOutcome(spec.nonce, entry.owner, outcome)
this.onLiveChanged()
}
child.once('exit', (code, signal) =>
@@ -171,6 +188,19 @@ export class DaemonProviderProcesses {
}
}
private retainOutcome(nonce: string, owner: Socket, outcome: HostedProviderOutcome): void {
const expiry = setTimeout(() => this.removeOutcome(nonce), OUTCOME_RETENTION_MS)
expiry.unref()
this.outcomes.set(nonce, { outcome, expiry })
this.publishOutcome(nonce, owner, outcome)
}
private publishOutcome(nonce: string, owner: Socket, outcome: HostedProviderOutcome): void {
if (!owner.destroyed) {
owner.write(encodeNdjson({ type: 'providerExit', nonce, outcome }))
}
}
private removeOutcome(nonce: string): void {
clearTimeout(this.outcomes.get(nonce)?.expiry)
this.outcomes.delete(nonce)
@@ -1,3 +1,4 @@
import type { Socket } from 'node:net'
import type { DaemonClientConnections, ConnectedDaemonClient } from './daemon-client-connections'
import type { DaemonTerminalAdmission } from './daemon-terminal-admission'
import type { TerminalHost } from '../terminal-host/terminal-host'
@@ -8,23 +9,24 @@ import type { DaemonServerLifecycle } from './daemon-server-lifecycle'
export function routeProviderRequest(
request: Extract<DaemonRequest, { type: 'spawnProvider' | 'signalProvider' | 'queryProvider' }>,
client: ConnectedDaemonClient | undefined,
options: { providerProcesses?: DaemonProviderProcesses; lifecycle: DaemonServerLifecycle }
options: { providerProcesses?: DaemonProviderProcesses; lifecycle: DaemonServerLifecycle },
socket: Socket
): unknown {
const providers = options.providerProcesses
if (
!client?.authenticatedPairEstablished ||
!providers?.enabled ||
!options.lifecycle.isAcceptingWork()
client.controlSocket !== socket ||
!providers?.enabled
) {
throw new Error('Provider host unavailable')
}
if (request.type === 'spawnProvider') {
return providers.spawn(request.payload, client.controlSocket)
return providers.spawn(request.payload, socket, options.lifecycle.isAcceptingProviderWork())
}
if (request.type === 'signalProvider') {
return providers.signal(request.payload.nonce, client.controlSocket, request.payload.signal)
return providers.signal(request.payload.nonce, socket, request.payload.signal)
}
return providers.query(request.payload.nonce, client.controlSocket, request.payload.acknowledge)
return providers.query(request.payload.nonce, socket, request.payload.acknowledge)
}
export function canRetireDaemon(
@@ -1,3 +1,4 @@
import type { Socket } from 'node:net'
import { canRetireDaemon, routeProviderRequest } from './daemon-provider-requests'
import type { DaemonProviderProcesses } from './daemon-provider-processes'
import { parseDaemonLaunchNoteChange } from '../protocol/daemon-launch-note-protocol'
@@ -42,13 +43,13 @@ export class DaemonRequestRouter {
this.health = new DaemonHealthRequests(options.ptySpawnHealthCheck, options.log)
}
async route(clientId: string, request: DaemonRequest): Promise<unknown> {
async route(clientId: string, request: DaemonRequest, socket?: Socket): Promise<unknown> {
const client = this.options.connections.get(clientId)
switch (request.type) {
case 'spawnProvider':
case 'signalProvider':
case 'queryProvider':
return routeProviderRequest(request, client, this.options)
return routeProviderRequest(request, client, this.options, socket ?? client!.controlSocket)
case 'startHistorySeedTransfer': {
if (!client?.authenticatedPairEstablished || client.streamSocket === null) {
throw new Error('Daemon client connection is incomplete; reconnect')
@@ -46,6 +46,10 @@ export class DaemonServerLifecycle {
return this.state === 'running'
}
isAcceptingProviderWork(): boolean {
return this.isAcceptingWork() && !this.options.endpoint.hasLostOwnership()
}
async start(handleConnection: (socket: Socket) => void): Promise<void> {
return new Promise((resolve, reject) => {
this.server = createServer(handleConnection)
+1 -1
View File
@@ -304,7 +304,7 @@ export class DaemonServer {
() => this.lifecycle.isAcceptingWork() && !this.endpoint.hasLostOwnership()
) ??
answerAutomaticWorkRequest(request, this.automaticWork, this.endpoint) ??
this.requestRouter.route(clientId, request),
this.requestRouter.route(clientId, request, socket),
() => this.lifecycle.startPendingShutdownReply(clientId, request.id)
)
}
@@ -1,3 +1,6 @@
import { waitForPiSpawn } from './rpc-startup-open'
import { JsonlRpcAgentConnection } from '../../jsonl-rpc/agent-connection'
import { EventEmitter } from 'node:events'
import { readFileSync } from 'node:fs'
import { PassThrough } from 'node:stream'
import { JsonlRpcPeer } from '../../jsonl-rpc/peer'
@@ -126,3 +129,32 @@ it('acquires a hosted Pi only after its real PID arrives', async () => {
await expect(restarting).resolves.toMatchObject({ acquired: { process: { pid: 5678 } } })
expect(h.onSpawned).toHaveBeenCalledWith(expect.objectContaining({ pid: 5678 }))
})
it('owns rejection of the real startup getter without discarding another promise', async () => {
const ready = Promise.withResolvers<void>()
const child = Object.assign(new EventEmitter(), {
pid: undefined,
ready: ready.promise,
stdin: new PassThrough(),
stdout: new PassThrough(),
stderr: new PassThrough(),
connected: false,
killed: false,
exitCode: null,
signalCode: null,
kill: vi.fn(() => false)
})
const connection = new JsonlRpcAgentConnection({ command: '/fake/pi', args: [] }, {}, () => {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Pending child fixture supplies streams, readiness and held lifecycle consumed by the production connection.
return child as never
})
const waiting = waitForPiSpawn(connection, new AbortController().signal)
const rejected = expect(waiting).rejects.toThrow('host refused startup')
ready.reject(new Error('host refused startup'))
await rejected
child.emit('error', Object.assign(new Error('not started'), { code: 'ENOENT' }))
child.stdout.end()
child.stderr.end()
child.emit('close', null, null)
await connection.close()
})
+3 -2
View File
@@ -70,7 +70,8 @@ export async function waitForPiSpawn(
connection: PiRpcConnection,
signal: AbortSignal
): Promise<void> {
if (connection.spawned) {
await waitForPromiseWithSignal(connection.spawned, signal)
const spawned = connection.spawned
if (spawned) {
await waitForPromiseWithSignal(spawned, signal)
}
}
@@ -1,6 +1,8 @@
import { runLocalSourceControlPlan } from '../text-generation/execution/source-control-local-process'
import { runCodexProcessWithHomeLock } from '../text-generation/execution/source-control-local-generation'
import { stopSupervisedChildProcess } from './supervised-child-process-stop'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { mkdtemp, realpath, rm } from 'node:fs/promises'
import { mkdtemp, realpath, rm, rename, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { once } from 'node:events'
@@ -52,6 +54,7 @@ describe.skipIf(process.platform === 'win32')('hosted provider production transp
await server.shutdown()
client.disconnect()
children.clear()
vi.restoreAllMocks()
await rm(directory, { recursive: true, force: true })
})
@@ -87,6 +90,71 @@ describe.skipIf(process.platform === 'win32')('hosted provider production transp
return child
}
it.each(['shutdown', 'endpoint'] as const)(
'settles a %s no-start before its refusal and releases the Codex home',
async (cause) => {
const releaseDisposal = Promise.withResolvers<void>()
const dispose = server.host.dispose.bind(server.host)
vi.spyOn(server.host, 'dispose').mockImplementation(async () => {
await releaseDisposal.promise
await dispose()
})
const request = client.request.bind(client)
let shutdown: Promise<void> | undefined
vi.spyOn(client, 'request').mockImplementation(async function <T>(
...args: Parameters<typeof request>
): Promise<T> {
if (args[0] === 'spawnProvider') {
if (cause === 'shutdown') {
shutdown = server.shutdown()
} else {
await rename(join(directory, 'daemon.sock'), join(directory, 'retired.sock'))
await writeFile(join(directory, 'daemon.sock'), 'replacement endpoint')
}
}
return request<T>(...args)
})
let child!: HostedProviderChild
const first = runCodexProcessWithHomeLock(directory, () =>
runLocalSourceControlPlan({
plan: { binary: 'codex', args: [], label: 'Codex', stdinPayload: null },
cwd: directory,
env: undefined,
operation: 'commit-message',
spawnAgent: () => {
child = launch('throw new Error("must never launch")')
return Object.assign(child, { supervised: true })
}
})
)
try {
expect(await first).toMatchObject({ success: false })
expect(child.spawnAbsenceKnown).toBe(true)
expect(child.pid).toBeUndefined()
expect(child.rootAbsent).toBe(true)
const next = runCodexProcessWithHomeLock(directory, () =>
runLocalSourceControlPlan({
plan: {
binary: process.execPath,
args: ['-e', 'process.stdout.write("next generation")'],
label: 'Codex',
stdinPayload: null
},
cwd: directory,
env: undefined,
operation: 'commit-message',
spawnAgent: (spec) =>
spawn(spec.binary, [...spec.args], { cwd: spec.cwd, env: spec.env, stdio: 'pipe' })
})
)
expect(await next).toMatchObject({ success: true, rawOutput: 'next generation' })
} finally {
releaseDisposal.resolve()
await shutdown
}
}
)
it('waits for real final writes after source EOF without expiring a live owner', async () => {
const gate = await createSupervisorOutputGate(directory)
try {
@@ -1,22 +1,28 @@
import { AsyncLocalStorage } from 'node:async_hooks'
import { retainProviderOutput } from './provider-output-handoff'
import { observeHostedProviderExit } from './hosted-provider-exit-observer'
import {
observeHostedProviderExit,
probeHostedProviderRoot,
queryHostedProviderState
} from './hosted-provider-exit-observer'
import type { ChildProcessWithoutNullStreams } from 'node:child_process'
import type { DaemonClient } from '../daemon/transport/client/daemon-client'
import type {
HostedProviderOutcome,
HostedProviderState
} from '../daemon/protocol/daemon-provider-process-protocol'
import type { HostedProviderOutcome } from '../daemon/protocol/daemon-provider-process-protocol'
import { ProviderOwnerListener, watchProviderControl } from './provider-owner-channels'
import { ProviderChildHandle } from './provider-child-handle'
import type { createProviderSpawnSpec } from './provider-process-supervisor'
import { probeAgentSessionProcessIdentity } from '../runtime/agent-session/ownership/agent-session-process-identity-probe'
import { sameDaemonIdentity } from '../daemon/transport/client/daemon-client-hello-handshake'
import { createHostedEnv, inspectHostedIdentity } from './hosted-provider-launch'
import {
createHostedLaunch,
inspectHostedIdentity,
withProviderStartDeadline
} from './hosted-provider-launch'
type ProviderSpawnSpec = ReturnType<typeof createProviderSpawnSpec>
const START_TIMEOUT_MS = 10_000
const EXIT_PROOF_TIMEOUT_MS = 8_000
const RPC_TIMEOUT_MS = 1_000
const observationNotifications = new AsyncLocalStorage<HostedProviderChild>()
/** Streams exist before a remote spawn; only the owning host may supply its PID and exit. */
export class HostedProviderChild extends ProviderChildHandle {
@@ -24,7 +30,8 @@ export class HostedProviderChild extends ProviderChildHandle {
private client: DaemonClient | null = null
private dispatched = false
private canceled = false
private observing = false
private observation: { connecting: boolean } | null = null
private observationPending = false
private removeEvents: () => void = () => {}
private removeDisconnect: () => void = () => {}
@@ -58,6 +65,7 @@ export class HostedProviderChild extends ProviderChildHandle {
}
private async signalHeldRoot(signal: NodeJS.Signals): Promise<boolean> {
const fromObservationFailure = observationNotifications.getStore() === this
if (this.direct) {
return this.direct.kill(signal)
}
@@ -80,7 +88,9 @@ export class HostedProviderChild extends ProviderChildHandle {
)
return result.delivered === true
} catch {
void this.observeLostHost()
if (!fromObservationFailure) {
void this.observeLostHost()
}
return false
}
}
@@ -110,26 +120,24 @@ export class HostedProviderChild extends ProviderChildHandle {
) {
return this.launchDirect()
}
const env = createHostedEnv(this.spec, identity.pid, {
const payload = createHostedLaunch(this.spec, identity.pid, {
path,
nonce: this.listener.nonce
})
this.removeEvents = this.client.onProviderExit(
this.listener.nonce,
(outcome) => this.observeOutcome(outcome),
() => void this.observeLostHost()
() => {
if (!this.observation?.connecting) {
void this.observeLostHost()
}
}
)
this.removeDisconnect = this.client.onDisconnected(() => void this.observeLostHost())
this.dispatched = true
const { pid } = await this.client.request<{ pid: number }>(
'spawnProvider',
{
nonce: this.listener.nonce,
source: this.spec.args[1],
args: this.spec.args.slice(3),
cwd: this.spec.cwd,
env
},
payload,
START_TIMEOUT_MS
)
if (!Number.isSafeInteger(pid) || pid <= 1) {
@@ -137,7 +145,7 @@ export class HostedProviderChild extends ProviderChildHandle {
}
this.pid = pid
this.listener.bindPid(pid)
const channels = await this.withStartDeadline(this.listener.ready)
const channels = await withProviderStartDeadline(this.listener.ready, START_TIMEOUT_MS)
if (this.canceled || this.settled) {
await this.listener.close()
await this.signalHeldRoot('SIGTERM')
@@ -163,6 +171,9 @@ export class HostedProviderChild extends ProviderChildHandle {
channels.control.write('{"go":true}\n')
channels.control.resume()
} catch (error) {
if (this.settled) {
return
}
const failure = error instanceof Error ? error : new Error('Provider could not start')
if (!this.dispatched) {
this.failBeforeDispatch(failure)
@@ -188,8 +199,9 @@ export class HostedProviderChild extends ProviderChildHandle {
this.rootAbsent = true
this.settled = true
this.rejectReady(error)
this.emit('error', error)
this.observationPending = false
void this.listener.close()
this.notifyTerminal('error', error)
this.finishClose(null, null)
}
@@ -198,6 +210,7 @@ export class HostedProviderChild extends ProviderChildHandle {
return
}
this.settled = true
this.observationPending = false
this.pid ??= outcome.pid ?? undefined
this.rootAbsent = true
this.spawnAbsenceKnown = outcome.processless
@@ -239,33 +252,23 @@ export class HostedProviderChild extends ProviderChildHandle {
}
private async observeLostHost(): Promise<void> {
if (this.observing || this.settled || !this.client) {
if (this.settled || !this.client) {
return
}
this.observing = true
if (this.observation !== null) {
this.observationPending = true
return
}
this.observationPending = false
const observation = { connecting: false }
this.observation = observation
try {
const state = await observeHostedProviderExit({
timeoutMs: EXIT_PROOF_TIMEOUT_MS,
isSettled: () => this.settled,
query: async () => {
await this.client!.ensureConnectedWithin(RPC_TIMEOUT_MS)
return this.client!.request<HostedProviderState>(
'queryProvider',
{ nonce: this.listener.nonce },
RPC_TIMEOUT_MS
)
},
probe: async () =>
this.pid
? probeAgentSessionProcessIdentity({
identity: {
hostId: 'local',
pid: this.pid,
processStartTimeMs: null,
spawnToken: this.listener.nonce
}
})
: { outcome: 'indeterminate', reason: 'Supervisor PID not known' }
query: () =>
queryHostedProviderState(this.client!, this.listener.nonce, RPC_TIMEOUT_MS, observation),
probe: () => probeHostedProviderRoot(this.pid, this.listener.nonce)
})
if (this.settled) {
return
@@ -278,9 +281,10 @@ export class HostedProviderChild extends ProviderChildHandle {
}
if (state.state === 'absent') {
this.settled = true
this.observationPending = false
this.rootAbsent = true
this.rejectReady(new Error('Provider supervisor ended before startup completed'))
this.emit('providerAbsent')
this.notifyTerminal('providerAbsent')
await this.listener.close()
this.finishClose(null, null)
return
@@ -289,29 +293,22 @@ export class HostedProviderChild extends ProviderChildHandle {
// No exit proof: adapters retain writer ownership through their existing unproven-close path.
const error = new Error('Agent connection ended before its exit could be verified')
this.rejectReady(error)
this.emit('error', error)
observationNotifications.run(this, () => this.notifyTerminal('error', error))
// Transport failure is not ChildProcess close: helpers still own any live writer.
}
} catch (error) {
if (!this.settled) {
observationNotifications.run(this, () => this.notifyTerminal('error', error))
}
} finally {
this.observing = false
this.observation = null
if (this.closed) {
this.removeEvents()
this.removeDisconnect()
}
}
}
private async withStartDeadline<T>(promise: Promise<T>): Promise<T> {
let timer: NodeJS.Timeout | undefined
try {
return await Promise.race([
promise,
new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new Error('Provider launch timed out')), START_TIMEOUT_MS)
})
])
} finally {
clearTimeout(timer)
if (this.observationPending && !this.settled) {
void this.observeLostHost()
}
}
}
}
@@ -1,3 +1,5 @@
import type { DaemonClient } from '../daemon/transport/client/daemon-client'
import { probeAgentSessionProcessIdentity } from '../runtime/agent-session/ownership/agent-session-process-identity-probe'
import type { AgentSessionOwnerProbe } from '../../shared/agent-session/lifecycle/agent-session-lease-adjudication'
import type { HostedProviderState } from '../daemon/protocol/daemon-provider-process-protocol'
import { withTimeout } from '../../shared/promise-timeout-fallback'
@@ -47,3 +49,27 @@ export async function observeHostedProviderExit(options: {
}
return { state: 'unproven' }
}
export function probeHostedProviderRoot(
pid: number | undefined,
nonce: string
): Promise<AgentSessionOwnerProbe> {
return pid
? probeAgentSessionProcessIdentity({
identity: { hostId: 'local', pid, processStartTimeMs: null, spawnToken: nonce }
})
: Promise.resolve({ outcome: 'indeterminate', reason: 'Supervisor PID not known' })
}
export async function queryHostedProviderState(
client: DaemonClient,
nonce: string,
timeoutMs: number,
observation: { connecting: boolean }
): Promise<HostedProviderState> {
observation.connecting = true
await client.ensureConnectedWithin(timeoutMs).finally(() => {
observation.connecting = false
})
return client.request<HostedProviderState>('queryProvider', { nonce }, timeoutMs)
}
@@ -1,3 +1,4 @@
import type { HostedProviderSpawn } from '../daemon/protocol/daemon-provider-process-protocol'
import type { createProviderSpawnSpec } from './provider-process-supervisor'
import type { DaemonClient } from '../daemon/transport/client/daemon-client'
import type { DaemonEndpointIdentity } from '../daemon/protocol/daemon-hello-protocol'
@@ -16,11 +17,11 @@ export async function inspectHostedIdentity(
: null
}
export function createHostedEnv(
export function createHostedLaunch(
spec: ReturnType<typeof createProviderSpawnSpec>,
ownerPid: number,
channel: { path: string; nonce: string }
): Record<string, string> {
): HostedProviderSpawn {
const env = Object.fromEntries(
Object.entries(spec.env).filter((entry): entry is [string, string] => entry[1] !== undefined)
)
@@ -31,5 +32,28 @@ export function createHostedEnv(
JSON.stringify({ ...supervisor, ownerPid })
).toString('base64')
env.ORCA_PROVIDER_OWNER_CHANNEL = JSON.stringify(channel)
return env
return {
nonce: channel.nonce,
source: spec.args[1],
args: spec.args.slice(3),
cwd: spec.cwd,
env
}
}
export async function withProviderStartDeadline<T>(
promise: Promise<T>,
timeoutMs: number
): Promise<T> {
let timer: NodeJS.Timeout | undefined
try {
return await Promise.race([
promise,
new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new Error('Provider launch timed out')), timeoutMs)
})
])
} finally {
clearTimeout(timer)
}
}
@@ -2,6 +2,8 @@ import type * as ProcessTable from '../../shared/libproc-process-table'
import type * as ProcessHost from '@orca/process-host'
import type * as OwnerChannels from './provider-owner-channels'
import type { PassThrough } from 'node:stream'
import { spawn } from 'node:child_process'
import { once } from 'node:events'
import type { ChildProcessWithoutNullStreams } from 'node:child_process'
import type { DaemonEndpointIdentity } from '../daemon/protocol/daemon-hello-protocol'
import {
@@ -88,9 +90,11 @@ afterEach(() => {
function pendingChild(
fallback: () => ChildProcessWithoutNullStreams = () => {
throw new Error('Duplicate spawn')
}
},
pid = 42
) {
let disconnected = () => {}
let reconnected = () => {}
let exit = (_outcome: HostedProviderOutcome) => {}
const spec = createProviderSpawnSpec(
{ command: '/fake/codex', args: [], cwd: '/tmp' },
@@ -105,10 +109,10 @@ function pendingChild(
launchNonce: 'original-host'
}),
ensureConnectedWithin: async () => {},
request: async (type: string) =>
type === 'spawnProvider' ? { pid: 42 } : { state: 'unknown' },
onProviderExit: (_nonce: string, listener: typeof exit) => {
request: async (type: string) => (type === 'spawnProvider' ? { pid } : { state: 'unknown' }),
onProviderExit: (_nonce: string, listener: typeof exit, reconnect?: () => void) => {
exit = listener
reconnected = reconnect ?? (() => {})
return () => {}
},
onDisconnected: (listener: () => void) => {
@@ -130,6 +134,7 @@ function pendingChild(
spec,
channels,
disconnect: () => disconnected(),
reconnect: () => reconnected(),
exit: (code = 1) => {
channels.duplex.end()
channels.stderr.end()
@@ -402,6 +407,120 @@ it.each(['generation', 'policy', 'models'] as const)(
}
)
it.skipIf(process.platform === 'win32')(
'consumes a control-close wake during the final sleep and releases the actual home lock',
async () => {
const root = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' })
const rootClosed = new Promise<void>((resolve) => root.once('close', () => resolve()))
await once(root, 'spawn')
try {
vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] })
const handle = pendingChild(undefined, root.pid!)
const close = vi.fn()
handle.child.on('close', close)
const first = runCodexProcessWithHomeLock('wake-real-root', () =>
runLocalSourceControlPlan({
plan,
cwd: '/tmp',
env: undefined,
operation: 'commit-message',
spawnAgent: () => Object.assign(handle.child, { supervised: true })
})
)
const failed = first.catch(() => ({ success: false }))
await handle.child.ready
handle.disconnect()
await vi.advanceTimersByTimeAsync(7_999)
expect(handle.child.rootAbsent).toBe(false)
const secondEntered = vi.fn()
const second = runCodexProcessWithHomeLock('wake-real-root', () => {
secondEntered()
return { result: Promise.resolve('next generation'), processClosed: Promise.resolve() }
})
await vi.advanceTimersByTimeAsync(0)
expect(secondEntered).not.toHaveBeenCalled()
handle.channels.control.destroy()
expect(root.kill('SIGTERM')).toBe(true)
await rootClosed
await vi.advanceTimersByTimeAsync(1)
await failed
expect(await second).toBe('next generation')
expect(handle.child.rootAbsent).toBe(true)
expect(close).toHaveBeenCalledExactlyOnceWith(null, null)
} finally {
vi.useRealTimers()
if (root.exitCode === null && root.signalCode === null) {
root.kill('SIGTERM')
}
await rootClosed
}
}
)
it('forwards direct stdin failure while preserving actual child exit and close', async () => {
const { EventEmitter } = await import('node:events')
const { PassThrough } = await import('node:stream')
const direct = Object.assign(new EventEmitter(), {
pid: 400,
connected: false,
stdin: new PassThrough(),
stdout: new PassThrough(),
stderr: new PassThrough(),
kill: vi.fn(() => true)
})
vi.mocked(inspectMacProcessCodeIdentity).mockResolvedValue({
identity: 'probe-failed',
executablePath: null
})
const handle = pendingChild(() => {
queueMicrotask(() => direct.emit('spawn'))
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Direct fixture supplies the streams and held-child lifecycle under test.
return direct as never
})
const inputFailed = new Promise<Error>((resolve) => handle.child.stdin.once('error', resolve))
const closed = new Promise<unknown[]>((resolve) =>
handle.child.once('close', (...args) => resolve(args))
)
await handle.child.ready
const failure = Object.assign(new Error('closed input'), { code: 'EPIPE' })
direct.stdin.destroy(failure)
expect(await inputFailed).toBe(failure)
expect(handle.child.exitCode).toBeNull()
direct.stdout.end()
direct.stderr.end()
direct.emit('exit', 7, null)
direct.emit('close', 7, null)
expect(await closed).toEqual([7, null])
})
it('does not renew loss observation from its own reconnect or error cleanup', async () => {
vi.useFakeTimers()
vi.spyOn(process, 'kill').mockReturnValue(true)
const handle = pendingChild()
const query = vi.spyOn(handle.client, 'request')
vi.spyOn(handle.client, 'ensureConnectedWithin').mockImplementation(async () =>
handle.reconnect()
)
await handle.child.ready
handle.child.on('error', () => {
void Promise.resolve().then(() => handle.child.signalRoot!('SIGTERM'))
})
query.mockImplementation(async (type) => {
if (type === 'signalProvider') {
throw new Error('Host unavailable')
}
return { state: 'unknown' }
})
handle.disconnect()
await vi.advanceTimersByTimeAsync(8_001)
const stopped = query.mock.calls.length
await vi.advanceTimersByTimeAsync(16_000)
expect(query.mock.calls.length).toBe(stopped)
expect(handle.child.rootAbsent).toBe(false)
handle.exit()
await vi.advanceTimersByTimeAsync(0)
})
it('keeps a bounded startup longer than one second hosted instead of spawning twice', async () => {
vi.useFakeTimers()
const handle = pendingChild()
@@ -41,6 +41,7 @@ export abstract class ProviderChildHandle extends EventEmitter implements Provid
this.direct = child
this.pid = child.pid
this.connected = child.connected
child.stdin.on('error', (error) => this.stdin.destroy(error))
this.stdin.pipe(child.stdin)
child.stdout.pipe(this.stdout)
child.stderr.pipe(this.stderr)
@@ -3,10 +3,10 @@
"appId": "com.stablyai.orca",
"stateSchemaVersion": 1,
"readableStateSchemaVersions": [1],
"daemonProtocolVersion": 57,
"daemonProtocolVersion": 58,
"attachableDaemonProtocolVersions": [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26,
27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50,
51, 52, 53, 54, 55, 56, 57
51, 52, 53, 54, 55, 56, 57, 58
]
}
@@ -3,10 +3,10 @@ export const LOCAL_BUILD_COMPATIBILITY_CONTRACT = {
appId: 'com.stablyai.orca',
stateSchemaVersion: 1,
readableStateSchemaVersions: [1],
daemonProtocolVersion: 57,
daemonProtocolVersion: 58,
attachableDaemonProtocolVersions: [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26,
27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50,
51, 52, 53, 54, 55, 56, 57
51, 52, 53, 54, 55, 56, 57, 58
]
} as const