From 010a12e247466b27213b679749044b847678fef6 Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 3 Aug 2026 12:55:34 +0800 Subject: [PATCH] fix(runtime): close lifecycle races exposed by CI --- .../src/__tests__/execution-host.test.ts | 50 ++++++++++++------- .../src/__tests__/host-kernel.test.ts | 10 +++- .../runtime-host/src/server/host-kernel.ts | 6 ++- .../src/__tests__/shell-run-manager.test.ts | 24 +++++++-- packages/runtime/src/shell-run-manager.ts | 19 +++++-- 5 files changed, 81 insertions(+), 28 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-host.test.ts b/packages/runtime-host/src/__tests__/execution-host.test.ts index 56bf31a8d40..a88e29b8946 100644 --- a/packages/runtime-host/src/__tests__/execution-host.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host.test.ts @@ -480,19 +480,25 @@ test('two Clients share one execution after the starting Client disconnects', as const secondProbe = new SubscriptionProbe(secondSubscription); const turnId = randomUUID(); - const started = await first.startTurn({ - sessionId: fixture.sessionId, - turnId, - content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, - }); + const started = await first.startTurn( + { + sessionId: fixture.sessionId, + turnId, + content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, + }, + PROCESS_TIMEOUT_MS, + ); assert.equal(started.turnId, turnId); await assert.rejects( () => - second.startTurn({ - sessionId: fixture.sessionId, - turnId: randomUUID(), - content: { text: 'must stay busy' }, - }), + second.startTurn( + { + sessionId: fixture.sessionId, + turnId: randomUUID(), + content: { text: 'must stay busy' }, + }, + PROCESS_TIMEOUT_MS, + ), operationError('session_busy'), ); @@ -548,17 +554,23 @@ test('two Clients share one execution after the starting Client disconnects', as ); const nextTurnId = randomUUID(); - const next = await second.startTurn({ - sessionId: fixture.sessionId, - turnId: nextTurnId, - content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, - }); - assert.deepEqual( - await second.startTurn({ + const next = await second.startTurn( + { sessionId: fixture.sessionId, - turnId, + turnId: nextTurnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, - }), + }, + PROCESS_TIMEOUT_MS, + ); + assert.deepEqual( + await second.startTurn( + { + sessionId: fixture.sessionId, + turnId, + content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, + }, + PROCESS_TIMEOUT_MS, + ), stopped, ); assert.deepEqual( diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index c93fb3af9a1..5e2cf6aa0a9 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -227,7 +227,7 @@ describe('non-serving Runtime Host kernel', () => { }); }); - test('composition process-exit retention requires termination without releasing ownership', async () => { + test('process-exit retention closes admission before requiring termination without releasing ownership', async () => { await withHostPaths(async (paths) => { const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' }); const owner = await tryAcquireInteractiveRootOwner(capability); @@ -251,6 +251,14 @@ describe('non-serving Runtime Host kernel', () => { error instanceof RuntimeHostProcessTerminationRequiredError && error.code === 'process_termination_required', ); + await assert.rejects( + () => openSocket(host.endpoint), + (error: unknown) => + error instanceof Error && + 'code' in error && + ((error as NodeJS.ErrnoException).code === 'ENOENT' || + (error as NodeJS.ErrnoException).code === 'ECONNREFUSED'), + ); assert.equal(await tryAcquireInteractiveRootOwner(capability), undefined); } finally { await owner.close(); diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index db074504a60..903e58137a4 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -558,9 +558,13 @@ export class RuntimeHostKernel { async #closeResources(): Promise { const errors: unknown[] = []; + // Stop new admissions before any asynchronous shutdown bookkeeping. The + // shutdown deadline may expire while publishing the draining registration; + // leaving the listener open in that case strands an unreachable, ref'ed + // server until the process is forcibly terminated. + const serverClosed = closeServer(this.#server).catch((error: unknown) => errors.push(error)); await this.#publishRegistration().catch((error: unknown) => errors.push(error)); this.#assertShutdownCanContinue(); - const serverClosed = closeServer(this.#server).catch((error: unknown) => errors.push(error)); const accepted = [...this.#acceptedTransports]; const handshaking = [...this.#handshakingTransports]; const operationDrain = this.#waitForOperations(); diff --git a/packages/runtime/src/__tests__/shell-run-manager.test.ts b/packages/runtime/src/__tests__/shell-run-manager.test.ts index c12963587a7..99a1e304db1 100644 --- a/packages/runtime/src/__tests__/shell-run-manager.test.ts +++ b/packages/runtime/src/__tests__/shell-run-manager.test.ts @@ -527,27 +527,45 @@ describe('ShellRunProcessManager', () => { }, async (context) => { const processDiscovery = delayPosixProcessDiscovery(context); const abort = new AbortController(); - const manager = await createTestManager(); + const backingStore = createLegacyShellRunStoreForTest(await workspace()); + const runningCommitted = deferred(); + const releaseRunning = deferred(); + const store: ShellRunStore = { + createShellRun: (...args) => backingStore.createShellRun(...args), + async updateShellRun(sessionId, shellRunId, patch) { + const updated = await backingStore.updateShellRun(sessionId, shellRunId, patch); + if (patch.status === 'running') { + runningCommitted.resolve(undefined); + await releaseRunning.promise; + } + return updated; + }, + readShellRun: (...args) => backingStore.readShellRun(...args), + listSessionShellRuns: (...args) => backingStore.listSessionShellRuns(...args), + }; + const manager = createManager(store); const run = manager.runForegroundBash( shellInput({ cwd: await workspace(), command: waitForeverCommand('ready'), timeoutMs: 50, abortSignal: abort.signal, - emitOutput: () => abort.abort(), }), ); try { + await runningCommitted.promise; await processDiscovery.started; - await new Promise((resolve) => setTimeout(resolve, 100)); + abort.abort(); processDiscovery.release(); + releaseRunning.resolve(undefined); const result = await run; assert.equal(result.status, 'cancelled'); assert.equal(result.exitCode, 130); } finally { processDiscovery.release(); + releaseRunning.resolve(undefined); await run.catch(() => undefined); await manager.terminateAll().catch(() => undefined); } diff --git a/packages/runtime/src/shell-run-manager.ts b/packages/runtime/src/shell-run-manager.ts index 637e537476f..8b3ecd4b6d2 100644 --- a/packages/runtime/src/shell-run-manager.ts +++ b/packages/runtime/src/shell-run-manager.ts @@ -251,6 +251,11 @@ export class ShellRunProcessManager async runForegroundBash(input: ShellRunBashInput): Promise { const onCompletion = onceShellRunCompletion(input.onCompletion); const ownedInput = onCompletion ? { ...input, onCompletion } : input; + let live: LiveShellRun | undefined; + const cancel = () => { + if (live) this.requestForcedTermination(live, 'cancel'); + }; + input.abortSignal?.addEventListener('abort', cancel, { once: true }); try { validateSourceToolCallId(input.sourceToolCallId); return await this.withPendingStartup(input.sessionId, async () => { @@ -259,15 +264,17 @@ export class ShellRunProcessManager if (input.abortSignal?.aborted) throw abortError('Command aborted before shell process started'); const timeoutMs = normalizeForegroundTimeoutMs(input.timeoutMs ?? DEFAULT_BASH_TIMEOUT_MS); - const live = await this.start(ownedInput, 'pipes', timeoutMs, true); - if ((await live.finished.waitFor(input.abortSignal)) === 'abort') { - this.requestForcedTermination(live, 'cancel'); - } + live = await this.start(ownedInput, 'pipes', timeoutMs, true, (admitted) => { + live = admitted; + if (input.abortSignal?.aborted) cancel(); + }); return this.markObservedAndReturnTerminal(await live.finished.join()); }); } catch (error) { notifyFailedStartup(onCompletion); throw error; + } finally { + input.abortSignal?.removeEventListener('abort', cancel); } } @@ -570,6 +577,7 @@ export class ShellRunProcessManager mode: ShellMode, timeoutMs: number | undefined, forwardLive: boolean, + onLiveAdmission?: (live: LiveShellRun) => void, ): Promise { const sessionEpoch = this.sessionTerminationEpoch(input.sessionId); this.assertStartAllowed(input.sessionId, sessionEpoch); @@ -589,6 +597,7 @@ export class ShellRunProcessManager forwardLive, slotReservation, sessionEpoch, + onLiveAdmission, ); } @@ -620,6 +629,7 @@ export class ShellRunProcessManager forwardLive: boolean, slotReservation: ShellRunSlotReservation, sessionEpoch: number, + onLiveAdmission: ((live: LiveShellRun) => void) | undefined, ): Promise { const collector = new PipeTailCollector(this.maxRetainedChars); const pending: Array<(live: LivePipeShellRun) => void> = []; @@ -673,6 +683,7 @@ export class ShellRunProcessManager for (const callback of pending) callback(live); driver.writeInputs(); await racePromiseWithAbort(driver.ready, input.abortSignal); + onLiveAdmission?.(live); this.armTimeout(live); this.assertLiveStartupAllowed(live, sessionEpoch, input.abortSignal); await this.markRunning(live);