deepseek-harness/packages/subprocess/subprocess-local/src/spawn-runner.ts

/** One-shot Linux exec bootstrap and Windows Job-owning subprocess runner. */

import { posix } from 'node:path'
import {
  closeCurrentProcessStandardStreams,
  closeHandleChecked,
  isJobEmpty,
  loadWin32ProcessBindings,
  pollProcessExit,
  spawnCurrentTokenJobProcess,
  terminateJob,
  Win32Error,
} from '@deepseek-ai/dsh-win32-process'
import type { NativePtr, Win32ProcessBindings } from '@deepseek-ai/dsh-win32-process'
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>): never
  loadWin32ProcessBindings(): Win32ProcessBindings
  spawnCurrentTokenJobProcess: typeof spawnCurrentTokenJobProcess
  closeCurrentProcessStandardStreams: typeof closeCurrentProcessStandardStreams
  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) => (process.execve as NonNullable<typeof process.execve>)(file, argv, env),
  loadWin32ProcessBindings,
  spawnCurrentTokenJobProcess,
  closeCurrentProcessStandardStreams,
  resolveWindowsExecutable,
  pollProcessExit,
  isJobEmpty,
  terminateJob,
  closeHandleChecked,
}

function replaceEnvironment(target: NodeJS.ProcessEnv, env: Record<string, string>): void {
  for (const key of Object.keys(target)) Reflect.deleteProperty(target, key)
  Object.assign(target, env)
}

function asSpawnError(error: unknown, program: string, args: readonly string[]): SerializedRunnerError {
  const serialized = serializeRunnerError(error)
  const code = error instanceof Win32Error
    ? error.win32Code === 2 || error.win32Code === 3 || error.win32Code === 267
      ? 'ENOENT'
      : error.win32Code === 5
        ? 'EACCES'
        : error.win32Code === 193
          ? 'EFTYPE'
          : 'UNKNOWN'
    : serialized.code
  if (code === undefined) return serialized
  return {
    ...serialized,
    message: `spawn ${program} ${code}: ${serialized.message}`,
    code,
    syscall: `spawn ${program}`,
    path: program,
    spawnargs: [...args],
  }
}

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 execLinuxTarget(
  request: { cwd: string; env: Record<string, string> },
  argv: string[],
  internals: SpawnRunnerInternals,
): never {
  const program = argv[0] as string
  if (program.includes('/')) return internals.execve(program, argv, request.env)
  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)
    } 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: 'runner-error', error: serializeRunnerError(error) })
    host.exitCode = 127
    return
  }
  try {
    host.chdir(request.cwd)
    execLinuxTarget(request, argv, internals)
  } catch (error) {
    writeLinuxStartupError(files, {
      type: 'spawn-error',
      error: asSpawnError(error, argv[0] as string, argv.slice(1)),
    })
    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: Win32ProcessBindings | undefined
  private processHandle: NativePtr | undefined
  private jobHandle: NativePtr | undefined
  private pollTimer: ReturnType<typeof setInterval> | undefined
  private startSeen = false
  private committed = false
  private terminateRequested = false
  private resultStarted = false
  private resultDelivered = false
  private jobEmpty = 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: 'start-cancelled' }, 0)
      return
    }
    await new Promise<void>((resolveImmediate) => { setImmediate(resolveImmediate) })
    if (this.finished) return
    if (this.startCancellationPending()) {
      await this.publishTerminalResult({ type: 'start-cancelled' }, 0)
      return
    }
    try {
      const [command, ...args] = this.argv
      const applicationName = this.internals.resolveWindowsExecutable(
        command as string,
        request.cwd,
        request.env,
        undefined,
        { ...this.host.env },
      )
      replaceEnvironment(this.host.env, request.env)
      this.api = this.internals.loadWin32ProcessBindings()
      const spawned = this.internals.spawnCurrentTokenJobProcess(this.api, {
        command: command as string,
        applicationName,
        args,
        cwd: request.cwd,
      })
      this.processHandle = spawned.process
      this.jobHandle = spawned.job
      this.committed = true
      this.internals.closeCurrentProcessStandardStreams()
      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({
          type: 'spawn-error',
          error: asSpawnError(error, this.argv[0] as string, this.argv.slice(1)),
        }, 0)
        return
      }
      await this.runnerFailure(error)
    }
  }

  private requestTermination(): void {
    if (this.terminateRequested) return
    this.terminateRequested = true
    if (this.committed) {
      try {
        this.terminateOwnedJob()
      } catch (error) {
        void this.runnerFailure(error)
      }
    }
  }

  private startCancellationPending(): boolean {
    return this.terminateRequested
  }

  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, signal: null })
        }
      }
      if (this.jobHandle !== undefined && this.internals.isJobEmpty(this.api, this.jobHandle)) {
        this.internals.closeHandleChecked(this.api, this.jobHandle, 'ordinary process Job')
        this.jobHandle = undefined
        this.jobEmpty = true
        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.jobEmpty) 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: 'runner-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: 'runner-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: 'runner-error', error: serializeRunnerError(error) })
    } catch {
      // The parent will report an unconsumed request or missing runner result.
    }
  }
  host.exitCode = 127
}