Files
deepseek-harness/packages/subprocess/subprocess-local/src/spawn-runner.ts
T
sjw1231 2bd9913e2f refactor(apply-patch): drop unrelated changes and redundant prose
Restore files outside the patch feature to their merge-base state: the
invariant-check refactors across acp, session-controller, client,
compaction, agent-loop, agent, session-title, and subprocess, the cast
baseline edit, and the rescope-vendor skip list added during the merge.

Rename tool-fs.enableApplyPatch to applyPatch, matching the read, write,
and edit switches. Remove the unused collectMutationPaths export and the
unconsumed PatchHunk.contextLineIndices field, including its SSH schema
and fixtures. Report preflight refusals and syntax failures directly
instead of padding every phase with outcome, affectedPaths, and
cancelled fields. Add apply_patch to the shipped tool catalog test.

Fold the default-activation Agent Note into the apply_patch file-editing
note and delete it. Delete the guarded-filesystem-mutations upgrade
guide: every change is additive and all FileSystem subclasses live in
this repository. Trim defensive and reviewer-addressed sentences from
the package READMEs, subsystem docs, FileSystem JSDoc, and the Agent
Note, keeping the English and Chinese pairs aligned.
2026-10-20 20:25:24 +08:00

491 lines
16 KiB
TypeScript

/** One-shot Linux exec bootstrap and Windows Job-owning subprocess runner. */
import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
import { closeSync } from 'node:fs'
import {
closeHandleChecked,
isJobEmpty,
loadWin32ProcessBindings,
pollProcessExit,
spawnCurrentTokenJobProcess,
terminateJob,
Win32Error,
} from '@deepseek-ai/dsh-win32-process'
import type {
CurrentTokenProcessBindings,
NativePtr,
} from '@deepseek-ai/dsh-win32-process'
import { loadLinuxExecve } from './linux-execve.ts'
import {
consumeLinuxLaunchRequest,
isWindowsTerminateRequest,
linuxLaunchFilesFromLocator,
parseWindowsStartRequest,
serializeRunnerError,
writeLinuxStartupError,
} from './runner-protocol.ts'
import type {
LinuxLaunchFiles,
SerializedRunnerError,
WindowsRunnerResult,
WindowsStartRequest,
} from './runner-protocol.ts'
import {
parseRunnerTargetArgv,
resolveWindowsExecutable,
SUBPROCESS_RUNNER_ENV,
WINDOWS_RUNNER_SELECTION,
} from './runner-launch.ts'
type RunnerHost = Pick<NodeJS.Process, 'env' | 'exitCode' | 'connected' | 'cwd' | 'chdir' | 'on' | 'off' | 'once' | 'disconnect'> & {
send?: NodeJS.Process['send']
}
/** Injectable operations used by the protocol-owner tests. */
export interface SpawnRunnerInternals {
execve(file: string, argv: string[], env: Record<string, string>, control?: 'pipe'): never
loadWin32ProcessBindings(): CurrentTokenProcessBindings
spawnCurrentTokenJobProcess: typeof spawnCurrentTokenJobProcess
closeFileDescriptor(fileDescriptor: number): void
resolveWindowsExecutable: typeof resolveWindowsExecutable
pollProcessExit: typeof pollProcessExit
isJobEmpty: typeof isJobEmpty
terminateJob: typeof terminateJob
closeHandleChecked: typeof closeHandleChecked
}
const defaultInternals: SpawnRunnerInternals = {
/* v8 ignore next -- source/built/packaged subprocess smoke executes this only in a replaceable child process. */
execve: (file, argv, env, control) => loadLinuxExecve()(file, argv, env, control),
loadWin32ProcessBindings,
spawnCurrentTokenJobProcess,
closeFileDescriptor: closeSync,
resolveWindowsExecutable,
pollProcessExit,
isJobEmpty,
terminateJob,
closeHandleChecked,
}
const NODE_SPAWN_DETAIL_CODES = new Set(['EACCES', 'ENOENT'])
const WINDOWS_SPAWN_ERROR_CODES = new Map<number, string>([
[2, 'ENOENT'],
[3, 'ENOENT'],
[267, 'ENOENT'],
[5, 'EPERM'],
[193, 'EFTYPE'],
[740, 'EACCES'],
])
function nodeSpawnError(
syscall: string,
code: string,
path?: string,
): SerializedRunnerError {
const message = `${syscall} ${code}`
return {
name: 'Error',
message,
code,
syscall,
...path === undefined ? {} : { path },
}
}
function asSpawnError(
error: unknown,
program: string,
): SerializedRunnerError {
const serialized = serializeRunnerError(error)
if (!(error instanceof Win32Error)) {
return serialized.code === undefined
? serialized
: nodeSpawnError(`spawn ${program}`, serialized.code, program)
}
const code = WINDOWS_SPAWN_ERROR_CODES.get(error.win32Code) ?? 'UNKNOWN'
if (NODE_SPAWN_DETAIL_CODES.has(code)) {
return nodeSpawnError(`spawn ${program}`, code, program)
}
return nodeSpawnError('spawn', code)
}
function windowsPathNotFoundError(program: string): SerializedRunnerError {
return nodeSpawnError(`spawn ${program}`, 'ENOENT', program)
}
function windowsStartCancelledError(): SerializedRunnerError {
return {
name: 'Error',
message: 'subprocess target start was cancelled',
}
}
function linuxPathNotFoundError(program: string): NodeJS.ErrnoException {
return Object.assign(new Error(`spawn ${program} ENOENT`), {
code: 'ENOENT',
errno: -2,
syscall: `spawn ${program}`,
path: program,
spawnargs: [] as string[],
})
}
function execLinuxFile(
file: string,
argv: string[],
env: Record<string, string>,
internals: SpawnRunnerInternals,
control?: 'pipe',
): never {
try {
return control === undefined ? internals.execve(file, argv, env) : internals.execve(file, argv, env, control)
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOEXEC') throw error
return control === undefined
? internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env)
: internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env, control)
}
}
function execLinuxTarget(
request: { cwd: string; env: Record<string, string>; control?: 'pipe' },
argv: string[],
internals: SpawnRunnerInternals,
): never {
const program = argv[0] as string
if (program.includes('/')) return execLinuxFile(program, argv, request.env, internals, request.control)
const path = request.env.PATH ?? '/usr/bin:/bin'
let permissionFailure: Error | undefined
for (const directory of path.split(':')) {
const root = directory.startsWith('/')
? directory
: `${request.cwd}${request.cwd.endsWith('/') ? '' : '/'}${directory}`
const candidate = `${root}${root.endsWith('/') ? '' : '/'}${program}`
try {
return execLinuxFile(candidate, argv, request.env, internals, request.control)
} catch (error) {
const code = (error as NodeJS.ErrnoException).code
if (code === 'EACCES') {
permissionFailure ??= error as Error
continue
}
if (code === 'ENOENT' || code === 'ENOTDIR') continue
throw error
}
}
throw permissionFailure ?? linuxPathNotFoundError(program)
}
function runLinux(
locator: string,
argv: string[],
host: RunnerHost,
internals: SpawnRunnerInternals,
): void {
const files = linuxLaunchFilesFromLocator(locator)
let request: ReturnType<typeof consumeLinuxLaunchRequest>
try {
request = consumeLinuxLaunchRequest(files.requestPath)
} catch (error) {
writeLinuxStartupError(files, { type: 'error', error: serializeRunnerError(error) })
host.exitCode = 127
return
}
try {
host.chdir(request.cwd)
execLinuxTarget({ ...request, cwd: host.cwd() }, argv, internals)
} catch (error) {
writeLinuxStartupError(files, {
type: 'error',
error: asSpawnError(error, argv[0] as string),
})
host.exitCode = 127
}
}
function sendMessage(host: RunnerHost, result: WindowsRunnerResult): Promise<void> {
return new Promise((resolve, reject) => {
if (!host.connected || host.send === undefined) {
reject(new Error('subprocess runner IPC is not connected'))
return
}
try {
host.send(result, (error) => {
if (error === null) resolve()
else reject(error)
})
} catch (error) {
/* v8 ignore next -- process.send throws Error instances. */
const failure = error instanceof Error ? error : new Error(String(error))
reject(failure)
}
})
}
class WindowsJobRunner {
private api: CurrentTokenProcessBindings | undefined
private processHandle: NativePtr | undefined
private jobHandle: NativePtr | undefined
private pollTimer: ReturnType<typeof setInterval> | undefined
private startSeen = false
private terminateRequested = false
private resultStarted = false
private resultDelivered = false
private finished = false
private readonly completion = Promise.withResolvers<void>()
constructor(
private readonly argv: string[],
private readonly host: RunnerHost,
private readonly internals: SpawnRunnerInternals,
) {}
run(): Promise<void> {
if (!this.host.connected || this.host.send === undefined) {
this.finish(127)
return this.completion.promise
}
this.host.on('message', this.onMessage)
this.host.once('disconnect', this.onDisconnect)
return this.completion.promise
}
private readonly onMessage = (value: unknown): void => {
if (this.finished) return
if (isWindowsTerminateRequest(value)) {
this.requestTermination()
return
}
if (this.startSeen) {
void this.runnerFailure(new Error('subprocess runner received more than one Windows start request'))
return
}
let request: WindowsStartRequest
try {
request = parseWindowsStartRequest(value)
} catch (error) {
void this.runnerFailure(error)
return
}
this.startSeen = true
void this.start(request)
}
private readonly onDisconnect = (): void => {
if (this.finished) return
this.releaseOwnedJob()
this.finish(127, false)
}
private async start(request: WindowsStartRequest): Promise<void> {
if (this.terminateRequested) {
await this.publishTerminalResult({ type: 'error', error: windowsStartCancelledError() }, 0)
return
}
await new Promise<void>((resolveImmediate) => { setImmediate(resolveImmediate) })
if (this.finished) return
// IPC may set this field while start() is suspended above.
// oxlint-disable-next-line typescript/no-unnecessary-condition
if (this.terminateRequested) {
await this.publishTerminalResult({ type: 'error', error: windowsStartCancelledError() }, 0)
return
}
try {
const [command, ...args] = this.argv
const applicationName = this.internals.resolveWindowsExecutable(
command as string,
request.cwd,
request.env,
undefined,
{ ...this.host.env },
)
if (applicationName === undefined) {
await this.publishTerminalResult({
type: 'error',
error: windowsPathNotFoundError(command as string),
}, 0)
return
}
this.api = this.internals.loadWin32ProcessBindings()
const spawned = this.internals.spawnCurrentTokenJobProcess(this.api, {
command: command as string,
applicationName,
args,
cwd: request.cwd,
env: request.env,
stdio: { stdin: 4, stdout: 5, stderr: 6, ...request.control === 'pipe' ? { control: SUBPROCESS_CONTROL_FD } : {} },
})
this.processHandle = spawned.process
this.jobHandle = spawned.job
for (const fileDescriptor of request.control === 'pipe' ? [4, 5, 6, SUBPROCESS_CONTROL_FD] : [4, 5, 6]) {
this.internals.closeFileDescriptor(fileDescriptor)
}
this.pollTimer = setInterval(() => { this.poll() }, 10)
} catch (error) {
if (this.jobHandle === undefined && error instanceof Win32Error && error.api === 'CreateProcessW') {
await this.publishTerminalResult({
type: 'error',
error: asSpawnError(error, this.argv[0] as string),
}, 0)
return
}
await this.runnerFailure(error)
}
}
private requestTermination(): void {
if (this.terminateRequested) return
this.terminateRequested = true
try {
this.terminateOwnedJob()
} catch (error) {
void this.runnerFailure(error)
}
}
private terminateOwnedJob(): void {
const job = this.jobHandle
if (job === undefined) return
/* v8 ignore next -- a Job handle is assigned only after the bindings are loaded;
* the guard above is the only reachable empty-owner state. */
if (this.api === undefined) return
this.internals.terminateJob(this.api, job, 1)
}
private poll(): void {
if (this.finished) return
/* v8 ignore next -- poll is installed only after start() stores the bindings; retained as a defensive invariant guard. */
if (this.api === undefined) return
try {
if (this.processHandle !== undefined) {
const exitCode = this.internals.pollProcessExit(this.api, this.processHandle)
if (exitCode !== undefined) {
this.internals.closeHandleChecked(this.api, this.processHandle, 'ordinary direct process')
this.processHandle = undefined
void this.publishTerminalResult({ type: 'target-exit', exitCode })
}
}
if (this.jobHandle !== undefined && this.internals.isJobEmpty(this.api, this.jobHandle)) {
this.internals.closeHandleChecked(this.api, this.jobHandle, 'ordinary process Job')
this.jobHandle = undefined
if (this.resultDelivered) this.finish(0)
}
} catch (error) {
void this.runnerFailure(error)
}
}
private async publishTerminalResult(result: WindowsRunnerResult, exitCode?: number): Promise<void> {
/* v8 ignore next -- each state transition has a single result call site; the guard contains only re-entrant internal defects. */
if (this.finished || this.resultStarted) return
this.resultStarted = true
try {
await sendMessage(this.host, result)
this.resultDelivered = true
} catch {
this.releaseOwnedJob()
this.finish(127, false)
return
}
if (exitCode !== undefined) {
this.finish(exitCode)
return
}
if (this.jobHandle === undefined) this.finish(0)
}
private async runnerFailure(error: unknown): Promise<void> {
/* v8 ignore next -- callers stop/detach on finish; this guard contains only an already-queued internal callback. */
if (this.finished) return
if (!this.resultStarted) {
this.resultStarted = true
try {
await sendMessage(this.host, { type: 'error', error: serializeRunnerError(error) })
this.resultDelivered = true
} catch {
// The disconnected parent observes runner infrastructure failure.
}
}
this.releaseOwnedJob()
this.finish(127)
}
private releaseOwnedJob(): void {
if (this.pollTimer !== undefined) clearInterval(this.pollTimer)
this.pollTimer = undefined
const api = this.api
if (api === undefined) return
if (this.jobHandle !== undefined) {
try { this.internals.terminateJob(api, this.jobHandle, 1) } catch { /* Continue to kill-on-close. */ }
try { this.internals.closeHandleChecked(api, this.jobHandle, 'ordinary process Job cleanup') } catch { /* Best effort after failure. */ }
this.jobHandle = undefined
}
if (this.processHandle !== undefined) {
try { this.internals.closeHandleChecked(api, this.processHandle, 'ordinary direct process cleanup') } catch { /* Best effort after failure. */ }
this.processHandle = undefined
}
}
private finish(exitCode: number, disconnect = true): void {
if (this.finished) return
this.finished = true
if (this.pollTimer !== undefined) clearInterval(this.pollTimer)
this.pollTimer = undefined
this.host.off('message', this.onMessage)
this.host.off('disconnect', this.onDisconnect)
this.host.exitCode = exitCode
if (disconnect && this.host.connected) this.host.disconnect()
this.completion.resolve()
}
}
/**
* Execute the selected Linux bootstrap or Windows Job runner.
* @param selection - Windows sentinel or Linux launch-request locator.
* @param argv - private runner arguments beginning with the target delimiter.
* @param host - process transport and lifecycle host.
* @param internals - native and filesystem operations used by the runner.
*/
export async function runSpawnRunner(
selection: string,
argv: readonly string[],
host: RunnerHost = process,
internals: SpawnRunnerInternals = defaultInternals,
): Promise<void> {
Reflect.deleteProperty(host.env, SUBPROCESS_RUNNER_ENV)
const targetArgv = parseRunnerTargetArgv(argv)
if (selection === WINDOWS_RUNNER_SELECTION) {
await new WindowsJobRunner(targetArgv, host, internals).run()
return
}
runLinux(selection, targetArgv, host, internals)
}
/**
* Best-effort reporting for failures before the selected runner established its owner.
* @param selection - Windows sentinel, Linux launch-request locator, or no selection.
* @param error - failure raised before normal runner settlement.
* @param host - process transport and lifecycle host.
*/
export async function reportSpawnRunnerFailure(
selection: string | undefined,
error: unknown,
host: RunnerHost = process,
): Promise<void> {
if (selection === WINDOWS_RUNNER_SELECTION) {
try { await sendMessage(host, { type: 'error', error: serializeRunnerError(error) }) } catch { /* No transport remains. */ }
host.exitCode = 127
if (host.connected) host.disconnect()
return
}
if (selection !== undefined) {
try {
const files: LinuxLaunchFiles = linuxLaunchFilesFromLocator(selection)
writeLinuxStartupError(files, { type: 'error', error: serializeRunnerError(error) })
} catch {
// The parent will report an unconsumed request or missing runner result.
}
}
host.exitCode = 127
}