import { PassThrough } from 'node:stream' import { Context } from '@deepseek-ai/cordis' import Loader from '@deepseek-ai/cordis-plugin-loader' import { describe, expect, it, vi } from 'vitest' import type { Agent } from '@deepseek-ai/dsh-agent' import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants' import type { ContentBlock } from '@deepseek-ai/dsh-llm' import SubagentRuntime, { type SubagentStopReason } from '@deepseek-ai/dsh-subagent' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import type { SubprocessHandle, SubprocessOutcome, } from '@deepseek-ai/dsh-subprocess' import LocalSubprocessRuntime from '@deepseek-ai/dsh-subprocess-local' import * as cursor from '../src/index.ts' import * as invariant from '../src/invariant.ts' import { assertDirectlySpawnable, cursorAgentArgv, DEFAULT_DISPOSE_GRACE_MS, disposeCursorChild, startCursorRun, textTask, type CursorRunSpec, } from '../src/run.ts' import { CursorStreamWire } from '../src/wire.ts' type JsonObject = Record const EXECUTABLE = '/opt/cursor/cursor-agent' const fakeParent = { id: 'parent', session: { header: { cwd: process.cwd() } }, } as unknown as Agent function request( prompt: ContentBlock[] = [{ type: 'text', text: 'do the task' }], signal = new AbortController().signal, ) { return { prompt, parent: fakeParent, signal } } async function nextTask(): Promise { await new Promise((resolve) => { setImmediate(resolve) }) } function initEvent(overrides: JsonObject = {}): JsonObject { return { type: 'system', subtype: 'init', apiKeySource: 'env', cwd: process.cwd(), session_id: 'chat-1', model: 'cursor-model', permissionMode: 'default', ...overrides, } } function assistantEvent(content: unknown): JsonObject { return { type: 'assistant', message: { role: 'assistant', content }, session_id: 'chat-1', } } function resultEvent(overrides: JsonObject = {}): JsonObject { return { type: 'result', subtype: 'success', duration_ms: 12, duration_api_ms: 10, is_error: false, result: 'the final answer', session_id: 'chat-1', ...overrides, } } /** Writes stream-json events the way the real CLI writes its stdout. */ class StreamPeer { constructor(private readonly output: PassThrough) {} send(...events: readonly JsonObject[]): void { this.output.write(`${events.map(event => JSON.stringify(event)).join('\n')}\n`) } raw(text: string): void { this.output.write(text) } } interface FakeChildOptions { readonly pid?: number readonly exitOnTerminate?: boolean readonly doneError?: Error } interface FakeChild { readonly handle: SubprocessHandle readonly peer: StreamPeer readonly fromChild: PassThrough readonly toChild: PassThrough readonly settle: (outcome?: SubprocessOutcome) => void readonly fail: (error: Error) => void readonly terminate: () => void readonly waitForExit: (signal?: AbortSignal) => Promise } function fakeChild(options: FakeChildOptions = {}): FakeChild { const fromChild = new PassThrough() const toChild = new PassThrough() const peer = new StreamPeer(fromChild) let exited = false let resolveDone!: (outcome: SubprocessOutcome) => void let rejectDone!: (error: Error) => void const done = new Promise((resolve, reject) => { resolveDone = resolve rejectDone = reject }) const settle = ( outcome: SubprocessOutcome = { exitCode: 0, signal: null }, ): void => { if (exited) return exited = true resolveDone(outcome) } const fail = (error: Error): void => { if (exited) return exited = true rejectDone(error) } if (options.doneError !== undefined) fail(options.doneError) const terminate = vi.fn(() => { if (options.exitOnTerminate !== false) settle() }) const waitForExit = vi.fn(async (signal?: AbortSignal) => { if (exited) return true if (signal === undefined) { await done.catch(() => {}) return true } return await new Promise((resolve) => { const onAbort = (): void => { resolve(false) } signal.addEventListener('abort', onAbort, { once: true }) void done.then( () => { signal.removeEventListener('abort', onAbort) resolve(true) }, () => { signal.removeEventListener('abort', onAbort) resolve(true) }, ) }) }) const handle: SubprocessHandle = { pid: options.pid ?? 4321, stdin: toChild, stdout: fromChild, stderr: undefined, collected: {}, done, terminate, waitForExit, } return { handle, peer, fromChild, toChild, settle, fail, terminate, waitForExit, } } function runSpec( child: FakeChild, overrides: Partial = {}, ): CursorRunSpec { return { cwd: process.cwd(), executable: EXECUTABLE, env: {}, disposeGraceMs: DEFAULT_DISPOSE_GRACE_MS, force: false, trust: false, spawn: () => child.handle, ...overrides, } } async function publishRun( child = fakeChild(), signal = new AbortController().signal, specOverrides: Partial = {}, ) { const starting = startCursorRun(request(undefined, signal), runSpec(child, specOverrides)) await nextTask() child.peer.send(initEvent()) return { child, run: await starting } } function startedWire(): { readonly child: FakeChild; readonly wire: CursorStreamWire } { const child = fakeChild() const wire = new CursorStreamWire(child.handle.stdout!) wire.start() return { child, wire } } describe('task admission and command construction', () => { it('accepts one or more text blocks and rejects empty, non-text, or option-shaped tasks', () => { expect(textTask([{ type: 'text', text: 'a' }, { type: 'text', text: 'b' }])).toBe('ab') expect(() => textTask([])).toThrow('must contain only text blocks') expect(() => textTask([{ type: 'reasoning', text: 'x' }])) .toThrow('must contain only text blocks') expect(() => textTask([{ type: 'text', text: ' ' }])).toThrow('must not be empty') expect(() => textTask([{ type: 'text', text: '--force me' }])) .toThrow('must not begin with "-"') }) it('refuses a Windows batch shim so task text never reaches a shell', () => { expect(assertDirectlySpawnable('C:\\bin\\cursor-agent.exe', 'win32')) .toBe('C:\\bin\\cursor-agent.exe') expect(assertDirectlySpawnable('/usr/bin/cursor-agent', 'linux')) .toBe('/usr/bin/cursor-agent') // A shim is only unusable where it needs cmd.exe. expect(assertDirectlySpawnable('/usr/bin/cursor-agent.cmd', 'linux')) .toBe('/usr/bin/cursor-agent.cmd') for (const shim of ['C:\\bin\\cursor-agent.cmd', 'C:\\bin\\cursor-agent.BAT']) { expect(() => assertDirectlySpawnable(shim, 'win32')).toThrow('is a batch shim') } // The omitted platform reads the host, so the expectation follows it. const underHostPlatform = (): string => assertDirectlySpawnable('C:\\bin\\cursor-agent.cmd') if (process.platform === 'win32') { expect(underHostPlatform).toThrow('is a batch shim') } else { expect(underHostPlatform()).toBe('C:\\bin\\cursor-agent.cmd') } }) it('builds the fixed print-mode argv and adds only selected permissions', () => { const base = { executable: EXECUTABLE, cwd: '/work', force: false, trust: false } expect(cursorAgentArgv(base, 'ship it')).toEqual([ EXECUTABLE, '--print', '--output-format', 'stream-json', '--workspace', '/work', 'ship it', ]) expect(cursorAgentArgv({ ...base, force: true, trust: true }, 'ship it')).toEqual([ EXECUTABLE, '--print', '--output-format', 'stream-json', '--workspace', '/work', '--force', '--trust', 'ship it', ]) }) }) describe('package contracts', () => { it('registers one fixed descriptor, validates config, and unregisters on HMR', async () => { const ctx = new Context() await ctx.plugin(SubagentRuntime) await ctx.plugin(LocalSubprocessRuntime) const fiber = await ctx.plugin(cursor, {}) expect(ctx.subagents.getProvider('cursor')).toMatchObject({ name: 'cursor', capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false, }, inheritsParentContext: false, }) expect(ctx.subagents.list()).toEqual(['cursor']) await fiber.dispose() expect(ctx.subagents.list()).toEqual([]) for (const disposeGraceMs of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) { await expect(ctx.plugin(cursor, { disposeGraceMs })) .rejects.toThrow('disposeGraceMs must be a positive finite number') } await expect(ctx.plugin(cursor, { disposeGraceMs: MAX_TIMER_DELAY_MS + 1 })) .rejects.toThrow(`disposeGraceMs must be no greater than ${MAX_TIMER_DELAY_MS}`) await ctx.fiber.dispose() }) it('requires a parent session cwd before resolving or spawning anything', async () => { const ctx = new Context() await ctx.plugin(SubagentRuntime) await ctx.plugin(LocalSubprocessRuntime) const resolveExecutable = vi.spyOn(ctx.subprocess, 'resolveExecutable') const spawn = vi.spyOn(ctx.subprocess, 'spawn') await ctx.plugin(cursor, {}) await expect(ctx.subagents.start('cursor', { prompt: [{ type: 'text', text: 'task' }], parent: { id: 'parent-without-cwd', session: { header: {} }, } as unknown as Agent, signal: new AbortController().signal, })).rejects.toThrow( 'subagent-cursor: no working directory for the child — delegate from a parent session that has one', ) expect(resolveExecutable).not.toHaveBeenCalled() expect(spawn).not.toHaveBeenCalled() await ctx.fiber.dispose() }) it('keeps the namespace export shape and package-owned empty invariant', async () => { expect('default' in cursor).toBe(false) expect(cursor.name).toBe('subagent-cursor') expect(cursor.inject).toEqual(['subagents', 'subprocess']) const loader = Object.create(Loader.prototype) as Loader expect(loader.unwrapExports(cursor)).toBe(cursor) const dispose = vi.fn() const register = vi.fn(( _packageName: string, _installer: InvariantInstaller, ) => dispose) const ctx = { invariants: { register } } as unknown as Context await expect(invariant.apply(ctx)).resolves.toBe(dispose) expect(register).toHaveBeenCalledWith( '@deepseek-ai/dsh-subagent-cursor', expect.any(Function), ) const install = register.mock.calls[0]![1] await install(new Context(), (message) => { throw new Error(message) }) expect(invariant.name).toBe('subagent-cursor-invariant') expect(invariant.inject).toEqual(['invariants']) }) }) describe('CursorStreamWire', () => { it('gates on init, keeps the terminal answer, and ignores unrelated events', async () => { const { child, wire } = startedWire() wire.start() child.peer.send(initEvent()) await expect(wire.ready()).resolves.toEqual({ sessionId: 'chat-1', model: 'cursor-model', }) child.peer.send( { type: 'user', message: { role: 'user', content: [{ type: 'text', text: 'task' }] } }, { type: 'tool_call', subtype: 'started', call_id: 'c1', tool_call: {} }, { type: 'system', subtype: 'usage', tokens: 12 }, assistantEvent([{ type: 'text', text: 'thinking out loud' }]), { type: 'newer_cli_event', payload: 1 }, ) await nextTask() expect(wire.collectOutput()).toEqual([{ type: 'text', text: 'thinking out loud' }]) child.peer.send(resultEvent()) await expect(wire.awaitResult()).resolves.toEqual({ output: [{ type: 'text', text: 'the final answer' }], stopReason: 'completed', }) }) it('reads events split across chunks and skips blank lines', async () => { const { child, wire } = startedWire() const frame = JSON.stringify(initEvent()) child.peer.raw(`\n \n${frame.slice(0, 10)}`) await nextTask() child.peer.raw(`${frame.slice(10)}\n`) await expect(wire.ready()).resolves.toMatchObject({ sessionId: 'chat-1' }) expect(wire.collectOutput()).toEqual([]) }) it('reports an absent model rather than inventing one', async () => { const { child, wire } = startedWire() child.peer.send(initEvent({ model: 42 })) await expect(wire.ready()).resolves.toEqual({ sessionId: 'chat-1', model: undefined, }) }) it('keeps the last non-empty assistant message and drops non-text blocks', async () => { const { child, wire } = startedWire() child.peer.send( initEvent(), assistantEvent([{ type: 'text', text: 'first' }]), assistantEvent([{ type: 'image', source: {} }]), assistantEvent([]), assistantEvent([{ type: 'text', text: 'second' }, { type: 'image', source: {} }]), ) await nextTask() expect(wire.collectOutput()).toEqual([{ type: 'text', text: 'second' }]) }) it('fails closed on every unusable terminal result', async () => { for (const [overrides, detail] of [ [{ subtype: 'error' }, 'terminal result subtype error'], [{ subtype: 7 }, 'terminal result subtype unknown'], [{ is_error: true }, 'success result was marked as an error'], [{ result: 12 }, 'success result was marked as an error'], [{ result: ' ' }, 'success result was marked as an error'], ] as const) { const { child, wire } = startedWire() child.peer.send(initEvent(), resultEvent(overrides)) await expect(wire.awaitResult()).rejects.toThrow(detail) } }) it('refuses a result that arrives without an announced session', async () => { const { child, wire } = startedWire() child.peer.send(resultEvent()) await expect(wire.ready()).rejects.toThrow('without announcing a session') await expect(wire.awaitResult()).rejects.toThrow('without announcing a session') }) it('treats malformed stdout as a protocol failure', async () => { for (const [line, detail] of [ ['not json at all', 'a stdout line that is not JSON'], ['[1,2]', 'an invalid event'], ['null', 'an invalid event'], [JSON.stringify({ type: 'system', subtype: 'init', session_id: '' }), 'an invalid init session id'], [JSON.stringify({ type: 'assistant', message: 'text' }), 'an invalid assistant message'], [JSON.stringify(assistantEvent('not an array')), 'an invalid assistant message content'], ] as const) { const { child, wire } = startedWire() child.peer.raw(`${line}\n`) await expect(wire.awaitResult()).rejects.toThrow(detail) } }) it('fails pending gates on stream error, end of stream, and close', async () => { const broken = startedWire() broken.child.fromChild.emit('error', new Error('stdout broke')) await expect(broken.wire.awaitResult()).rejects.toThrow('stdout broke') const ended = startedWire() ended.child.fromChild.end() await expect(ended.wire.awaitResult()).rejects.toThrow('ended without a terminal result') const closed = startedWire() closed.wire.close() closed.wire.close() await expect(closed.wire.ready()).rejects.toThrow('event stream closed') // A completed run keeps its result across teardown and end of stream. const done = startedWire() done.child.peer.send(initEvent(), resultEvent()) await expect(done.wire.awaitResult()).resolves.toMatchObject({ stopReason: 'completed' }) done.wire.close() done.child.fromChild.end() await expect(done.wire.awaitResult()).resolves.toMatchObject({ stopReason: 'completed' }) }) it('is safe to close before it is started', async () => { const child = fakeChild() const wire = new CursorStreamWire(child.handle.stdout!) wire.close() await expect(wire.ready()).rejects.toThrow('event stream closed') }) }) describe('run lifecycle and quiescence', () => { it('spawns the fixed command, publishes after init, and disposes once', async () => { const child = fakeChild() const spawn = vi.fn(() => child.handle) const starting = startCursorRun( request(), runSpec(child, { spawn, force: true, trust: true, disposeGraceMs: 40, env: { CURSOR_API_KEY: 'fake' } }), ) await nextTask() expect(spawn).toHaveBeenCalledWith(expect.objectContaining({ argv: [ EXECUTABLE, '--print', '--output-format', 'stream-json', '--workspace', process.cwd(), '--force', '--trust', 'do the task', ], cwd: process.cwd(), graceMs: 40, env: { CURSOR_API_KEY: 'fake' }, })) child.peer.send(initEvent()) const run = await starting expect(run.id).toMatch(/^[0-9a-f-]{36}$/) expect(run.localAgent).toBeUndefined() child.peer.send(resultEvent()) await expect(run.result).resolves.toEqual({ output: [{ type: 'text', text: 'the final answer' }], stopReason: 'completed', }) const first = run.dispose() expect(run.dispose()).toBe(first) await first expect(child.terminate).toHaveBeenCalledTimes(1) }) it('closes stdin so an unattended prompt cannot stall the child', async () => { const { child, run } = await publishRun() expect(child.toChild.writableEnded).toBe(true) child.peer.send(resultEvent()) await run.result await run.dispose() }) it('settles a cancelled run as aborted with the output collected so far', async () => { const controller = new AbortController() const child = fakeChild({ exitOnTerminate: false }) const { run } = await publishRun(child, controller.signal) child.peer.send(assistantEvent([{ type: 'text', text: 'partial work' }])) await nextTask() controller.abort(new Error('parent stopped waiting')) await expect(run.result).resolves.toEqual({ output: [{ type: 'text', text: 'partial work' }], stopReason: 'aborted', }) child.settle({ exitCode: null, signal: 'SIGTERM' }) await run.dispose() }) it('flattens a child exit and a protocol failure after publication', async () => { const exited = await publishRun() exited.child.settle({ exitCode: 2, signal: null }) await expect(exited.run.result).resolves.toEqual({ output: [], stopReason: 'error', }) await exited.run.dispose() const onError = vi.fn<(error: Error, stopReason: SubagentStopReason) => void>() const malformed = await publishRun(fakeChild(), new AbortController().signal, { onError }) malformed.child.peer.raw('garbage\n') await expect(malformed.run.result).resolves.toMatchObject({ stopReason: 'error' }) expect(onError).toHaveBeenCalledTimes(1) const [reported, reportedReason] = onError.mock.calls[0]! expect(reported.message).toContain('not JSON') expect(reportedReason).toBe('error') await malformed.run.dispose() }) it('rejects before spawn when the request is already aborted', async () => { const controller = new AbortController() controller.abort() const child = fakeChild() const spawn = vi.fn(() => child.handle) await expect(startCursorRun(request(undefined, controller.signal), runSpec(child, { spawn }))) .rejects.toThrow('aborted before cursor-agent startup') expect(spawn).not.toHaveBeenCalled() }) it('rolls the child back when startup fails or is aborted before publication', async () => { const failed = fakeChild() const failing = startCursorRun(request(), runSpec(failed)) await nextTask() failed.peer.raw('not json\n') await expect(failing).rejects.toThrow('not JSON') expect(failed.terminate).toHaveBeenCalledTimes(1) const controller = new AbortController() const aborted = fakeChild() const aborting = startCursorRun(request(undefined, controller.signal), runSpec(aborted)) await nextTask() controller.abort() await expect(aborting).rejects.toThrow('aborted before run publication') expect(aborted.terminate).toHaveBeenCalledTimes(1) const exited = fakeChild() const exiting = startCursorRun(request(), runSpec(exited)) await nextTask() exited.settle({ exitCode: 3, signal: null }) await expect(exiting).rejects.toThrow('exited before the run settled') // A failed spawn reports pid -1, so rollback has no tree to signal. const broken = fakeChild({ pid: -1, doneError: new Error('spawn observer failed') }) await expect(startCursorRun(request(), runSpec(broken))) .rejects.toThrow('spawn observer failed') expect(broken.terminate).not.toHaveBeenCalled() }) it('reports a startup failure whose cleanup also failed as an aggregate', async () => { const child = fakeChild() child.handle.waitForExit = vi.fn(() => Promise.reject(new Error('tree never exited'))) const starting = startCursorRun(request(), runSpec(child)) await nextTask() child.peer.raw('not json\n') const failure = await starting.catch((error: unknown) => error) expect(failure).toBeInstanceOf(AggregateError) expect((failure as AggregateError).message) .toContain('startup failed and cursor-agent cleanup also failed') }) it('keeps overlapping runs isolated', async () => { const first = await publishRun() const second = await publishRun() first.child.peer.send(resultEvent({ result: 'first answer' })) second.child.peer.send(resultEvent({ subtype: 'error' })) await expect(first.run.result).resolves.toMatchObject({ output: [{ type: 'text', text: 'first answer' }], stopReason: 'completed', }) await expect(second.run.result).resolves.toMatchObject({ stopReason: 'error' }) await first.run.dispose() await second.run.dispose() }) it('uses the registered provider config, the resolved executable, and logs flattened errors', async () => { const ctx = new Context() await ctx.plugin(SubagentRuntime) await ctx.plugin(LocalSubprocessRuntime) const child = fakeChild() const resolveExecutable = vi.spyOn(ctx.subprocess, 'resolveExecutable') .mockResolvedValue(EXECUTABLE) const spawn = vi.spyOn(ctx.subprocess, 'spawn').mockReturnValue(child.handle) const warnings: string[] = [] ctx.logger.warn = ((message: unknown) => { warnings.push(String(message)) }) as typeof ctx.logger.warn await ctx.plugin(cursor, { env: { CURSOR_API_KEY: 'fake' }, disposeGraceMs: 25, force: true, }) const starting = ctx.subagents.start('cursor', { prompt: [{ type: 'text', text: 'task' }], parent: fakeParent, signal: new AbortController().signal, }) await nextTask() child.peer.send(initEvent()) const run = await starting child.settle({ exitCode: 1, signal: null }) await expect(run.result).resolves.toMatchObject({ stopReason: 'error' }) expect(resolveExecutable).toHaveBeenCalledWith( 'cursor-agent', { CURSOR_API_KEY: 'fake' }, expect.any(AbortSignal), ) expect(spawn).toHaveBeenCalledTimes(1) const spawned = spawn.mock.calls[0]![0] expect(spawned.argv).toContain('--force') expect(spawned.env).toEqual({ CURSOR_API_KEY: 'fake' }) expect(spawned.graceMs).toBe(25) expect(spawned.cwd).toBe(process.cwd()) expect(warnings).toEqual([ expect.stringContaining('subagent-cursor: child run failed (error):'), ]) await run.dispose().catch(() => {}) await ctx.fiber.dispose() }) }) describe('disposeCursorChild', () => { it('closes the stream, terminates, and waits for the managed tree', async () => { const child = fakeChild() const wire = new CursorStreamWire(child.handle.stdout!) wire.start() await disposeCursorChild(wire, child.handle) expect(child.terminate).toHaveBeenCalledTimes(1) expect(child.waitForExit).toHaveBeenCalledTimes(1) }) it('does not finish disposal before the managed tree exits', async () => { const child = fakeChild({ exitOnTerminate: false }) const wire = new CursorStreamWire(child.handle.stdout!) let settled = false const disposing = disposeCursorChild(wire, child.handle).then(() => { settled = true }) await nextTask() expect(settled).toBe(false) child.settle() await disposing expect(settled).toBe(true) }) it('skips signalling a failed spawn and contains its observer rejection', async () => { const child = fakeChild({ pid: -1, doneError: new Error('spawn failed') }) const wire = new CursorStreamWire(child.handle.stdout!) await expect(disposeCursorChild(wire, child.handle)).resolves.toBeUndefined() expect(child.terminate).not.toHaveBeenCalled() }) it('reports a direct-child observer failure from a live tree', async () => { const child = fakeChild({ exitOnTerminate: false }) const wire = new CursorStreamWire(child.handle.stdout!) const disposing = disposeCursorChild(wire, child.handle) child.fail(new Error('observer failed')) await expect(disposing).rejects.toThrow('observer failed') }) })