From a0bc2ea803c80976257df7498e62f6ce35db16c2 Mon Sep 17 00:00:00 2001 From: pku-xht Date: Fri, 28 Aug 2026 22:34:09 +0800 Subject: [PATCH] fix(subprocess): preserve native runner launch semantics --- .../subprocess-local/src/spawn-runner.ts | 21 +++- .../subprocess/subprocess-local/src/spawn.ts | 21 ++-- .../tests/spawn-runner.spec.ts | 112 +++++++++++++++--- .../subprocess-local/tests/spawn.spec.ts | 35 ++++++ 4 files changed, 160 insertions(+), 29 deletions(-) diff --git a/packages/subprocess/subprocess-local/src/spawn-runner.ts b/packages/subprocess/subprocess-local/src/spawn-runner.ts index 096c783371..c33ba5570a 100644 --- a/packages/subprocess/subprocess-local/src/spawn-runner.ts +++ b/packages/subprocess/subprocess-local/src/spawn-runner.ts @@ -100,19 +100,33 @@ function linuxPathNotFoundError(program: string): NodeJS.ErrnoException { }) } +function execLinuxFile( + file: string, + argv: string[], + env: Record, + internals: SpawnRunnerInternals, +): never { + try { + return internals.execve(file, argv, env) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOEXEC') throw error + return internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env) + } +} + function execLinuxTarget( request: { cwd: string; env: Record }, argv: string[], internals: SpawnRunnerInternals, ): never { const program = argv[0] as string - if (program.includes('/')) return internals.execve(program, argv, request.env) + if (program.includes('/')) return execLinuxFile(program, argv, request.env, internals) const path = request.env.PATH ?? '/usr/bin:/bin' let permissionFailure: Error | undefined for (const directory of path.split(':')) { const candidate = posix.resolve(request.cwd, directory, program) try { - return internals.execve(candidate, argv, request.env) + return execLinuxFile(candidate, argv, request.env, internals) } catch (error) { const code = (error as NodeJS.ErrnoException).code if (code === 'EACCES') { @@ -143,7 +157,7 @@ function runLinux( } try { host.chdir(request.cwd) - execLinuxTarget(request, argv, internals) + execLinuxTarget({ ...request, cwd: host.cwd() }, argv, internals) } catch (error) { writeLinuxStartupError(files, { type: 'spawn-error', @@ -263,7 +277,6 @@ class WindowsJobRunner { this.internals.closeCurrentProcessStandardStreams(this.api) if (this.startCancellationPending()) this.terminateOwnedJob() this.pollTimer = setInterval(() => { this.poll() }, 10) - this.poll() } catch (error) { if (!this.committed && error instanceof Win32Error && error.api === 'CreateProcessW') { await this.publishTerminalResult({ diff --git a/packages/subprocess/subprocess-local/src/spawn.ts b/packages/subprocess/subprocess-local/src/spawn.ts index deaf77d2e6..5a3772c3a8 100644 --- a/packages/subprocess/subprocess-local/src/spawn.ts +++ b/packages/subprocess/subprocess-local/src/spawn.ts @@ -485,10 +485,19 @@ export function bindManagedProcess( let rangeExitObservation: Promise | undefined let settled = false + const scheduleOwnerCleanup = (): boolean => { + if (launch.owner.cleanup === undefined) return false + queueMicrotask(() => { void done.finally(() => { launch.owner.cleanup?.() }).catch(() => {}) }) + return true + } + /** - * Start or reuse the handle's managed-range exit observer. A failed read can - * be retried; the first confirmed absence is the permanent no-more-signals - * boundary and cancels pending escalation before stale identity can be used. + * Start or reuse the handle's managed-range exit observer. A failed read + * before direct settlement can be retried. Once direct settlement permits + * cleanup, retain a failed observation because removing its private evidence + * must not turn a later wait into a false success. The first confirmed + * absence is the permanent no-more-signals boundary and cancels pending + * escalation before stale identity can be used. */ const observeRangeExit = (): Promise => { rangeExitObservation ??= (async () => { @@ -497,11 +506,9 @@ export function bindManagedProcess( if (graceTimer !== undefined) clearTimeout(graceTimer) graceTimer = undefined spec.signal?.removeEventListener('abort', onAbort) - if (launch.owner.cleanup !== undefined) { - queueMicrotask(() => { void done.finally(() => { launch.owner.cleanup?.() }).catch(() => {}) }) - } + scheduleOwnerCleanup() })().catch((error: unknown) => { - rangeExitObservation = undefined + if (!settled || !scheduleOwnerCleanup()) rangeExitObservation = undefined throw error }) return rangeExitObservation diff --git a/packages/subprocess/subprocess-local/tests/spawn-runner.spec.ts b/packages/subprocess/subprocess-local/tests/spawn-runner.spec.ts index 187682dea1..0109d0e0be 100644 --- a/packages/subprocess/subprocess-local/tests/spawn-runner.spec.ts +++ b/packages/subprocess/subprocess-local/tests/spawn-runner.spec.ts @@ -11,7 +11,7 @@ import { writeFileSync, } from 'node:fs' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { join, posix } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' import { Win32Error } from '@deepseek-ai/dsh-win32-process' import type { NativePtr, Win32ProcessBindings } from '@deepseek-ai/dsh-win32-process' @@ -69,7 +69,7 @@ class FakeRunnerHost extends EventEmitter { sendThrown: unknown cwd(): string { return this.directory } - chdir(path: string): void { this.directory = path } + chdir(path: string): void { this.directory = posix.resolve(this.directory, path) } disconnect(): void { if (!this.connected) return this.connected = false @@ -405,6 +405,39 @@ describe('Linux one-shot exec bootstrap', () => { }) }) + it('resolves relative PATH entries from the cwd after chdir', async () => { + const files = track(createLinuxLaunchFiles({ cwd: 'work', env: { PATH: 'bin:' } })) + const host = new FakeRunnerHost() + host.directory = '/base' + const execve = vi.fn(() => { throw Object.assign(new Error('not found'), { code: 'ENOENT' }) }) + await runSpawnRunner(files.requestPath, ['--', 'tool'], hostArgument(host), internals({ execve })) + expect(host.directory).toBe('/base/work') + expect(execve.mock.calls.map(call => call[0])).toEqual([ + '/base/work/bin/tool', + '/base/work/tool', + ]) + }) + + it('retries ENOEXEC through /bin/sh with the resolved file and original arguments', async () => { + const files = track(createLinuxLaunchFiles({ cwd: '/work', env: { PATH: 'bin' } })) + const execve = vi.fn() + .mockImplementationOnce(() => { throw Object.assign(new Error('exec format'), { code: 'ENOEXEC' }) }) + .mockImplementationOnce(() => { throw Object.assign(new Error('shell failed'), { code: 'EIO' }) }) + await runSpawnRunner( + files.requestPath, + ['--', 'tool', 'literal arg'], + hostArgument(new FakeRunnerHost()), + internals({ execve: execve as never }), + ) + expect(execve.mock.calls).toEqual([ + ['/work/bin/tool', ['tool', 'literal arg'], { PATH: 'bin' }], + ['/bin/sh', ['/bin/sh', '/work/bin/tool', 'literal arg'], { PATH: 'bin' }], + ]) + expect(readLinuxStartupError(files.startupErrorPath)).toMatchObject({ + type: 'spawn-error', error: { code: 'EIO', path: 'tool' }, + }) + }) + it('uses the default PATH and stops on a non-search error', async () => { const files = track(createLinuxLaunchFiles({ cwd: '/work', env: {} })) const execve = vi.fn((_file: string) => { throw Object.assign(new Error('denied'), { code: 'EACCES' }) }) @@ -513,6 +546,39 @@ describe('Windows Job runner protocol owner', () => { expect(host.env).toEqual({ TARGET: 'yes', dsh_subprocess_runner: 'restored' }) }) + it('lets asynchronous runner stdio close before the first Windows poll', async () => { + let tick: (() => void) | undefined + const interval = vi.spyOn(globalThis, 'setInterval').mockImplementation((callback: () => void) => { + tick = callback + return 1 as unknown as ReturnType + }) + try { + const events: string[] = [] + let closeComplete = false + const host = new FakeRunnerHost() + const native = internals({ + closeCurrentProcessStandardStreams: vi.fn(() => { + events.push('close-start') + queueMicrotask(() => { + closeComplete = true + events.push('close-complete') + tick?.() + }) + }), + pollProcessExit: vi.fn(() => { + events.push(`poll:${String(closeComplete)}`) + return 0 + }), + }) + await runWindows(host, native) + expect(events).toEqual(['close-start', 'close-complete', 'poll:true']) + expect(host.sent).toEqual([{ type: 'target-exit', exitCode: 0, signal: null }]) + expect(host.exitCode).toBe(0) + } finally { + interval.mockRestore() + } + }) + it('exhausts spawn-error, runner-error, and payload-free start-cancelled', async () => { const spawnHost = new FakeRunnerHost() await runWindows(spawnHost, internals({ @@ -691,22 +757,32 @@ describe('Windows Job runner protocol owner', () => { }) it('cleans a direct handle after the Job identity was already cleared', async () => { - const host = new FakeRunnerHost() - const native = internals({ pollProcessExit: vi.fn(() => undefined), isJobEmpty: vi.fn(() => true) }) - const running = runSpawnRunner( - WINDOWS_RUNNER_SELECTION, - ['--', 'tool.exe'], - hostArgument(host), - native, - ) - host.emit('message', { type: 'start', cwd: 'C:\\target', env: {} }) - await new Promise((resolveImmediate) => { setImmediate(resolveImmediate) }) - host.emit('message', { type: 'terminate' }) - host.disconnect() - await running - expect(native.closeHandleChecked).toHaveBeenCalledWith( - expect.anything(), 10n, 'ordinary direct process cleanup', - ) + let tick: (() => void) | undefined + const interval = vi.spyOn(globalThis, 'setInterval').mockImplementation((callback: () => void) => { + tick = callback + return 1 as unknown as ReturnType + }) + try { + const host = new FakeRunnerHost() + const native = internals({ pollProcessExit: vi.fn(() => undefined), isJobEmpty: vi.fn(() => true) }) + const running = runSpawnRunner( + WINDOWS_RUNNER_SELECTION, + ['--', 'tool.exe'], + hostArgument(host), + native, + ) + host.emit('message', { type: 'start', cwd: 'C:\\target', env: {} }) + await new Promise((resolveImmediate) => { setImmediate(resolveImmediate) }) + tick?.() + host.emit('message', { type: 'terminate' }) + host.disconnect() + await running + expect(native.closeHandleChecked).toHaveBeenCalledWith( + expect.anything(), 10n, 'ordinary direct process cleanup', + ) + } finally { + interval.mockRestore() + } }) it('fails closed for malformed or duplicate start messages and disconnected reporting', async () => { diff --git a/packages/subprocess/subprocess-local/tests/spawn.spec.ts b/packages/subprocess/subprocess-local/tests/spawn.spec.ts index e839201d65..9d78af0b4b 100644 --- a/packages/subprocess/subprocess-local/tests/spawn.spec.ts +++ b/packages/subprocess/subprocess-local/tests/spawn.spec.ts @@ -814,6 +814,41 @@ describe('coverage seams', () => { await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) }) + it('retries an early range read but cleans and retains a terminal range failure', async () => { + const direct = Promise.withResolvers<{ exitCode: number; signal: null }>() + const earlyFailure = new Error('temporary range read failure') + const terminalFailure = new Error('scope ended before launch request consumption') + const waitForExit = vi.fn() + .mockRejectedValueOnce(earlyFailure) + .mockRejectedValueOnce(terminalFailure) + const cleanup = vi.fn() + const handle = bindManagedProcess(spec('true', { + stdio: { stdin: 'ignore', stdout: 'inherit', stderr: 'inherit' }, + }), { + stdin: null, + stdout: null, + stderr: null, + direct: direct.promise, + owner: { + signal: vi.fn(), + waitForExit, + terminateForHostExit: vi.fn(), + cleanup, + }, + }) + + await expect(handle.waitForExit()).rejects.toBe(earlyFailure) + expect(cleanup).not.toHaveBeenCalled() + + direct.resolve({ exitCode: 0, signal: null }) + await expect(handle.done).resolves.toEqual({ exitCode: 0, signal: null }) + await expect(handle.waitForExit()).rejects.toBe(terminalFailure) + await vi.waitFor(() => { expect(cleanup).toHaveBeenCalledOnce() }) + + await expect(handle.waitForExit()).rejects.toBe(terminalFailure) + expect(waitForExit).toHaveBeenCalledTimes(2) + }) + it('does not deliver a stale escalation after range exit wins the timer race', async () => { vi.useFakeTimers() const clearTimer = vi.spyOn(globalThis, 'clearTimeout').mockImplementation(() => {})