import { Context } from '@deepseek-ai/cordis' import { PtcRuntime } from '@deepseek-ai/dsh-ptc-runtime' import type { PtcBindingFunction, PtcRunRequest, PtcRunResult, PtcRunSpec } from '@deepseek-ai/dsh-ptc-runtime' import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import SessionProjections from '@deepseek-ai/dsh-session-projection' import SandboxPolicy from '@deepseek-ai/dsh-sandbox-policy' import SubagentRuntime from '@deepseek-ai/dsh-subagent' import type { SubagentResult } from '@deepseek-ai/dsh-subagent' import { describe, expect, it, onTestFinished, vi } from 'vitest' import PtcWorkflowEngine from '../src/index.ts' import { fakeParent } from './setup.ts' const completed: PtcRunResult = { logs: [], value: { value: null, stopReason: 'completed', agentsStarted: 0 } } type HostBindings = Record class ControlledRuntime extends PtcRuntime { language = 'typescript' readonly isolation = 'process' execute: (spec: PtcRunSpec) => Promise = () => Promise.resolve(completed) resolve(request: PtcRunRequest): PtcRunSpec { return { ...request, cwd: request.cwd ?? process.cwd(), timeoutMs: request.timeoutMs === undefined ? 120_000 : request.timeoutMs } } run(spec: PtcRunSpec): Promise { return this.execute(spec) } } async function setup(execute?: (bindings: HostBindings, spec: PtcRunSpec) => Promise, language = 'typescript') { const ctx = new Context() onTestFinished(async () => { await ctx.fiber.dispose() }) await ctx.plugin(SessionStore) await ctx.plugin(SessionProjections) await ctx.plugin(SandboxPolicy, { mode: 'read-only' }) await ctx.plugin(SubagentRuntime) ctx.subagents.registerProvider({ name: 'stub', capabilities: { agentOptions: true, outputSchema: true, depthLimit: false, toolFilter: false, persona: false }, inheritsParentContext: false, start: async () => ({ id: SessionId('host-child'), localAgent: undefined, result: Promise.resolve({ output: [], stopReason: 'completed' }), dispose: () => Promise.resolve(), }), }) await ctx.plugin(ControlledRuntime) const runtime = ctx.ptcRuntime as ControlledRuntime runtime.language = language if (execute !== undefined) runtime.execute = spec => execute(spec.bindings[0]!.functions, spec) await ctx.plugin(PtcWorkflowEngine, { provider: 'stub' }) const parent = fakeParent(ctx) const start = () => ctx.workflowEngine.start({ script: 'return null', meta: { name: 'host-test', description: 'workflow callbacks' }, parent, }) return { ctx, runtime, start, parent } } describe('workflow host callback validation', () => { it.each([ ['startChild', null, 'requires an object'], ['startChild', [], 'requires an object'], ['startChild', { prompt: 7 }, 'prompt must be a string'], ['startChild', { prompt: 'p', provider: 7 }, 'provider must be a string'], ['startChild', { prompt: 'p', model: false }, 'model must be a string'], ['progress', {}, 'requires an array of events'], ['progress', [{ type: 'phase', title: null }], 'phase must be a string'], ['progress', [{ type: 'agent-start', info: { seq: 0 } }], 'sequence must be a positive integer'], ['progress', [{ type: 'agent-end', info: { outcome: 'unknown' } }], 'invalid workflow agent outcome'], ['progress', [{ type: 'unknown' }], 'invalid workflow progress event'], ['childResult', { callId: '1' }, 'call id must be an integer'], ['disposeChild', { callId: 99 }, 'child call is not active'], ] as const)('rejects malformed %s callback data %j', async (name, input, message) => { const { start } = await setup(async (bindings) => { await expect(Promise.resolve().then(() => bindings[name]!(input))).rejects.toThrow(message) return completed }) const handle = start() try { expect((await handle.result).stopReason).toBe('completed') } finally { await handle.dispose() } }) it('does not emit duplicate agent-end notifications', async () => { const info = { seq: 1, label: 'child', childId: 'host-child' } const { ctx, start } = await setup(async (bindings) => { await bindings.progress!([{ type: 'agent-start', info }]) await bindings.progress!([{ type: 'agent-end', info: { ...info, outcome: 'failed' } }]) await bindings.progress!([{ type: 'agent-end', info: { ...info, outcome: 'cancelled' } }]) return completed }) const ended = vi.fn() ctx.on('workflow/agent-end', ended) const handle = start() try { expect((await handle.result).stopReason).toBe('completed') expect(ended).toHaveBeenCalledOnce() } finally { await handle.dispose() } }) it('carries Session authority and an explicit unlimited deadline to the runtime', async () => { const { ctx, parent, start } = await setup(async (bindings, spec) => { expect(spec.timeoutMs).toBeNull() expect(spec.cwd).toBe(parent.session.header.cwd) expect(spec.sandboxPolicy).toEqual(ctx.sandboxPolicy.resolve({ session: parent.session })) expect(await bindings.begin!({})).toMatchObject({ body: 'return null', meta: { name: 'host-test' } }) return completed }) const handle = start() try { expect((await handle.result).stopReason).toBe('completed') } finally { await handle.dispose() } }) }) describe('workflow runtime outcomes', () => { it.each([ [null, 'requires an object'], [{ value: null, stopReason: 'unknown', agentsStarted: 0 }, 'invalid workflow stop reason'], [{ value: null, stopReason: 'cancelled', agentsStarted: -1 }, 'invalid workflow agent count'], [{ stopReason: 'completed', agentsStarted: 0 }, 'missing its value'], [{ value: null, stopReason: 'error', agentsStarted: 0, error: 5 }, 'error must be a string'], ] as const)('maps an invalid terminal result %j to a workflow error', async (value, message) => { const { start } = await setup(() => Promise.resolve({ logs: [], value })) const handle = start() try { const result = await handle.result expect(result.stopReason).toBe('error') expect(result.error).toContain(message) } finally { await handle.dispose() } }) it('preserves cancellation when the runtime rejects during cancellation', async () => { const entered = Promise.withResolvers() const failure = Promise.withResolvers() const { start } = await setup(() => { entered.resolve(undefined); return failure.promise }) const handle = start() try { await entered.promise handle.cancel('stop requested') failure.reject(new Error('execution provider failed while stopping')) const result = await handle.result expect(result.stopReason).toBe('cancelled') expect(result.error).toContain('stop requested') } finally { failure.reject(new Error('test cleanup')) await handle.dispose() } }) it('rejects a non-TypeScript runtime while loading the workflow provider', async () => { await expect(setup(undefined, 'python')).rejects.toThrow('requires the Node TypeScript PTC runtime') }) it('stops waiting for child output after disposal releases the child resources', async () => { const childResult = Promise.withResolvers() const disposed = vi.fn(() => Promise.resolve()) let outputWait: Promise | undefined const { ctx, start } = await setup(async (bindings) => { const child = await bindings.startChild!({ prompt: 'child' }) outputWait = bindings.childResult!(child) void outputWait.catch(() => {}) return completed }) vi.spyOn(ctx.subagents.getProvider('stub')!, 'start').mockResolvedValue({ id: SessionId('output-pending'), localAgent: undefined, result: childResult.promise, dispose: disposed, }) const handle = start() let settled = false void handle.result.then(() => { settled = true }) try { await new Promise(resolve => setImmediate(resolve)) expect(settled).toBe(true) expect(disposed).toHaveBeenCalledOnce() await expect(outputWait).rejects.toBe('workflow settled') expect((await handle.result).stopReason).toBe('completed') } finally { childResult.resolve({ output: [], stopReason: 'aborted' }) await handle.dispose() } }) })