Merge origin/master into codex/truncated-design

This commit is contained in:
Dudu-0223
2026-07-17 18:21:54 +08:00
1057 changed files with 42961 additions and 18526 deletions

View File

@@ -1,11 +1,10 @@
import { mkdtempSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
import { BashTaskId } from '@deepseek-ai/dsh-bash'
import type { BashTaskRead } from '@deepseek-ai/dsh-bash'
import type { BashProcess } from '@deepseek-ai/dsh-bash'
const spillDir = mkdtempSync(join(tmpdir(), 'dsh-bash-exec-spec-'))
@@ -18,36 +17,20 @@ async function setup(config: ConstructorParameters<typeof LocalBashExecutor>[1]
return { ctx, bash }
}
/** Poll until a pid no longer exists. */
async function waitGone(pid: number, timeoutMs = 5_000): Promise<void> {
/**
* Poll a handle's consuming readOutput until the ACCUMULATED delta contains
* `expected`; returns the accumulation (reads never re-deliver, so the caller
* gets everything produced up to the match).
*/
async function readUntil(proc: BashProcess, expected: string, timeoutMs = 5_000): Promise<string> {
const deadline = Date.now() + timeoutMs
let all = ''
while (Date.now() < deadline) {
try {
process.kill(pid, 0)
} catch {
return
}
all += proc.readOutput().delta
if (all.includes(expected)) return all
await new Promise(resolve => setTimeout(resolve, 20))
}
throw new Error(`pid ${pid} still alive after ${timeoutMs}ms`)
}
async function readUntil(
bash: LocalBashExecutor,
id: BashTaskId,
expected: string,
timeoutMs = 5_000,
): Promise<BashTaskRead> {
const deadline = Date.now() + timeoutMs
let last: BashTaskRead | undefined
let delta = ''
while (Date.now() < deadline) {
last = bash.readOutput(id)
delta += last.delta
if (delta.includes(expected)) return { ...last, delta }
await new Promise(resolve => setTimeout(resolve, 20))
}
throw new Error(`task ${id} output did not include ${JSON.stringify(expected)}; output was ${JSON.stringify(delta)}, last delta was ${JSON.stringify(last?.delta ?? '')}`)
throw new Error(`process output did not include ${JSON.stringify(expected)}; accumulated ${JSON.stringify(all)}`)
}
describe('LocalBashExecutor.run', () => {
@@ -107,15 +90,6 @@ describe('LocalBashExecutor.run', () => {
expect(result.stderr.text.length).toBeLessThanOrEqual(100)
})
it('kill escalation uses the configured graceMs (a TERM-trapping task dies by SIGKILL)', async () => {
const { bash } = await setup() // setup pins graceMs: 200 via config
const task = bash.start(bash.resolve({ command: 'trap \'\' TERM; echo ready; while :; do sleep 60 & wait $!; done' }))
await readUntil(bash, task.id, 'ready\n')
bash.kill(task.id)
await task.done
expect(task.signal).toBe('SIGKILL')
})
it('per-call timeout takes precedence under the cap and kills on expiry', async () => {
const { bash } = await setup({ timeoutMs: 60_000 })
const result = await bash.run(bash.resolve({ command: 'sleep 60', timeoutMs: 100 }))
@@ -171,229 +145,183 @@ describe('LocalBashExecutor.run', () => {
})
})
describe('LocalBashExecutor background tasks', () => {
it('start returns immediately with a registered running task', async () => {
describe('LocalBashExecutor.start (background process handles)', () => {
it('start returns immediately with a running handle that settles as completed', async () => {
const { bash } = await setup()
const before = Date.now()
const task = bash.start(bash.resolve({ command: 'sleep 0.2; echo done' }))
const proc = bash.start(bash.resolve({ command: 'sleep 0.2; echo done' }))
expect(Date.now() - before).toBeLessThan(150)
expect(task.status).toBe('running')
expect(bash.get(task.id)).toBe(task)
expect(bash.list()).toContain(task)
await task.done
expect(task.status).toBe('completed')
expect(task.exitCode).toBe(0)
expect(proc.status).toBe('running')
await proc.done
expect(proc.status).toBe('completed')
expect(proc.exitCode).toBe(0)
})
it('assigns sequential ids', async () => {
it('threads stdin and extra env into a background process', async () => {
const { bash } = await setup()
const first = bash.start(bash.resolve({ command: 'true' }))
const second = bash.start(bash.resolve({ command: 'true' }))
expect(first.id).toBe('bash-1')
expect(second.id).toBe('bash-2')
await Promise.all([first.done, second.done])
})
it('threads stdin and extra env into a background task', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({
const proc = bash.start(bash.resolve({
command: 'cat; echo "[$DSH_BG_VAR]"',
stdin: 'bg-stdin\n',
env: { DSH_BG_VAR: 'bg-env' },
}))
const read = await readUntil(bash, task.id, '[bg-env]')
expect(read.delta).toContain('bg-stdin')
await task.done
expect(task.exitCode).toBe(0)
const output = await readUntil(proc, '[bg-env]')
expect(output).toContain('bg-stdin')
await proc.done
expect(proc.exitCode).toBe(0)
})
it('readOutput returns increments without re-delivery', async () => {
it('readOutput is consuming: increments are never re-delivered, and reads stay valid after exit', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'echo first; sleep 1; echo second' }))
const first = await readUntil(bash, task.id, 'first\n')
expect(first.delta).toBe('first\n')
expect(first.lossy).toBe(false)
await task.done
const second = bash.readOutput(task.id)
const proc = bash.start(bash.resolve({ command: 'echo first; sleep 1; echo second' }))
const first = await readUntil(proc, 'first\n')
expect(first).toBe('first\n')
await proc.done
// Read-after-exit returns the remaining buffered output — once.
const second = proc.readOutput()
expect(second.delta).toBe('second\n')
const third = bash.readOutput(task.id)
expect(third.delta).toBe('')
expect(second.lossy).toBe(false)
expect(proc.readOutput().delta).toBe('')
})
it('readOutput marks stderr sections', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'echo out; echo err >&2' }))
await task.done
const read = bash.readOutput(task.id)
expect(read.delta).toBe('out\n[stderr]\nerr\n')
const proc = bash.start(bash.resolve({ command: 'echo out; echo err >&2' }))
await proc.done
expect(proc.readOutput().delta).toBe('out\n[stderr]\nerr\n')
})
it('readOutput reports stderr-only deltas without a leading newline', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'echo err >&2' }))
await task.done
expect(bash.readOutput(task.id).delta).toBe('[stderr]\nerr\n')
const proc = bash.start(bash.resolve({ command: 'echo err >&2' }))
await proc.done
expect(proc.readOutput().delta).toBe('[stderr]\nerr\n')
})
it('readOutput flags lossy reads and reports spill paths', async () => {
it('readOutput adds a separator only when stdout lacks a trailing newline', async () => {
const { bash } = await setup()
const proc = bash.start(bash.resolve({ command: 'printf out; echo err >&2' }))
await proc.done
expect(proc.readOutput().delta).toBe('out\n[stderr]\nerr\n')
})
it('readOutput flags lossy reads and reports stdout spill paths', async () => {
const { bash } = await setup({ maxOutputBytes: 100 })
const task = bash.start(bash.resolve({ command: 'for i in $(seq 1 100); do printf "line-%04d\\n" $i; done' }))
await task.done
const read = bash.readOutput(task.id)
const proc = bash.start(bash.resolve({ command: 'for i in $(seq 1 100); do printf "line-%04d\\n" $i; done' }))
await proc.done
const read = proc.readOutput()
// Window slid past offset 0 → lossy, spill path points at the full stream.
expect(read.lossy).toBe(true)
expect(read.stdoutSpillPath).toBeDefined()
})
it('readOutput throws for unknown ids', async () => {
const { bash } = await setup()
expect(() => bash.readOutput(BashTaskId('nope'))).toThrow(/unknown bash task "nope"/)
})
it('kill terminates the process group and reports status killed', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'sleep 60' }))
expect(bash.kill(task.id)).toBe(true)
await task.done
expect(task.status).toBe('killed')
expect(task.signal).toBe('SIGTERM')
})
it('kill returns false for finished tasks and throws for unknown ids', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'true' }))
await task.done
expect(bash.kill(task.id)).toBe(false)
expect(() => bash.kill(BashTaskId('nope'))).toThrow(/unknown bash task "nope"/)
})
it('notifies onTaskDone listeners on completion', async () => {
const { bash } = await setup()
const seen: [string, string][] = []
bash.onTaskDone(task => void seen.push([task.id, task.status]))
const task = bash.start(bash.resolve({ command: 'true' }))
await task.done
expect(seen).toEqual([[task.id, 'completed']])
})
it('notifies onTaskDone for killed tasks too', async () => {
const { bash } = await setup()
const listener = vi.fn()
bash.onTaskDone(listener)
const task = bash.start(bash.resolve({ command: 'sleep 60' }))
bash.kill(task.id)
await task.done
expect(listener).toHaveBeenCalledWith(task)
expect(task.status).toBe('killed')
})
it('marks tasks killed when the background spawn itself fails', async () => {
const { bash } = await setup()
const listener = vi.fn()
bash.onTaskDone(listener)
const task = bash.start(bash.resolve({ command: 'true', workdir: '/nonexistent-dsh' }))
await task.done
expect(task.status).toBe('killed')
expect(listener).toHaveBeenCalledWith(task)
expect(bash.readOutput(task.id).delta).toContain('spawn failed')
})
it('readOutput adds a separator only when stdout lacks a trailing newline', async () => {
const { bash } = await setup()
const task = bash.start(bash.resolve({ command: 'printf out; echo err >&2' }))
await task.done
expect(bash.readOutput(task.id).delta).toBe('out\n[stderr]\nerr\n')
})
it('readOutput reports stderr spill paths', async () => {
const { bash } = await setup({ maxOutputBytes: 100 })
const task = bash.start(bash.resolve({ command: 'for i in $(seq 1 100); do printf "line-%04d\\n" $i >&2; done' }))
await task.done
const read = bash.readOutput(task.id)
const proc = bash.start(bash.resolve({ command: 'for i in $(seq 1 100); do printf "line-%04d\\n" $i >&2; done' }))
await proc.done
const read = proc.readOutput()
expect(read.lossy).toBe(true)
expect(read.stderrSpillPath).toBeDefined()
expect(read.delta).toContain('[stderr]')
})
it('disposing with already-finished tasks only kills the running ones', async () => {
it('kill() terminates the process group: true once, false after settlement', async () => {
const { bash } = await setup()
const proc = bash.start(bash.resolve({ command: 'sleep 60' }))
expect(proc.kill()).toBe(true)
await proc.done
expect(proc.status).toBe('killed')
expect(proc.signal).toBe('SIGTERM')
expect(proc.kill()).toBe(false)
})
it('kill() returns false for a naturally completed process', async () => {
const { bash } = await setup()
const proc = bash.start(bash.resolve({ command: 'true' }))
await proc.done
expect(proc.status).toBe('completed')
expect(proc.kill()).toBe(false)
})
it('kill escalation uses the configured graceMs (a TERM-trapping process dies by SIGKILL)', async () => {
const { bash } = await setup() // setup pins graceMs: 200 via config
// The child echoes AFTER arming the trap, so waiting for the marker
// guarantees SIGTERM is already ignored when the kill lands (a fixed sleep
// is load-flaky: a slow spawn would take the SIGTERM before the trap).
const proc = bash.start(bash.resolve({ command: 'trap \'\' TERM; echo armed; sleep 60' }))
await readUntil(proc, 'armed')
proc.kill()
await proc.done
expect(proc.status).toBe('killed')
expect(proc.signal).toBe('SIGKILL')
})
it('a spec.signal abort settles the handle as killed, not completed', async () => {
const { bash } = await setup()
const controller = new AbortController()
const proc = bash.start(bash.resolve({ command: 'sleep 60', signal: controller.signal }))
controller.abort()
await proc.done
expect(proc.status).toBe('killed')
expect(proc.signal).toBe('SIGTERM')
})
it('a self-signal exit settles the handle as killed, not completed', async () => {
const { bash } = await setup()
const proc = bash.start(bash.resolve({ command: 'kill -TERM $$' }))
await proc.done
expect(proc.status).toBe('killed')
expect(proc.exitCode).toBeNull()
expect(proc.signal).toBe('SIGTERM')
})
it('a background spawn failure settles as killed with the error readable on stderr', async () => {
const { bash } = await setup()
const proc = bash.start(bash.resolve({ command: 'true', workdir: '/nonexistent-dsh' }))
// done resolves (never rejects) even though the process never ran.
await expect(proc.done).resolves.toBeUndefined()
expect(proc.status).toBe('killed')
expect(proc.readOutput().delta).toContain('spawn failed:')
})
})
describe('LocalBashExecutor disposal', () => {
it('disposing the fiber kills running processes and AWAITS their exit (no orphans, SIGKILL escalation included)', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(LocalBashExecutor, { graceMs: 200 })
const bash = ctx.bash as LocalBashExecutor
bash.internals = { spillDir }
const finished = bash.start(bash.resolve({ command: 'true' }))
// The child prints its own pid ($$ = the detached bash group leader) so
// the test can probe liveness through the public read surface alone.
const proc = bash.start(bash.resolve({ command: 'trap \'\' TERM; echo $$; sleep 60' }))
const pid = Number((await readUntil(proc, '\n')).trim())
expect(Number.isInteger(pid) && pid > 0).toBe(true)
await fiber.dispose()
// Disposal itself waited: the pid must already be gone, no grace left —
// even for a TERM-trapping child held until the SIGKILL escalation landed.
expect(() => process.kill(pid, 0)).toThrow()
expect(proc.status).toBe('killed')
await proc.done
})
it('settled processes already left the live map: dispose does not touch them', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(LocalBashExecutor, { graceMs: 200 })
const bash = ctx.bash as LocalBashExecutor
bash.internals = { spillDir }
const finished = bash.start(bash.resolve({ command: 'echo done' }))
await finished.done
expect(finished.status).toBe('completed')
const running = bash.start(bash.resolve({ command: 'sleep 60' }))
await fiber.dispose()
await running.done
// The teardown marks every LIVE entry killed; a settled process had
// already left the map, so its status stays completed.
expect(finished.status).toBe('completed')
expect(running.status).toBe('killed')
await running.done
expect(running.signal).toBe('SIGTERM')
expect(bash.list()).toEqual([])
})
it('disposing the executor fiber kills running tasks (no orphans)', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(LocalBashExecutor, { graceMs: 200 })
const bash = ctx.bash as LocalBashExecutor
bash.internals = { spillDir }
const listener = vi.fn()
bash.onTaskDone(listener)
const task = bash.start(bash.resolve({ command: 'sleep 60' }))
const running = bash.get(task.id)!
await new Promise(resolve => setTimeout(resolve, 50))
// Grab the pid before dispose clears the registry.
const pid = (running as unknown as { running: { pid: number } }).running.pid
await fiber.dispose()
await waitGone(pid)
expect(bash.list()).toEqual([])
// Listener silenced by base-class teardown — no late notifications.
expect(listener).not.toHaveBeenCalled()
})
})
describe('review fixes: lifecycle hardening', () => {
it('start honors a pre-aborted or later-aborted AbortSignal', async () => {
const { bash } = await setup()
const controller = new AbortController()
const task = bash.start(bash.resolve({ command: 'sleep 60', signal: controller.signal }))
controller.abort()
await task.done
expect(task.status).toBe('killed')
expect(task.signal).toBe('SIGTERM')
})
it('a throwing onTaskDone listener does not reject task.done or starve later listeners', async () => {
const { bash } = await setup()
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
const second = vi.fn()
try {
bash.onTaskDone(() => { throw new Error('listener bug') })
bash.onTaskDone(second)
const task = bash.start(bash.resolve({ command: 'true' }))
await expect(task.done).resolves.toBeUndefined()
expect(second).toHaveBeenCalledWith(task)
expect(errorSpy).toHaveBeenCalled()
} finally {
errorSpy.mockRestore()
}
})
it('dispose AWAITS a TERM-trapping process (SIGKILL escalation included)', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(LocalBashExecutor, { graceMs: 200 })
const bash = ctx.bash as LocalBashExecutor
bash.internals = { spillDir }
const task = bash.start(bash.resolve({ command: 'trap \'\' TERM; sleep 60' }))
await new Promise(resolve => setTimeout(resolve, 100))
const pid = (task as unknown as { running: { pid: number } }).running.pid
await fiber.dispose()
// Disposal itself waited: the pid must already be gone, no grace left.
expect(() => process.kill(pid, 0)).toThrow()
expect(task.status).toBe('killed')
})
})

View File

@@ -2,8 +2,8 @@ import { mkdtempSync, readFileSync, statSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { dirname, join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { killGroup, OutputCollector, runBash } from '@deepseek-ai/dsh-bash-local'
import type { RunningBash } from '@deepseek-ai/dsh-bash-local'
import { killGroup, OutputCollector, runBash } from '../src/run.ts'
import type { RunningBash } from '../src/run.ts'
const { failNextClose } = vi.hoisted(() => ({ failNextClose: { value: false } }))
vi.mock('node:fs', async (importOriginal) => {
@@ -50,7 +50,7 @@ async function waitGone(pid: number, timeoutMs = 5_000): Promise<void> {
async function waitForStdout(running: RunningBash, expected: string, timeoutMs = 5_000): Promise<void> {
const deadline = Date.now() + timeoutMs
while (Date.now() < deadline) {
if (running.stdout.snapshot().text.includes(expected)) return
if (running.stdout.readFrom(0).text.includes(expected)) return
await new Promise(resolve => setTimeout(resolve, 20))
}
throw new Error(`stdout did not include ${JSON.stringify(expected)} after ${timeoutMs}ms`)
@@ -190,12 +190,8 @@ describe('stdin and extra env (set by in-process plugins)', () => {
})
it('gives fd 0 the exact pre-seam type: /dev/null when no stdin, a pipe when supplied', async () => {
// The no-stdin path must stay observationally identical to the pre-seam
// `ignore` default: a command that probes stdin's file type sees a char
// device (/dev/null). Regressing to an always-open pipe would make fd 0 a
// socket (node's spawn pipe is an AF_UNIX socket, not a FIFO), flipping
// `test -c /dev/stdin` for every model-driven call. When bytes ARE supplied,
// fd 0 is that pipe (a socket), as it must be to carry them.
// With no bytes, fd 0 remains the pre-seam `ignore` default (/dev/null, a character device).
// Supplied bytes use Node's spawn pipe, which is an AF_UNIX socket rather than a FIFO.
const none = await runBash(spec('test -c /dev/stdin && echo char || echo other')).done
expect(none.stdout.text).toBe('char\n')
const piped = await runBash(spec('test -S /dev/stdin && echo socket || echo other', { stdin: 'x' })).done
@@ -220,9 +216,8 @@ describe('stdin and extra env (set by in-process plugins)', () => {
})
it('does not crash or reject when the child ignores a large stdin (EPIPE)', async () => {
// The child exits immediately without reading; closing our end of a stdin
// pipe still holding ~1MiB triggers EPIPE on the write. The handler must
// swallow it: `done` resolves normally with the child's real exit.
// The child exits without reading, so closing a stdin pipe holding ~1 MiB triggers EPIPE.
// The handler swallows that write error and `done` reports the child's real exit.
const big = 'x'.repeat(1024 * 1024)
const result = await runBash(spec('exit 7', { stdin: big })).done
expect(result.exitCode).toBe(7)
@@ -315,19 +310,11 @@ describe('OutputCollector', () => {
expect(third.spillPath).toBeDefined()
})
it('tracks totalBytes across drops', () => {
const collector = new OutputCollector(4, 'test', spillDir)
collector.push(Buffer.from('aaaa'))
collector.push(Buffer.from('bbbb'))
expect(collector.totalBytes).toBe(8)
expect(collector.finalize().text).toBe('bbbb')
})
it('contains close failures and drops the spill path', () => {
const collector = new OutputCollector(4, 'closefail', spillDir)
collector.push(Buffer.from('aaaa'))
collector.push(Buffer.from('bbbb'))
expect(collector.snapshot().spillPath).toBeDefined()
expect(collector.readFrom(0).spillPath).toBeDefined()
failNextClose.value = true
let out: ReturnType<typeof collector.finalize>
@@ -375,7 +362,7 @@ describe('abort edge cases', () => {
})
})
describe('review fixes: env scrubbing and spill hardening', () => {
describe('environment and spill-file hardening', () => {
it('scrubs credential-shaped env vars from child processes', async () => {
process.env.DSH_TEST_API_KEY = 'super-secret'
process.env.DSH_TEST_TOKEN = 'also-secret'