winewill
cfa84ed4e3
OutputCollector.spillAll ran openSync/writeSync inside the stream 'data' listener with no containment. When the private per-process spill directory had been removed by a temporary-file cleaner, the first overflowing chunk threw ENOENT from Pipe.onStreamRead, became an uncaught exception in the Desktop Host, and the Electron shell reported it as a fatal failure. Spilling is best-effort by contract (spillPath is optional; a failed final close already withholds it). Every filesystem failure while opening or appending the spill file now discards the spill, reports once through the owner's logger, and leaves in-memory collection untouched. Spill inputs move into one SpillOptions object with the reporter, so tail-only collectors no longer pass an empty directory; prepareManagedProcessBinding is the explicit resolve step for the directory and reporter defaults, and the runtime and SSH helper supply their plugin loggers. Tests cover ENOENT on open (removed directory), ENOTDIR on open, and ENOSPC on append after the file exists, asserting the tail stays byte-exact, the spill file is withdrawn, and the reporter fires once.
346 lines
17 KiB
TypeScript
346 lines
17 KiB
TypeScript
/** Remote subprocess and PTY handles with independent SSH streams and helper-owned process lifetimes. */
|
|
import { Duplex, PassThrough, type Readable, type Writable } from 'node:stream'
|
|
import type { Socket } from 'node:net'
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
import { SubprocessRuntime, SubprocessExecutableNotFoundError } from '@deepseek-ai/dsh-subprocess'
|
|
import type { SubprocessCollectedOutputs, SubprocessHandle, SubprocessOutcome, SubprocessOutputMode, SubprocessSpawnSpec, SubprocessTerminalHandle, SubprocessTerminalEnvironment, SubprocessTerminalSignal, SubprocessTerminalSpawnSpec } from '@deepseek-ai/dsh-subprocess'
|
|
import { OutputCollector } from '@deepseek-ai/dsh-subprocess-local/output'
|
|
import type { SshConnection } from '@deepseek-ai/dsh-ssh'
|
|
import { doneSchema, foregroundSchema, terminalActivitySchema, outputSnapshotFrameLimit, outputSnapshotSchema, preparedSchema, remotePath, streamEndpointSchema } from '@deepseek-ai/dsh-ssh/schemas'
|
|
import type { SshProcessId } from '@deepseek-ai/dsh-ssh/schemas'
|
|
import { SshRpcPeer, RemoteOperationError } from '@deepseek-ai/dsh-ssh/protocol'
|
|
import { z } from 'zod'
|
|
|
|
function environment(env?: NodeJS.ProcessEnv): Record<string, string | null> | undefined {
|
|
return env === undefined ? undefined : Object.fromEntries(Object.entries(env).map(([key, value]) => [key, value ?? null]))
|
|
}
|
|
|
|
class RemoteCleanupError extends AggregateError {}
|
|
|
|
/** One remote ordinary process; stdin and control remain usable during asynchronous SSH allocation. */
|
|
class RemoteProcess implements SubprocessHandle {
|
|
readonly stdin: Writable | undefined
|
|
readonly stdout: Readable | undefined
|
|
readonly stderr: Readable | undefined
|
|
readonly control: Duplex | undefined
|
|
readonly collected: SubprocessCollectedOutputs
|
|
readonly done: Promise<SubprocessOutcome>
|
|
readonly streamsClosed: Promise<void>
|
|
private readonly inbound = new PassThrough()
|
|
private readonly out = new PassThrough()
|
|
private readonly err = new PassThrough()
|
|
private readonly toControl = new PassThrough()
|
|
private readonly fromControl = new PassThrough()
|
|
private readonly controller = new AbortController()
|
|
private readonly started: Promise<void>
|
|
private id: SshProcessId | undefined
|
|
private sockets: Socket[] = []
|
|
private quiescent = false
|
|
private committed = false
|
|
private termination: Promise<void> | undefined
|
|
private readonly detachAbort: () => void
|
|
private spills: { stdout?: string | undefined; stderr?: string | undefined } = {}
|
|
private readonly updateCollection: Partial<Record<'stdout' | 'stderr', (snapshot: { tail: string; totalBytes: number }, final: boolean) => void>> = {}
|
|
|
|
constructor(private readonly ssh: SshConnection, private readonly spec: SubprocessSpawnSpec) {
|
|
this.stdin = spec.stdio.stdin === 'pipe' ? this.inbound : undefined
|
|
this.stdout = spec.stdio.stdout === 'pipe' ? this.out : undefined
|
|
this.stderr = spec.stdio.stderr === 'pipe' ? this.err : undefined
|
|
const requestedControl = (spec.stdio as typeof spec.stdio & { control?: 'pipe' }).control
|
|
this.control = requestedControl === undefined ? undefined : Duplex.from({ writable: this.toControl, readable: this.fromControl })
|
|
for (const stream of [this.inbound, this.out, this.err, this.toControl, this.fromControl, this.control]) stream?.on('error', () => {})
|
|
const collect = (name: 'stdout' | 'stderr', stream: PassThrough, mode: SubprocessOutputMode) => {
|
|
if (mode === 'pipe') return undefined
|
|
if (mode === 'inherit') { stream.pipe(name === 'stdout' ? process.stdout : process.stderr, { end: false }); return undefined }
|
|
let collector = new OutputCollector(mode.maxBytes, name, undefined)
|
|
let base = 0
|
|
let total = 0
|
|
let finalized = false
|
|
this.updateCollection[name] = (snapshot, final): void => {
|
|
const bytes = Buffer.from(snapshot.tail, 'base64')
|
|
if (bytes.length > mode.maxBytes || snapshot.totalBytes < bytes.length) throw new Error('SSH helper returned invalid collected output coordinates')
|
|
if (finalized) return
|
|
if (snapshot.totalBytes < total) throw new Error('SSH helper rewound collected output')
|
|
finalized = final
|
|
collector = new OutputCollector(mode.maxBytes, name, undefined)
|
|
collector.push(bytes)
|
|
base = snapshot.totalBytes - bytes.length
|
|
total = snapshot.totalBytes
|
|
}
|
|
return {
|
|
readFrom: (offset: number) => {
|
|
const snapshot = collector.readFrom(offset - base)
|
|
return {
|
|
...snapshot, nextOffset: snapshot.nextOffset + base,
|
|
...(this.spills[name] === undefined ? {} : { spillPath: this.spills[name] }),
|
|
}
|
|
},
|
|
}
|
|
}
|
|
const stdout = collect('stdout', this.out, spec.stdio.stdout)
|
|
const stderr = collect('stderr', this.err, spec.stdio.stderr)
|
|
this.collected = { ...(stdout === undefined ? {} : { stdout }), ...(stderr === undefined ? {} : { stderr }) }
|
|
const onAbort = (): void => { this.terminate() }
|
|
spec.signal?.addEventListener('abort', onAbort, { once: true })
|
|
this.detachAbort = () => { spec.signal?.removeEventListener('abort', onAbort) }
|
|
this.started = this.start()
|
|
this.streamsClosed = this.started.then(async () => {
|
|
await Promise.all(this.sockets.map(socket => socket.closed ? Promise.resolve() : new Promise<void>((resolve) => {
|
|
socket.once('close', () => { resolve() })
|
|
})))
|
|
}, () => {})
|
|
this.done = this.started.then(async () => {
|
|
const result = await ssh.request('process.done', { id: this.id }, doneSchema, undefined, true)
|
|
await this.drainOutput()
|
|
this.spills = result.spills
|
|
for (const name of ['stdout', 'stderr'] as const) {
|
|
const snapshot = result.collected[name]
|
|
const finish = this.updateCollection[name]
|
|
if ((snapshot === undefined) !== (finish === undefined)) throw new Error('SSH helper returned mismatched output collection modes')
|
|
if (snapshot !== undefined) finish?.(snapshot, true)
|
|
}
|
|
return { exitCode: result.outcome.exitCode, signal: result.outcome.signal as NodeJS.Signals | null }
|
|
}).catch((error: unknown) => {
|
|
this.terminate()
|
|
for (const socket of this.sockets) socket.destroy()
|
|
for (const stream of [this.inbound, this.out, this.err, this.toControl, this.fromControl]) {
|
|
stream.destroy(error instanceof Error ? error : new Error(String(error)))
|
|
}
|
|
throw error
|
|
})
|
|
void this.done.catch(() => {})
|
|
}
|
|
|
|
private async start(): Promise<void> {
|
|
this.spec.signal?.throwIfAborted()
|
|
const prepared = await this.ssh.request(
|
|
'process.prepare', { ...this.spec, signal: undefined, env: environment(this.spec.env) }, preparedSchema, this.controller.signal,
|
|
)
|
|
this.id = prepared.id
|
|
try {
|
|
const sockets = await Promise.all(Object.entries(prepared.streams).map(async ([name, path]) =>
|
|
[name, await this.ssh.connectStream(path, this.controller.signal)] as const))
|
|
this.sockets = sockets.map(([, socket]) => socket)
|
|
for (const [name, socket] of sockets) {
|
|
if (name === 'stdout' || name === 'stderr') {
|
|
const mode = this.spec.stdio[name]
|
|
if (typeof mode === 'object') {
|
|
new SshRpcPeer(socket, socket, outputSnapshotFrameLimit(mode.maxBytes), 1, (method, raw) => Promise.resolve().then(() => {
|
|
if (method !== 'snapshot') throw new Error('Unexpected SSH output-stream operation')
|
|
this.updateCollection[name]?.(outputSnapshotSchema.parse(raw), false)
|
|
return null
|
|
}))
|
|
} else {
|
|
socket.end()
|
|
const output = name === 'stdout' ? this.out : this.err
|
|
const closeSocket = (): void => { socket.destroy() }
|
|
output.once('close', closeSocket)
|
|
socket.once('close', () => { output.off('close', closeSocket) })
|
|
if (output.destroyed) closeSocket()
|
|
else socket.pipe(output)
|
|
}
|
|
}
|
|
if (name === 'stdin') this.inbound.pipe(socket)
|
|
if (name === 'control') { this.toControl.pipe(socket); socket.pipe(this.fromControl) }
|
|
}
|
|
this.controller.signal.throwIfAborted()
|
|
await this.ssh.request('process.start', { id: this.id }, z.object({}).strict(), this.controller.signal)
|
|
this.committed = true
|
|
} catch (error) {
|
|
this.terminate()
|
|
await this.termination?.catch(() => {})
|
|
for (const socket of this.sockets) socket.destroy()
|
|
throw error
|
|
}
|
|
}
|
|
|
|
terminate(): void {
|
|
if (this.quiescent || this.termination !== undefined) return
|
|
if (!this.committed) this.controller.abort(new Error('SSH process terminated before launch acknowledgement'))
|
|
if (this.id !== undefined) {
|
|
this.termination = this.ssh.request('process.terminate', { id: this.id }, z.null()).then(() => {
|
|
this.quiescent = true
|
|
this.detachAbort()
|
|
}).catch((error: unknown) => {
|
|
// A connection unable to confirm termination must release its helper lease.
|
|
void this.ssh.dispose().catch(() => {})
|
|
throw error
|
|
})
|
|
void this.termination.catch(() => {})
|
|
}
|
|
}
|
|
|
|
closeStreams(): void {
|
|
for (const socket of this.sockets) socket.destroy()
|
|
for (const stream of [this.inbound, this.out, this.err, this.toControl, this.fromControl, this.control]) stream?.destroy()
|
|
}
|
|
|
|
async waitForExit(signal?: AbortSignal): Promise<boolean> {
|
|
if (this.quiescent) return true
|
|
if (signal?.aborted) return false
|
|
if (signal === undefined) return this.observeExit()
|
|
const cancelled = Promise.withResolvers<boolean>()
|
|
const abort = (): void => { cancelled.resolve(false) }
|
|
signal.addEventListener('abort', abort, { once: true })
|
|
try { return await Promise.race([this.observeExit(signal), cancelled.promise]) }
|
|
finally { signal.removeEventListener('abort', abort) }
|
|
}
|
|
|
|
private async observeExit(signal?: AbortSignal): Promise<boolean> {
|
|
try { await this.started } catch {
|
|
// Startup errors remain on done; a prepared process must first finish termination.
|
|
if (this.termination !== undefined) await this.termination
|
|
this.quiescent = true
|
|
this.detachAbort()
|
|
return true
|
|
}
|
|
if (this.termination !== undefined) { await this.termination; return true }
|
|
const result = await this.ssh.request('process.wait', { id: this.id }, z.boolean(), signal, true)
|
|
if (result) { this.quiescent = true; this.detachAbort() }
|
|
return result
|
|
}
|
|
|
|
private async drainOutput(): Promise<void> {
|
|
const disposers: Array<() => void> = []
|
|
const outputs = (['stdout', 'stderr'] as const).map((name) => {
|
|
if (typeof this.spec.stdio[name] === 'object') return Promise.resolve()
|
|
const stream = name === 'stdout' ? this.out : this.err
|
|
if (stream.readableEnded || stream.destroyed) return Promise.resolve()
|
|
return new Promise<void>((resolve) => {
|
|
const done = (): void => { resolve() }
|
|
for (const event of ['end', 'close', 'error']) stream.once(event, done)
|
|
disposers.push(() => { for (const event of ['end', 'close', 'error']) stream.off(event, done) })
|
|
})
|
|
})
|
|
let timer: NodeJS.Timeout | undefined
|
|
try {
|
|
await Promise.race([
|
|
Promise.all(outputs),
|
|
new Promise<void>((resolve) => { timer = setTimeout(resolve, this.spec.graceMs) }),
|
|
])
|
|
} finally {
|
|
clearTimeout(timer)
|
|
for (const dispose of disposers) dispose()
|
|
}
|
|
}
|
|
}
|
|
|
|
/** SSH provider paired with the SSH filesystem; the remote helper selects POSIX process ownership. */
|
|
export class SshSubprocessRuntime extends SubprocessRuntime {
|
|
static inject = ['ssh']
|
|
private readonly live = new Set<RemoteProcess>()
|
|
private readonly terminals = new Set<SubprocessTerminalHandle>()
|
|
private readonly terminalAllocations = new Set<Promise<SubprocessTerminalHandle>>()
|
|
private readonly lifetime = new AbortController()
|
|
|
|
constructor(ctx: Context) {
|
|
super(ctx)
|
|
ctx.effect(() => async () => {
|
|
this.lifetime.abort(new Error('SSH subprocess provider disposed'))
|
|
for (const handle of this.live) handle.terminate()
|
|
const results = await Promise.allSettled([
|
|
...[...this.live].map(async (handle) => {
|
|
try { await handle.waitForExit() } finally { handle.closeStreams() }
|
|
}),
|
|
...[...this.terminalAllocations].map(allocation => allocation.catch((error: unknown) => {
|
|
if (error instanceof RemoteCleanupError) throw error
|
|
})),
|
|
...[...this.terminals].map(handle => handle.terminate()),
|
|
])
|
|
const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason as unknown] : [])
|
|
if (errors.length > 0) throw new AggregateError(errors, 'SSH process cleanup could not be confirmed')
|
|
})
|
|
}
|
|
|
|
override async resolveExecutable(command: string, env?: Readonly<Record<string, string>>, signal?: AbortSignal): Promise<string> {
|
|
try {
|
|
return await this.ctx.ssh.request('executable', { command, env }, remotePath, signal)
|
|
} catch (error) {
|
|
if (error instanceof RemoteOperationError && error.code === 'SUBPROCESS_EXECUTABLE_NOT_FOUND') {
|
|
throw new SubprocessExecutableNotFoundError(error.message, { cause: error })
|
|
}
|
|
throw error
|
|
}
|
|
}
|
|
|
|
override terminalEnvironment(signal?: AbortSignal): Promise<SubprocessTerminalEnvironment> {
|
|
return this.ctx.ssh.request('terminal.environment', {}, z.object({
|
|
platform: z.enum(['posix', 'windows']), defaultShell: z.string().optional(),
|
|
}).strict().transform(value => ({ platform: value.platform,
|
|
...(value.defaultShell === undefined ? {} : { defaultShell: value.defaultShell }),
|
|
})), signal)
|
|
}
|
|
|
|
override spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
|
|
this.lifetime.signal.throwIfAborted()
|
|
spec.signal?.throwIfAborted()
|
|
const handle = new RemoteProcess(this.ctx.ssh, spec)
|
|
this.live.add(handle)
|
|
void handle.done.then(() => handle.waitForExit()).then(() => handle.streamsClosed)
|
|
.then(() => { this.live.delete(handle) }).catch(() => {})
|
|
return handle
|
|
}
|
|
|
|
override async spawnTerminal(spec: SubprocessTerminalSpawnSpec): Promise<SubprocessTerminalHandle> {
|
|
this.lifetime.signal.throwIfAborted()
|
|
const signal = spec.signal === undefined ? this.lifetime.signal : AbortSignal.any([spec.signal, this.lifetime.signal])
|
|
signal.throwIfAborted()
|
|
const allocation = this.createTerminal(spec, signal)
|
|
this.terminalAllocations.add(allocation)
|
|
try { return await allocation } finally { this.terminalAllocations.delete(allocation) }
|
|
}
|
|
|
|
private async createTerminal(spec: SubprocessTerminalSpawnSpec, signal: AbortSignal): Promise<SubprocessTerminalHandle> {
|
|
const ssh = this.ctx.ssh
|
|
const prepared = await ssh.request('process.prepare', {
|
|
argv: spec.argv, cwd: spec.cwd, env: environment(spec.env), graceMs: spec.graceMs,
|
|
terminal: { rows: spec.rows, cols: spec.cols, terminalType: spec.terminalType, shellActivity: spec.shellActivity },
|
|
}, preparedSchema, signal)
|
|
const id = prepared.id
|
|
let socket: Socket | undefined
|
|
try {
|
|
signal.throwIfAborted()
|
|
socket = await ssh.connectStream(streamEndpointSchema.parse(prepared.streams.terminal), signal)
|
|
signal.throwIfAborted()
|
|
socket.end()
|
|
const output = new PassThrough()
|
|
socket.pipe(output)
|
|
const started = await ssh.request('process.start', { id }, z.object({ pid: z.number().int().positive() }).strict(), signal)
|
|
signal.throwIfAborted()
|
|
const done = ssh.request('process.done', { id }, doneSchema, undefined, true).then(result => ({ exitCode: result.outcome.exitCode, signal: result.outcome.signal as NodeJS.Signals | null }))
|
|
void done.catch(() => {})
|
|
let closing: Promise<void> | undefined
|
|
const abort = (): void => { void handle.terminate().catch(() => { void ssh.dispose().catch(() => {}) }) }
|
|
const handle: SubprocessTerminalHandle = {
|
|
pid: started.pid, output, done,
|
|
resize: async (cols, rows) => { await ssh.request('terminal.resize', { id, cols, rows }, z.null()) },
|
|
write: async (data) => { await ssh.request('terminal.write', { id, value: data }, z.null()) },
|
|
inspectForeground: async () => await ssh.request('terminal.inspect', { id }, foregroundSchema) ?? undefined,
|
|
inspectActivity: () => ssh.request('terminal.activity', { id }, terminalActivitySchema),
|
|
signalForeground: (signal: SubprocessTerminalSignal) => ssh.request('terminal.signal', { id, value: signal }, z.number().int().positive()),
|
|
terminate: () => {
|
|
closing ??= ssh.request('process.terminate', { id }, z.null(), undefined, true).then(() => {
|
|
socket?.destroy()
|
|
output.destroy()
|
|
signal.removeEventListener('abort', abort)
|
|
this.terminals.delete(handle)
|
|
}).catch((error: unknown) => { closing = undefined; throw error })
|
|
return closing
|
|
},
|
|
}
|
|
signal.addEventListener('abort', abort, { once: true })
|
|
this.terminals.add(handle)
|
|
return handle
|
|
} catch (error) {
|
|
socket?.destroy()
|
|
try { await ssh.request('process.terminate', { id }, z.null()) } catch (cleanupError) {
|
|
void ssh.dispose().catch(() => {})
|
|
throw new RemoteCleanupError([error, cleanupError], 'SSH terminal allocation failed and remote cleanup is unknown')
|
|
}
|
|
throw error
|
|
}
|
|
}
|
|
}
|
|
|
|
export default SshSubprocessRuntime
|