feat(tools): require cancellation signal on every invocation
This commit is contained in:
@@ -162,7 +162,7 @@ export async function runRipgrep(
|
||||
command,
|
||||
stdoutMaxBytes: rawOutputMaxBytes,
|
||||
...cwd !== undefined ? { workdir: cwd } : {},
|
||||
...exec.signal ? { signal: exec.signal } : {},
|
||||
signal: exec.signal,
|
||||
})
|
||||
let result: BashRunResult
|
||||
try {
|
||||
|
||||
@@ -16,10 +16,12 @@ import { join } from 'node:path'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import ToolRegistry, { TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
|
||||
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
|
||||
import * as ToolFsSearch from '@deepseek-ai/dsh-tool-fs-search'
|
||||
|
||||
const testToolSignal = new AbortController().signal
|
||||
|
||||
const hasRg = spawnSync('rg', ['--version'], { encoding: 'utf8' }).status === 0
|
||||
|
||||
let dir: string
|
||||
@@ -28,6 +30,7 @@ let ctx: Context
|
||||
let callCounter = 0
|
||||
function call(name: string, args: unknown, agentObj?: object) {
|
||||
return ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId(`it-${++callCounter}`),
|
||||
name,
|
||||
arguments: args,
|
||||
@@ -165,8 +168,8 @@ describe.skipIf(!hasRg)('search tools over the real bash executor + real rg', ()
|
||||
})
|
||||
})
|
||||
|
||||
describe('bash-start infrastructure failures stay in the SEARCH_* taxonomy', () => {
|
||||
it('a pre-aborted exec.signal (real executor rejects before spawn) is SEARCH_ABORTED', async () => {
|
||||
describe('pre-dispatch cancellation and bash-start failures', () => {
|
||||
it('a pre-aborted registry call is ABORTED_BEFORE_DISPATCH', async () => {
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
const result = await ctx.tools.execute({
|
||||
@@ -176,7 +179,7 @@ describe.skipIf(!hasRg)('search tools over the real bash executor + real rg', ()
|
||||
signal: controller.signal,
|
||||
})
|
||||
expect(result.isError).toBe(true)
|
||||
expect(result.error).toMatchObject({ name: 'SearchError', code: 'SEARCH_ABORTED' })
|
||||
expect(result.error).toMatchObject({ name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH })
|
||||
})
|
||||
|
||||
it('an unusable session cwd (spawn failure) is SEARCH_FAILED', async () => {
|
||||
|
||||
@@ -14,7 +14,7 @@ import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SystemPrompt, { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import ToolRegistry, { TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
|
||||
import { BashExecutor } from '@deepseek-ai/dsh-bash'
|
||||
import type { BashExecRequest, BashExecSpec, BashProcess, BashRunResult } from '@deepseek-ai/dsh-bash'
|
||||
import { SpillLocator, SpillStore } from '@deepseek-ai/dsh-spill'
|
||||
@@ -31,6 +31,8 @@ import {
|
||||
toWorkdirRelative,
|
||||
} from '@deepseek-ai/dsh-tool-fs-search'
|
||||
|
||||
const testToolSignal = new AbortController().signal
|
||||
|
||||
/** A successful run result over the given stdout; overrides script the failure shapes. */
|
||||
function runResult(stdout: string, overrides?: Partial<BashRunResult>): BashRunResult {
|
||||
return {
|
||||
@@ -55,6 +57,7 @@ class FakeBash extends BashExecutor {
|
||||
requests: BashExecRequest[] = []
|
||||
specs: BashExecSpec[] = []
|
||||
startCalls = 0
|
||||
forwardSignal = true
|
||||
handler: (spec: BashExecSpec) => BashRunResult = () => runResult('')
|
||||
|
||||
override resolve(request: BashExecRequest): BashExecSpec {
|
||||
@@ -64,7 +67,7 @@ class FakeBash extends BashExecutor {
|
||||
workdir: request.workdir ?? '/work',
|
||||
timeoutMs: request.timeoutMs ?? 60_000,
|
||||
stdoutMaxBytes: request.stdoutMaxBytes ?? 64_000,
|
||||
signal: request.signal,
|
||||
...this.forwardSignal ? { signal: request.signal } : {},
|
||||
sandboxMode: request.sandboxMode,
|
||||
}
|
||||
}
|
||||
@@ -117,6 +120,7 @@ const agent = (cwd?: string) => ({ session: { header: { id: 'session-1', ...cwd
|
||||
let callCounter = 0
|
||||
function call(ctx: Context, name: string, args: unknown, options: { agent?: object; signal?: AbortSignal } = {}) {
|
||||
return ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId(`call-${++callCounter}`),
|
||||
name,
|
||||
arguments: args,
|
||||
@@ -248,16 +252,13 @@ describe('workdir derivation and signal forwarding', () => {
|
||||
expect(bash.requests[1]).not.toHaveProperty('workdir')
|
||||
})
|
||||
|
||||
it('forwards exec.signal into the bash spec (the abort reaches the backend)', async () => {
|
||||
it('forwards exec.signal into the bash spec', async () => {
|
||||
const { ctx, bash } = await setup()
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
bash.handler = spec => runResult('', { aborted: spec.signal?.aborted === true })
|
||||
bash.handler = () => runResult('')
|
||||
const result = await call(ctx, 'grep', { pattern: 'x' }, { signal: controller.signal })
|
||||
expect(bash.specs[0]?.signal).toBe(controller.signal)
|
||||
expect(result.isError).toBe(true)
|
||||
expect(result.error).toMatchObject({ name: 'SearchError', code: 'SEARCH_ABORTED' })
|
||||
expect(text(result)).toContain('aborted')
|
||||
expect(result.isError).toBe(false)
|
||||
})
|
||||
|
||||
it('reports the bash executor timeout as SEARCH_ABORTED with the budget', async () => {
|
||||
@@ -269,20 +270,46 @@ describe('workdir derivation and signal forwarding', () => {
|
||||
expect(text(result)).toContain('timed out after 1234ms')
|
||||
})
|
||||
|
||||
it('translates a run() rejection under a pre-aborted signal into SEARCH_ABORTED', async () => {
|
||||
// The seam contract: run() REJECTS for a pre-aborted signal (it never
|
||||
// spawns). The plain rejection must not escape the SEARCH_* taxonomy.
|
||||
it('skips a pre-aborted registry call before run()', async () => {
|
||||
const { ctx, bash } = await setup()
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
bash.handler = () => { throw new Error('aborted before spawn') }
|
||||
const result = await call(ctx, 'grep', { pattern: 'x' }, { signal: controller.signal })
|
||||
expect(result.isError).toBe(true)
|
||||
expect(result.error).toMatchObject({ name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH })
|
||||
expect(bash.specs).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('translates a run() rejection after the forwarded signal aborts', async () => {
|
||||
const { ctx, bash } = await setup()
|
||||
const controller = new AbortController()
|
||||
bash.handler = () => {
|
||||
controller.abort('cancel search')
|
||||
throw new Error('executor stopped on abort')
|
||||
}
|
||||
|
||||
const result = await call(ctx, 'grep', { pattern: 'x' }, { signal: controller.signal })
|
||||
|
||||
expect(result.isError).toBe(true)
|
||||
expect(result.error).toMatchObject({ name: 'SearchError', code: 'SEARCH_ABORTED' })
|
||||
expect(text(result)).toContain('aborted before completion')
|
||||
})
|
||||
|
||||
it('translates an aborted executor result after dispatch starts', async () => {
|
||||
const { ctx, bash } = await setup()
|
||||
bash.handler = () => runResult('', { aborted: true, exitCode: null })
|
||||
|
||||
const result = await call(ctx, 'glob', { pattern: '*' })
|
||||
|
||||
expect(result.isError).toBe(true)
|
||||
expect(result.error).toMatchObject({ name: 'SearchError', code: 'SEARCH_ABORTED' })
|
||||
expect(text(result)).toContain('aborted before completion')
|
||||
})
|
||||
|
||||
it('translates a run() rejection without an abort (unusable workdir) into SEARCH_FAILED', async () => {
|
||||
const { ctx, bash } = await setup()
|
||||
bash.forwardSignal = false
|
||||
bash.handler = () => { throw new Error('spawn bash ENOENT') }
|
||||
const result = await call(ctx, 'glob', { pattern: '*' })
|
||||
expect(result.isError).toBe(true)
|
||||
|
||||
@@ -28,6 +28,6 @@ export function sessionResolveOptions(exec: ToolExecution): { cwd?: string; sign
|
||||
const cwd = sessionCwd(exec)
|
||||
return {
|
||||
...cwd !== undefined ? { cwd } : {},
|
||||
...exec.signal !== undefined ? { signal: exec.signal } : {},
|
||||
signal: exec.signal,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,11 +12,13 @@ import { join } from 'node:path'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import ToolRegistry, { TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
|
||||
import { LocalFileSystem } from '@deepseek-ai/dsh-fs-local'
|
||||
import * as FsPolicy from '@deepseek-ai/dsh-fs-policy'
|
||||
import * as ToolFs from '@deepseek-ai/dsh-tool-fs'
|
||||
|
||||
const testToolSignal = new AbortController().signal
|
||||
|
||||
let dir: string
|
||||
let ctx: Context
|
||||
let fiber: Awaited<ReturnType<Context['plugin']>>
|
||||
@@ -26,6 +28,7 @@ const session = { header: {} }
|
||||
let callCounter = 0
|
||||
function call(name: string, args: unknown) {
|
||||
return ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId(`call-${++callCounter}`),
|
||||
name,
|
||||
arguments: args,
|
||||
@@ -299,6 +302,7 @@ describe('per-session cwd', () => {
|
||||
|
||||
const callIn = (sessionObj: object, name: string, args: unknown) =>
|
||||
ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId(`call-${++callCounter}`),
|
||||
name,
|
||||
arguments: args,
|
||||
@@ -344,26 +348,25 @@ describe('signal, concurrency, and the fs/observed contract', () => {
|
||||
const callSig = (signal: AbortSignal, name: string, args: unknown) =>
|
||||
ctx.tools.execute({ callId: CallId(`c-${++callCounter}`), name, arguments: args, agent: { session } as never, signal })
|
||||
const callOwned = (name: string, args: unknown) =>
|
||||
ctx.tools.execute({ callId: CallId(`c-${++callCounter}`), name, arguments: args, agent: { session } as never })
|
||||
ctx.tools.execute({ signal: testToolSignal, callId: CallId(`c-${++callCounter}`), name, arguments: args, agent: { session } as never })
|
||||
|
||||
it('a pre-aborted signal makes read/write/edit return isError FS_ABORTED', async () => {
|
||||
it('a pre-aborted registry call skips read/write/edit with ABORTED_BEFORE_DISPATCH', async () => {
|
||||
await writeFile(join(dir, 'a.txt'), 'hello')
|
||||
const read = await callSig(AbortSignal.abort(), 'read', { file_path: 'a.txt' })
|
||||
expect(read.isError).toBe(true)
|
||||
expect(read.error).toMatchObject({ code: 'FS_ABORTED' })
|
||||
expect(read.error).toMatchObject({ name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH })
|
||||
|
||||
const write = await callSig(AbortSignal.abort(), 'write', { file_path: 'new.txt', content: 'x' })
|
||||
expect(write.isError).toBe(true)
|
||||
expect(write.error).toMatchObject({ code: 'FS_ABORTED' })
|
||||
expect(write.error).toMatchObject({ name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH })
|
||||
await expect(readFile(join(dir, 'new.txt'), 'utf8')).rejects.toMatchObject({ code: 'ENOENT' })
|
||||
|
||||
// Read first (un-aborted, SAME session owner) so the edit clears the
|
||||
// observation gate; then the aborted edit fails on the signal, not on
|
||||
// FS_NOT_OBSERVED.
|
||||
// observation gate; then the registry skips the aborted edit before its body.
|
||||
expect((await callOwned('read', { file_path: 'a.txt' })).isError).toBe(false)
|
||||
const edit = await callSig(AbortSignal.abort(), 'edit', { file_path: 'a.txt', old_string: 'hello', new_string: 'bye' })
|
||||
expect(edit.isError).toBe(true)
|
||||
expect(edit.error).toMatchObject({ code: 'FS_ABORTED' })
|
||||
expect(edit.error).toMatchObject({ name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH })
|
||||
expect(await readFile(join(dir, 'a.txt'), 'utf8')).toBe('hello') // unchanged
|
||||
})
|
||||
|
||||
|
||||
@@ -25,6 +25,8 @@ import { STREAM_MIN_SIZE } from '../src/read.ts'
|
||||
import { formatReadOutput } from '../src/read-render.ts'
|
||||
import type { FileReadOutcome } from '../src/read-render.ts'
|
||||
|
||||
const testToolSignal = new AbortController().signal
|
||||
|
||||
/** An in-memory fake provider; a test can arm a rejection on any primitive. */
|
||||
class FakeFs extends FileSystem {
|
||||
files = new Map<string, string>()
|
||||
@@ -91,6 +93,7 @@ async function setup() {
|
||||
let callCounter = 0
|
||||
function call(ctx: Context, name: string, args: unknown, agent?: object) {
|
||||
return ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId(`call-${++callCounter}`),
|
||||
name,
|
||||
arguments: args,
|
||||
@@ -110,11 +113,11 @@ describe('registration', () => {
|
||||
|
||||
it('declares read parallel-safe while write/edit remain exclusive', async () => {
|
||||
const { ctx } = await setup()
|
||||
expect(ctx.tools.executionMode({ callId: CallId('read-safe'), name: 'read', arguments: { file_path: 'a.txt' } }))
|
||||
expect(ctx.tools.executionMode({ signal: testToolSignal, callId: CallId('read-safe'), name: 'read', arguments: { file_path: 'a.txt' } }))
|
||||
.toEqual({ kind: 'parallel' })
|
||||
expect(ctx.tools.executionMode({ callId: CallId('write-exclusive'), name: 'write', arguments: { file_path: 'a.txt', content: 'x' } }))
|
||||
expect(ctx.tools.executionMode({ signal: testToolSignal, callId: CallId('write-exclusive'), name: 'write', arguments: { file_path: 'a.txt', content: 'x' } }))
|
||||
.toEqual({ kind: 'exclusive' })
|
||||
expect(ctx.tools.executionMode({ callId: CallId('edit-exclusive'), name: 'edit', arguments: { file_path: 'a.txt', old_string: 'x', new_string: 'y' } }))
|
||||
expect(ctx.tools.executionMode({ signal: testToolSignal, callId: CallId('edit-exclusive'), name: 'edit', arguments: { file_path: 'a.txt', old_string: 'x', new_string: 'y' } }))
|
||||
.toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user