From b567fdbcc8d11dcf2173779e4ca302c03465f170 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:24:40 -0800 Subject: [PATCH] Settle refused agent launches and preserve exit observation wakes --- .../macos-daemon-tcc-attribution.md | 8 +- .../claude/claude-stream-json-connection.ts | 13 +- .../claude-stream-json-connection.test.ts | 24 ++++ ...ed-session-acquisition-processless.test.ts | 133 +++++++++++------- .../claude-structured-session-acquisition.ts | 1 + .../protocol/daemon-protocol-version.ts | 5 +- .../daemon-provider-process-protocol.ts | 2 +- .../adapter/daemon-pty-preserved-host-loss.ts | 3 +- ...pty-adapter-protocol-compatibility.test.ts | 18 +++ .../server/daemon-provider-processes.test.ts | 35 +++++ .../server/daemon-provider-processes.ts | 70 ++++++--- .../daemon/server/daemon-provider-requests.ts | 14 +- .../daemon/server/daemon-request-router.ts | 5 +- .../daemon/server/daemon-server-lifecycle.ts | 4 + src/main/daemon/server/daemon-server.ts | 2 +- src/main/pi/rpc/rpc-session-startup.test.ts | 32 +++++ src/main/pi/rpc/rpc-startup-open.ts | 5 +- .../hosted-provider-child.integration.test.ts | 70 ++++++++- .../provider-process/hosted-provider-child.ts | 113 ++++++++------- .../hosted-provider-exit-observer.ts | 26 ++++ .../hosted-provider-launch.ts | 30 +++- .../hosted-provider-ownership.test.ts | 127 ++++++++++++++++- .../provider-process/provider-child-handle.ts | 1 + .../local-build-compatibility-contract.json | 4 +- .../local-build-compatibility-contract.ts | 4 +- 25 files changed, 583 insertions(+), 166 deletions(-) diff --git a/docs/agents-md-refs/macos-daemon-tcc-attribution.md b/docs/agents-md-refs/macos-daemon-tcc-attribution.md index 796d89894c6..307bfb7aeb1 100644 --- a/docs/agents-md-refs/macos-daemon-tcc-attribution.md +++ b/docs/agents-md-refs/macos-daemon-tcc-attribution.md @@ -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 diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts index f4bfde6f818..c7e7abe7c54 100644 --- a/src/main/claude/claude-stream-json-connection.ts +++ b/src/main/claude/claude-stream-json-connection.ts @@ -62,6 +62,7 @@ function withoutKeys(env: Record, keys: readonly string[]): Reco } export type ClaudeStreamJsonConnectionHandlers = { + onConnection?: (connection: ClaudeStreamJsonConnection) => void onMessage?: (message: Record) => 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 } 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 ed3bf767324..7ad9f860270 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 @@ -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() + const retained = Promise.withResolvers() + const opening = openClaudeStreamJsonConnection( + launchFor(scenario), + { + onConnection: (connection) => retained.resolve(connection) + }, + (spec) => { + const child = spawnProcess(spec) + spawnedChildren.push(child) + childClosures.push(new Promise((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 () => { diff --git a/src/main/claude/structured/claude-structured-session-acquisition-processless.test.ts b/src/main/claude/structured/claude-structured-session-acquisition-processless.test.ts index 4f0d8a92514..67ff0cb5dbb 100644 --- a/src/main/claude/structured/claude-structured-session-acquisition-processless.test.ts +++ b/src/main/claude/structured/claude-structured-session-acquisition-processless.test.ts @@ -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() - }) + ) }) diff --git a/src/main/claude/structured/claude-structured-session-acquisition.ts b/src/main/claude/structured/claude-structured-session-acquisition.ts index b8af7323531..b97be60b63c 100644 --- a/src/main/claude/structured/claude-structured-session-acquisition.ts +++ b/src/main/claude/structured/claude-structured-session-acquisition.ts @@ -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) { diff --git a/src/main/daemon/protocol/daemon-protocol-version.ts b/src/main/daemon/protocol/daemon-protocol-version.ts index 88d663a99c4..52d9b9f96b2 100644 --- a/src/main/daemon/protocol/daemon-protocol-version.ts +++ b/src/main/daemon/protocol/daemon-protocol-version.ts @@ -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 { diff --git a/src/main/daemon/protocol/daemon-provider-process-protocol.ts b/src/main/daemon/protocol/daemon-provider-process-protocol.ts index 90aa0a77c95..8912429d57e 100644 --- a/src/main/daemon/protocol/daemon-provider-process-protocol.ts +++ b/src/main/daemon/protocol/daemon-provider-process-protocol.ts @@ -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 diff --git a/src/main/daemon/providers/adapter/daemon-pty-preserved-host-loss.ts b/src/main/daemon/providers/adapter/daemon-pty-preserved-host-loss.ts index 982b3b12dee..6c9dbf93ea3 100644 --- a/src/main/daemon/providers/adapter/daemon-pty-preserved-host-loss.ts +++ b/src/main/daemon/providers/adapter/daemon-pty-preserved-host-loss.ts @@ -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 { diff --git a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-protocol-compatibility.test.ts b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-protocol-compatibility.test.ts index fdd29a43b26..fa0855502b6 100644 --- a/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-protocol-compatibility.test.ts +++ b/src/main/daemon/providers/adapter/testing/daemon-pty-adapter-protocol-compatibility.test.ts @@ -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 () => { diff --git a/src/main/daemon/server/daemon-provider-processes.test.ts b/src/main/daemon/server/daemon-provider-processes.test.ts index 722b153d49d..a64c553c1e0 100644 --- a/src/main/daemon/server/daemon-provider-processes.test.ts +++ b/src/main/daemon/server/daemon-provider-processes.test.ts @@ -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() + } +}) diff --git a/src/main/daemon/server/daemon-provider-processes.ts b/src/main/daemon/server/daemon-provider-processes.ts index f42d95d97d3..a36fb66e6ef 100644 --- a/src/main/daemon/server/daemon-provider-processes.ts +++ b/src/main/daemon/server/daemon-provider-processes.ts @@ -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) diff --git a/src/main/daemon/server/daemon-provider-requests.ts b/src/main/daemon/server/daemon-provider-requests.ts index fe334451373..edf99cb75d0 100644 --- a/src/main/daemon/server/daemon-provider-requests.ts +++ b/src/main/daemon/server/daemon-provider-requests.ts @@ -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, 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( diff --git a/src/main/daemon/server/daemon-request-router.ts b/src/main/daemon/server/daemon-request-router.ts index 0db437e3265..faf413475c6 100644 --- a/src/main/daemon/server/daemon-request-router.ts +++ b/src/main/daemon/server/daemon-request-router.ts @@ -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 { + async route(clientId: string, request: DaemonRequest, socket?: Socket): Promise { 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') diff --git a/src/main/daemon/server/daemon-server-lifecycle.ts b/src/main/daemon/server/daemon-server-lifecycle.ts index 7de86824e3b..589fa8fff4b 100644 --- a/src/main/daemon/server/daemon-server-lifecycle.ts +++ b/src/main/daemon/server/daemon-server-lifecycle.ts @@ -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 { return new Promise((resolve, reject) => { this.server = createServer(handleConnection) diff --git a/src/main/daemon/server/daemon-server.ts b/src/main/daemon/server/daemon-server.ts index 3e09da5a007..e626e4b3b01 100644 --- a/src/main/daemon/server/daemon-server.ts +++ b/src/main/daemon/server/daemon-server.ts @@ -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) ) } diff --git a/src/main/pi/rpc/rpc-session-startup.test.ts b/src/main/pi/rpc/rpc-session-startup.test.ts index f7f8b39e93c..7eac4b88fb1 100644 --- a/src/main/pi/rpc/rpc-session-startup.test.ts +++ b/src/main/pi/rpc/rpc-session-startup.test.ts @@ -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() + 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() +}) diff --git a/src/main/pi/rpc/rpc-startup-open.ts b/src/main/pi/rpc/rpc-startup-open.ts index 62a73e84c8e..eea9cb36b10 100644 --- a/src/main/pi/rpc/rpc-startup-open.ts +++ b/src/main/pi/rpc/rpc-startup-open.ts @@ -70,7 +70,8 @@ export async function waitForPiSpawn( connection: PiRpcConnection, signal: AbortSignal ): Promise { - if (connection.spawned) { - await waitForPromiseWithSignal(connection.spawned, signal) + const spawned = connection.spawned + if (spawned) { + await waitForPromiseWithSignal(spawned, signal) } } 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 be13fef601b..ebd680685ea 100644 --- a/src/main/provider-process/hosted-provider-child.integration.test.ts +++ b/src/main/provider-process/hosted-provider-child.integration.test.ts @@ -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() + 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 | undefined + vi.spyOn(client, 'request').mockImplementation(async function ( + ...args: Parameters + ): Promise { + 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(...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 { diff --git a/src/main/provider-process/hosted-provider-child.ts b/src/main/provider-process/hosted-provider-child.ts index 70012bc3feb..7500c2dc235 100644 --- a/src/main/provider-process/hosted-provider-child.ts +++ b/src/main/provider-process/hosted-provider-child.ts @@ -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 const START_TIMEOUT_MS = 10_000 const EXIT_PROOF_TIMEOUT_MS = 8_000 const RPC_TIMEOUT_MS = 1_000 +const observationNotifications = new AsyncLocalStorage() /** 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 { + 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 { - 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( - '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(promise: Promise): Promise { - let timer: NodeJS.Timeout | undefined - try { - return await Promise.race([ - promise, - new Promise((_, reject) => { - timer = setTimeout(() => reject(new Error('Provider launch timed out')), START_TIMEOUT_MS) - }) - ]) - } finally { - clearTimeout(timer) + if (this.observationPending && !this.settled) { + void this.observeLostHost() + } } } } diff --git a/src/main/provider-process/hosted-provider-exit-observer.ts b/src/main/provider-process/hosted-provider-exit-observer.ts index af3b5388348..5db77526071 100644 --- a/src/main/provider-process/hosted-provider-exit-observer.ts +++ b/src/main/provider-process/hosted-provider-exit-observer.ts @@ -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 { + 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 { + observation.connecting = true + await client.ensureConnectedWithin(timeoutMs).finally(() => { + observation.connecting = false + }) + return client.request('queryProvider', { nonce }, timeoutMs) +} diff --git a/src/main/provider-process/hosted-provider-launch.ts b/src/main/provider-process/hosted-provider-launch.ts index 2768f51d787..49cd8ec35bc 100644 --- a/src/main/provider-process/hosted-provider-launch.ts +++ b/src/main/provider-process/hosted-provider-launch.ts @@ -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, ownerPid: number, channel: { path: string; nonce: string } -): Record { +): 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( + promise: Promise, + timeoutMs: number +): Promise { + let timer: NodeJS.Timeout | undefined + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('Provider launch timed out')), timeoutMs) + }) + ]) + } finally { + clearTimeout(timer) + } } diff --git a/src/main/provider-process/hosted-provider-ownership.test.ts b/src/main/provider-process/hosted-provider-ownership.test.ts index 84b94a31e3e..afcb3d046f1 100644 --- a/src/main/provider-process/hosted-provider-ownership.test.ts +++ b/src/main/provider-process/hosted-provider-ownership.test.ts @@ -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((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((resolve) => handle.child.stdin.once('error', resolve)) + const closed = new Promise((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() diff --git a/src/main/provider-process/provider-child-handle.ts b/src/main/provider-process/provider-child-handle.ts index 3c218e5c0ef..97db3d17b09 100644 --- a/src/main/provider-process/provider-child-handle.ts +++ b/src/main/provider-process/provider-child-handle.ts @@ -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) diff --git a/src/shared/local-build-compatibility-contract.json b/src/shared/local-build-compatibility-contract.json index 0fccd4f93ef..89f3e7ca478 100644 --- a/src/shared/local-build-compatibility-contract.json +++ b/src/shared/local-build-compatibility-contract.json @@ -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 ] } diff --git a/src/shared/local-build-compatibility-contract.ts b/src/shared/local-build-compatibility-contract.ts index 10c7d4110c9..aa9d0e6f71e 100644 --- a/src/shared/local-build-compatibility-contract.ts +++ b/src/shared/local-build-compatibility-contract.ts @@ -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