Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

workflow: dynamic workflows — script-driven multi-agent orchestration

Browse Source 
A new capability family at packages/workflow/ in the bash seam shape,
modeled on Claude Code's dynamic workflows: the model writes a JavaScript
orchestration script (export const meta = {...} + plain-JS body), a runtime
executes it, and the script — not the conversation — holds the loop, the
branching, and the intermediate results.

- dsh-workflow (ctx.workflows): abstract WorkflowService + run vocabulary
  (WorkflowRun whose result NEVER rejects) + observe-only workflow/* events
  carrying data snapshots (id + meta, never the live run), per-listener
  contained like subagent/*.
- dsh-workflow-vm: in-process node:vm engine. Meta extraction via a
  string/comment-aware scanner (template interpolation rejected; literal
  evaluated alone in an empty timed context; statement blanked line-
  preservingly so stacks keep script line numbers). Hooks: agent(prompt,
  {label, phase, schema, model}) over ctx.subagents, parallel(), pipeline()
  (no cross-stage barrier), phase(), log(), args. Fatal-vs-null discipline:
  hook misuse (unknown/deferred options, bad arguments, unsupported
  schemas, tripped caps, seam start failures, cancellation) throws fatal
  WorkflowErrors the combinators RE-THROW — never dissolved into the
  per-item null reserved for child failures. Realm boundary: inbound values
  materialized by descriptor walks that never invoke accessors (defineProperty
  copies, __proto__-safe); outbound values rebuilt in-realm via the
  context's own JSON.parse. Determinism bans (Date.now/Math.random/argless
  new Date) kept so future resume support cannot break scripts. Caps and
  timeouts are validated Config. Every hook promise carries a no-op
  rejection consumer (app-boot exits on unhandled rejections).
- dsh-tool-workflow: the model-facing workflow tool, synchronous like
  dsh-tool-subagent (start → await → try/finally dispose; abort bridged;
  non-completed → isError). Generic render card titled by a textual
  meta.name sniff. The tool description carries the authoring contract.

Wired into examples/{coding-agent,acp-agent} with explicit-ask-only
guidance. Coverage at every tier: unit (meta scanner, materializer incl.
counting-getter and __proto__ regressions, combinator semantics,
concurrency ceiling, caps, cancellation, no-unhandled-rejection abandon),
integration over the real spawn stack, with-key e2e (real two-phase run +
the tool through the registry pipeline), and a recorded ACP snapshot
scenario (workflow-run, 1 child session). RFC:
docs/rfc/implemented/feature/2026-07-05-dynamic-workflows.md (deferred
work explicitly listed). AGENTS.md budget 1575 → 1590 for the new group's
layout line.
This commit is contained in:
Tianyi Cui
2026-07-05 13:29:35 +08:00
parent dafb81be7b
commit 1d43ea3cd5
52 changed files with 4459 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
packages/workflow/README.md Normal file
View File

@@ -0,0 +1,13 @@
# workflow/ — dynamic-workflow capability family
The workflow seam: a model-written JavaScript orchestration script that fans out subagents at scale (phases, structured per-agent results, concurrency caps), modeled on Claude Code's dynamic workflows. A capability seam (see [capability seams](../../docs/rfc/implemented/architecture/2026-06-13-capability-seams.md)) in the bash shape: ONE engine implementation per context registers as `ctx.workflows`; the model-facing tool consumes it.
| Package | Role | ctx key |
|---|---|---|
| `workflow/` | Abstract workflow seam: service base class + run vocabulary + `workflow/*` events | `ctx.workflows` |
| `workflow-vm/` | In-process `node:vm` engine: parses the script, injects the hooks, drives `ctx.subagents` | (provides `ctx.workflows`) |
| `tool-workflow/` | Model-facing `workflow` tool over `ctx.workflows` | (registers on `ctx.tools`) |
The interface lives at `workflow/workflow/`. The engine's `agent()` hook rides the [subagent seam](../subagent/README.md) (any registered provider; the shipped examples use `spawn`), and `agent({ schema })` rides the structured-output support the in-process backends implement. The seam split exists for engine hardening: `node:vm` is in-process and cannot kill a pathological synchronous spin — a worker-thread or isolated-vm engine swaps in behind the same interface if that ever matters.
The proposal, decisions, and deferred work: [docs/rfc/implemented/feature/2026-07-05-dynamic-workflows.md](../../docs/rfc/implemented/feature/2026-07-05-dynamic-workflows.md).

22
packages/workflow/tool-workflow/README.md Normal file
View File

@@ -0,0 +1,22 @@
# @deepseek-ai/dsh-tool-workflow
The model-facing **`workflow` tool**: run a JavaScript orchestration script that fans out subagents, and return the script's final value. Pure schema + lifecycle shaping over [`ctx.workflows`](../workflow/README.md) — script parsing, execution, caps, and cancellation live behind the seam, so a hardened engine swaps in without touching what the model sees.
## What the model sees
Two parameters: `script` (required — the full `export const meta = {...}` + body text; the tool DESCRIPTION carries the complete authoring contract: hooks, semantics, determinism bans, the supported schema subset) and `args` (optional JSON object exposed to the script as the `args` global; a bare list is wrapped as a field, a deliberate deviation from Claude Code's any-JSON `args` so the wire schema stays honest).
## Lifecycle
Collection is SYNCHRONOUS this cut (like [`dsh-tool-subagent`](../../subagent/tool-subagent/README.md)): `execute` starts a run and awaits `run.result` inside a `try/finally` that always disposes the run, so the script and its children reach quiescence on every path. `exec.signal` is bridged to `run.cancel()` (including the already-aborted-before-start case). A non-`completed` stop reason maps to an `isError` result reporting the reason — never partial output as success; a parse/meta failure thrown synchronously by `start()` becomes an `isError` the model can correct from. The completed result renders the meta name, the agent count, and the return value as JSON, truncated at `maxResultChars` with an explicit notice.
## Render intent
Decided up front (per the [render-intent RFC](../../../docs/rfc/implemented/architecture/2026-07-02-tool-render-intent-union.md)): a `generic` card titled `workflow: <meta.name>`, the name sniffed TEXTUALLY from `args.script` (presentation must be a pure function of args, so it cannot ask the engine to parse); the script text rides as `rawInput`. The result keeps the generic card.
## Config
| Key | Default | Meaning |
|---|---|---|
| `toolName` | `workflow` | The model-facing tool name to register. |
| `maxResultChars` | `50000` | Rendered-result ceiling; longer JSON is truncated with a notice. |

43
packages/workflow/tool-workflow/package.json Normal file
View File

@@ -0,0 +1,43 @@
{
"name": "@deepseek-ai/dsh-tool-workflow",
"description": "Model-facing workflow tool: run a JavaScript orchestration script over ctx.workflows",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-tools": "^0.0.1",
"@deepseek-ai/dsh-workflow": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"dependencies": {
"schemastery": "^3.18.0"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-workflow": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

179
packages/workflow/tool-workflow/src/index.ts Normal file
View File

@@ -0,0 +1,179 @@
/**
* The model-facing `workflow` tool: run a JavaScript orchestration script that
* fans out subagents, and return the script's final value. Pure schema +
* lifecycle shaping — script parsing, execution, caps, and cancellation live
* behind `ctx.workflows` (`@deepseek-ai/dsh-workflow`), so a hardened engine
* swaps in without touching what the model sees.
*
* Collection is SYNCHRONOUS this cut (like `dsh-tool-subagent`): `execute`
* starts a run and awaits `run.result` inside a `try/finally` that always
* disposes the run, so the script and its children are torn down on every
* path. A non-`completed` stop reason maps to an `isError` tool result (by
* throwing) rather than returning partial output as success. Background
* collection is deferred to the cross-tool background redesign.
*
* Render intent (decided up front, per the render-intent RFC): a `generic`
* card whose title carries the script's `meta.name`, sniffed textually from
* the args — presentation must be a pure function of `args`, so it cannot ask
* the engine to parse.
*
* @module @deepseek-ai/dsh-tool-workflow
*/
import type { Context } from 'cordis'
import z from 'schemastery'
import { defineTool } from '@deepseek-ai/dsh-tools'
import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { WorkflowResult, WorkflowRun } from '@deepseek-ai/dsh-workflow'
export const name = 'tool-workflow'
export const inject = ['tools', 'workflows']
/** Config: the model-facing tool name plus result rendering caps. */
export interface Config {
/** The model-facing tool name to register (default `workflow`). */
toolName?: string
/** Rendered-result ceiling, in characters: a longer JSON value is truncated with a notice (default 50000). */
maxResultChars?: number
}
export const Config: z<Config> = z.object({
toolName: z.string().default('workflow'),
maxResultChars: z.natural().min(1).default(50_000),
})
/**
* The script-authoring contract, embedded in the tool description. This IS the
* model-facing spec: the meta block, the hooks and their exact semantics, the
* determinism bans, and the supported schema subset.
*/
const DESCRIPTION = `Run a JavaScript workflow script that orchestrates subagents at scale. Use this for work that fans out across many independent pieces — an audit over many files, a migration, multi-angle research, adversarial verification of findings — where you write the orchestration as a script instead of delegating turn by turn.
The script MUST begin with \`export const meta = {...}\` — a PURE object literal (no variables, calls, or template interpolation) with required \`name\` (short kebab-case) and \`description\` strings, optional \`whenToUse\` string and \`phases\` array (\`{title, detail?, model?}\`). The body after it is plain JavaScript (NOT TypeScript) running with top-level await; end with \`return <value>\` — the value must be JSON-serializable and is this tool's result.
Script-body hooks:
- \`agent(prompt, opts?): Promise<any>\` — run one subagent to completion. Without \`opts.schema\` it resolves to the child's final text; with \`opts.schema\` (an object-rooted JSON Schema using ONLY type/properties/required/additionalProperties/items/enum/const — no oneOf/pattern/format/numeric bounds) it resolves to the validated object. Resolves \`null\` when the child fails (filter with \`.filter(Boolean)\`). Other opts: \`label\` (display), \`phase\` (progress group), \`model\` (override). Anything else (\`effort\`/\`isolation\`/\`agentType\`) is rejected loudly.
- \`pipeline(items, ...stages): Promise<any[]>\` — run each item through the stages independently with NO barrier between stages (prefer this for multi-stage work). Each stage receives \`(prev, item, index)\`. An ordinary stage throw drops that ITEM to \`null\` and skips its remaining stages.
- \`parallel(thunks): Promise<any[]>\` — run zero-argument functions concurrently and await ALL of them (a barrier; use only when a stage genuinely needs every prior result together). A throwing thunk resolves to \`null\`.
- \`phase(title)\` — start a progress phase; \`log(message)\` — narrate progress; \`args\` — the tool call's \`args\` input, verbatim.
Misused hooks (bad arguments, unknown options, unsupported schemas, tripped caps) throw errors that ALWAYS kill the script — they never dissolve into a per-item \`null\`.
Constraints: concurrency and total-agent caps apply; \`Date.now()\`, \`Math.random()\`, and argless \`new Date()\` throw (pass timestamps via \`args\`); no filesystem, network, timers, or Node.js APIs — the agents do the work, the script only coordinates them. The run executes in the foreground: this call returns when the whole script finishes.`
type WorkflowCallArgs = { script: string; args?: Record<string, unknown> }
/** Best-effort meta.name sniff for presentation (pure textual; no evaluation). */
function sniffMetaName(script: string): string | undefined {
const match = /export\s+const\s+meta\s*=\s*\{[^{}]*?name\s*:\s*(['"`])([^'"`\n]{1,64})\1/.exec(script)
return match?.[2]
}
/** The pending-state card: a generic card titled by the script's meta name. */
function presentWorkflowCall(args: WorkflowCallArgs): ToolCallView {
const name = sniffMetaName(args.script)
return {
card: 'generic',
title: name !== undefined ? `workflow: ${name}` : 'workflow',
rawInput: args.script,
}
}
/** The completed-state card: keep the pending title; render the result content as-is. */
function presentWorkflowResult(args: WorkflowCallArgs, result: { content: ContentBlock[]; isError: boolean }): ToolResultView {
void args
void result
return { card: 'generic' }
}
/** A non-`completed` stop reason means the script did not finish cleanly. */
function stopReasonError(result: WorkflowResult): string | undefined {
switch (result.stopReason) {
case 'completed':
return undefined
case 'cancelled':
return `workflow run was cancelled${result.error !== undefined ? ` (${result.error})` : ''}`
case 'error':
return `workflow run failed: ${result.error ?? 'unknown error'}`
/* v8 ignore start -- defensive: WorkflowStopReason is a closed union, exhaustive by construction; a future variant fails here loudly */
default:
return `workflow run ended abnormally (${String(result.stopReason satisfies never)})`
/* v8 ignore stop */
}
}
/** Render the run's outcome text: the meta name, agent count, and the JSON value (capped). */
function renderResult(run: WorkflowRun, result: WorkflowResult, maxChars: number): string {
// The engine returns JSON data (null for a valueless script), so stringify never yields undefined.
const rendered = JSON.stringify(result.value, null, 2)
const clipped = rendered.length > maxChars
? `${rendered.slice(0, maxChars)}\n… [truncated: ${rendered.length - maxChars} more characters]`
: rendered
return `workflow "${run.meta.name}" completed (${result.agentsStarted} agent${result.agentsStarted === 1 ? '' : 's'}).\nReturn value:\n${clipped}`
}
export function apply(ctx: Context, config: Config): void {
const maxResultChars = config.maxResultChars ?? 50_000
ctx.tools.register(defineTool({
name: config.toolName ?? 'workflow',
description: DESCRIPTION,
parameters: {
script: {
type: 'string',
required: true,
description: 'The complete workflow script: `export const meta = {...}` followed by the plain-JS body (top-level await allowed; end with `return <json-value>`).',
},
args: {
type: 'object',
description: 'Optional JSON input exposed to the script as the `args` global (wrap a bare list as a field, e.g. {"files": [...]}).',
},
},
async execute(args, exec): Promise<ContentBlock[]> {
const parent = exec.agent
if (!parent) {
// The loop sets `exec.agent` for every model-driven call; its absence
// means a non-agent caller invoked the tool directly, which has no
// parent to attribute the children to. Fail loud rather than guess.
throw new Error('workflow tool requires a calling agent (exec.agent was undefined)')
}
// Parse failures (SCRIPT_PARSE/META_INVALID) throw synchronously here
// and become isError results via the registry — the model sees the
// violation list and can correct the script.
const run: WorkflowRun = ctx.workflows.start({
script: args.script,
...args.args !== undefined ? { args: args.args } : {},
parent,
...exec.signal ? { signal: exec.signal } : {},
})
// Bridge the tool's abort signal to the run: if the parent step is
// aborted while the script is in flight, cancel the whole run. The
// engine also receives `signal` directly, but an explicit bridge keeps
// the tool's contract local (and covers an engine that ignores it).
const onAbort = (): void => { run.cancel('parent step aborted') }
exec.signal?.addEventListener('abort', onAbort, { once: true })
// `addEventListener` does NOT fire for a signal already aborted before
// this line — cancel explicitly in that case.
if (exec.signal?.aborted) run.cancel('parent step aborted')
try {
const result = await run.result
const error = stopReasonError(result)
if (error !== undefined) {
// Map a non-clean finish to an isError result (the registry turns a
// throw into an isError). Report the reason, not partial output.
throw new Error(error)
}
return [{ type: 'text', text: renderResult(run, result, maxResultChars) }]
} finally {
exec.signal?.removeEventListener('abort', onAbort)
// Always reach run quiescence — never leak a live script or children.
await run.dispose()
}
},
presentCall: args => presentWorkflowCall(args),
presentResult: (args, result) => presentWorkflowResult(args, result),
}))
}

224
packages/workflow/tool-workflow/tests/tool-workflow.spec.ts Normal file
View File

@@ -0,0 +1,224 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import type { ToolExecutionResult } from '@deepseek-ai/dsh-tools'
import { AgentId } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { WorkflowRunId, WorkflowService } from '@deepseek-ai/dsh-workflow'
import type { WorkflowResult, WorkflowRun, WorkflowStartRequest } from '@deepseek-ai/dsh-workflow'
import { CallId } from '@deepseek-ai/dsh-llm'
import * as toolWorkflow from '../src/index.ts'
/** A controllable engine standing in behind ctx.workflows (the tool's only seam). */
class StubEngine extends WorkflowService {
requests: WorkflowStartRequest[] = []
cancels: string[] = []
disposed = 0
settle!: (result: WorkflowResult) => void
startError: Error | undefined
start(request: WorkflowStartRequest): WorkflowRun {
if (this.startError) throw this.startError
this.requests.push(request)
const result = new Promise<WorkflowResult>((resolve) => { this.settle = resolve })
request.signal?.addEventListener('abort', () => {
this.settle({ value: null, stopReason: 'cancelled', error: 'signal', agentsStarted: 0 })
}, { once: true })
return {
id: WorkflowRunId('run-1'),
meta: { name: 'stub-flow', description: 'd' },
result,
cancel: (reason?: string) => {
this.cancels.push(reason ?? 'cancelled')
this.settle({ value: null, stopReason: 'cancelled', ...reason !== undefined ? { error: reason } : {}, agentsStarted: 0 })
},
dispose: () => {
this.disposed += 1
return Promise.resolve()
},
}
}
}
async function setup(config?: { toolName?: string; maxResultChars?: number }) {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(StubEngine)
await ctx.plugin(toolWorkflow, config ?? {})
const engine = ctx.workflows as StubEngine
const parent = { id: AgentId('caller'), options: {} } as unknown as Agent
return { ctx, engine, parent }
}
const SCRIPT = "export const meta = { name: 'audit', description: 'd' }\nreturn 1"
function execute(ctx: Context, args: unknown, extra?: { agent?: Agent; signal?: AbortSignal }): Promise<ToolExecutionResult> {
return ctx.tools.execute({
callId: CallId('call-1'),
name: 'workflow',
arguments: args,
...extra?.agent ? { agent: extra.agent } : {},
...extra?.signal ? { signal: extra.signal } : {},
})
}
describe('dsh-tool-workflow', () => {
it('starts a run with the script/args/parent/signal and renders the completed value', async () => {
const { ctx, engine, parent } = await setup()
const controller = new AbortController()
const pending = execute(ctx, { script: SCRIPT, args: { files: ['a.ts'] } }, { agent: parent, signal: controller.signal })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
expect(engine.requests[0]).toMatchObject({ script: SCRIPT, args: { files: ['a.ts'] }, parent })
expect(engine.requests[0]!.signal).toBe(controller.signal)
engine.settle({ value: { findings: [1, 2] }, stopReason: 'completed', agentsStarted: 7 })
const result = await pending
expect(result.isError).toBe(false)
const rendered = (result.content[0] as { text: string }).text
expect(rendered).toContain('workflow "stub-flow" completed (7 agents)')
expect(rendered).toContain('"findings"')
expect(engine.disposed).toBe(1)
})
it('maps a non-completed stop reason to an isError result (and still disposes)', async () => {
const { ctx, engine, parent } = await setup()
const pending = execute(ctx, { script: SCRIPT }, { agent: parent })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
engine.settle({ value: null, stopReason: 'error', error: 'script threw: boom', agentsStarted: 2 })
const result = await pending
expect(result.isError).toBe(true)
expect((result.content[0] as { text: string }).text).toContain('workflow run failed: script threw: boom')
expect(engine.disposed).toBe(1)
})
it('reports a cancelled run distinctly (with and without a reason)', async () => {
const { ctx, engine, parent } = await setup()
const pending = execute(ctx, { script: SCRIPT }, { agent: parent })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
engine.settle({ value: null, stopReason: 'cancelled', error: 'user', agentsStarted: 0 })
const result = await pending
expect(result.isError).toBe(true)
expect((result.content[0] as { text: string }).text).toContain('workflow run was cancelled (user)')
const bare = execute(ctx, { script: SCRIPT }, { agent: parent })
await vi.waitFor(() => { expect(engine.requests.length).toBe(2) })
engine.settle({ value: null, stopReason: 'cancelled', agentsStarted: 0 })
expect(((await bare).content[0] as { text: string }).text.trim().endsWith('cancelled')).toBe(true)
})
it('an error result without a message renders the unknown-error fallback', async () => {
const { ctx, engine, parent } = await setup()
const pending = execute(ctx, { script: SCRIPT }, { agent: parent })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
engine.settle({ value: null, stopReason: 'error', agentsStarted: 0 })
expect(((await pending).content[0] as { text: string }).text).toContain('unknown error')
})
it('cancels the run when exec.signal aborts MID-FLIGHT (the abort bridge)', async () => {
const { ctx, engine, parent } = await setup()
const controller = new AbortController()
const pending = execute(ctx, { script: SCRIPT }, { agent: parent, signal: controller.signal })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
controller.abort()
const result = await pending
expect(result.isError).toBe(true)
expect(engine.cancels).toContain('parent step aborted')
expect(engine.disposed).toBe(1)
})
it('applies raw-config fallbacks when loaded without schemastery defaults (direct apply)', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(StubEngine)
// Direct apply with an empty RAW config: the `??` fallbacks resolve the
// tool name and render cap without schemastery having filled them.
toolWorkflow.apply(ctx, {})
expect(ctx.tools.get('workflow')).toBeDefined()
})
it('a synchronous engine start throw (parse/meta failure) becomes an isError result', async () => {
const { ctx, engine, parent } = await setup()
engine.startError = new Error('script must begin with `export const meta = {...}`')
const result = await execute(ctx, { script: 'nope' }, { agent: parent })
expect(result.isError).toBe(true)
expect((result.content[0] as { text: string }).text).toContain('must begin with')
})
it('requires a calling agent (fails loud without exec.agent)', async () => {
const { ctx, engine } = await setup()
const result = await execute(ctx, { script: SCRIPT })
expect(result.isError).toBe(true)
expect((result.content[0] as { text: string }).text).toContain('requires a calling agent')
expect(engine.requests.length).toBe(0)
})
it('validates its own arguments via the schema DSL (missing script)', async () => {
const { ctx, parent } = await setup()
const result = await execute(ctx, {}, { agent: parent })
expect(result.isError).toBe(true)
expect(result.error?.code).toBe('INVALID_ARGS')
})
it('cancels the run when exec.signal is ALREADY aborted at call time', async () => {
const { ctx, engine, parent } = await setup()
const controller = new AbortController()
controller.abort()
const result = await execute(ctx, { script: SCRIPT }, { agent: parent, signal: controller.signal })
expect(result.isError).toBe(true)
expect(engine.cancels).toContain('parent step aborted')
expect(engine.disposed).toBe(1)
})
it('truncates an oversized rendered value with a notice (maxResultChars)', async () => {
const { ctx, engine, parent } = await setup({ maxResultChars: 40 })
const pending = execute(ctx, { script: SCRIPT }, { agent: parent })
await vi.waitFor(() => { expect(engine.requests.length).toBe(1) })
engine.settle({ value: { blob: 'x'.repeat(500) }, stopReason: 'completed', agentsStarted: 1 })
const rendered = ((await pending).content[0] as { text: string }).text
expect(rendered).toContain('[truncated:')
expect(rendered.length).toBeLessThan(400)
})
it('registers under a configured toolName and unregisters on fiber dispose (HMR safety)', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(StubEngine)
const fiber = await ctx.plugin(toolWorkflow, { toolName: 'orchestrate' })
expect(ctx.tools.get('orchestrate')).toBeDefined()
expect(ctx.tools.get('workflow')).toBeUndefined()
await fiber.dispose()
expect(ctx.tools.get('orchestrate')).toBeUndefined()
})
it('presents a generic pending card titled by the sniffed meta name, with the script as rawInput', async () => {
const { ctx } = await setup()
const tool = ctx.tools.get('workflow')!
const view = tool.presentCall!({ script: SCRIPT })
expect(view).toMatchObject({ card: 'generic', title: 'workflow: audit', rawInput: SCRIPT })
const anonymous = tool.presentCall!({ script: 'export const meta = {}\nreturn 1' })
expect(anonymous).toMatchObject({ card: 'generic', title: 'workflow' })
})
it('presentResult keeps the generic card; presentation is pure and replay-safe on malformed args', async () => {
const { ctx } = await setup()
const tool = ctx.tools.get('workflow')!
expect(tool.presentResult!({ script: SCRIPT }, { content: [], isError: false })).toEqual({ card: 'generic' })
// defineTool soft-validates presentation args: a malformed logged shape
// falls back to undefined instead of throwing mid-replay.
expect(tool.presentCall!({ not: 'the schema' })).toBeUndefined()
})
it('has the namespace-plugin export shape (no stray default)', () => {
expect('default' in toolWorkflow).toBe(false)
expect(toolWorkflow.name).toBe('tool-workflow')
expect(toolWorkflow.inject).toEqual(['tools', 'workflows'])
const loader = Object.create(Loader.prototype) as Loader
const unwrapped = loader.unwrapExports(toolWorkflow) as Record<string, unknown>
expect(unwrapped).toBe(toolWorkflow)
expect(typeof unwrapped.apply).toBe('function')
})
})

33
packages/workflow/tool-workflow/tsconfig.json Normal file
View File

@@ -0,0 +1,33 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../../vendor/schemastery"
},
{
"path": "../../core/agent"
},
{
"path": "../../llm/llm"
},
{
"path": "../../core/tools"
},
{
"path": "../workflow"
}
]
}

30
packages/workflow/workflow-vm/README.md Normal file
View File

@@ -0,0 +1,30 @@
# @deepseek-ai/dsh-workflow-vm
The first [`WorkflowService`](../workflow/README.md) implementation: an in-process **`node:vm` engine**. It parses the Claude Code-format script (`export const meta = {...}` + plain-JS body), runs the body in a fresh vm context with the workflow hooks injected, and fans `agent()` calls out to [`ctx.subagents`](../../subagent/README.md).
## The script contract it executes
- **Meta extraction** (`extractMeta`): a string/comment-aware brace scanner finds the leading `export const meta` literal (template interpolation rejected — the literal must be pure), evaluates it ALONE in an empty timed vm context, materializes the result to plain JSON data, validates the shape (`name`/`description` required; unknown fields rejected loud), and blanks the statement line-preservingly so error stacks keep the script's own line numbers.
- **Hooks**: `agent(prompt, {label, phase, schema, model})` (schema = the [structured-output subset](../../core/tools/README.md), forwarded as `outputSchema`; result = validated object, or final text without a schema; a failed child resolves `null`), `parallel(thunks)`, `pipeline(items, ...stages)` with NO cross-stage barrier and `(prev, item, index)` stage callbacks, `phase(title)`, `log(message)`, and the `args` global. Anything else — `effort`/`isolation`/`agentType`, unknown options, malformed arguments, schemas outside the subset — throws a FATAL `WorkflowError` that `parallel`/`pipeline` re-throw rather than nulling (see the seam README's failure discipline).
- **Determinism bans**: `Date.now()`, `Math.random()`, and argless `new Date()` throw (kept even though resume is deferred, so scripts stay resume-compatible); no timers, filesystem, or Node APIs exist in the context.
## Realm discipline
Values ENTERING the host (the meta literal, hook options/schemas, the script's return) are materialized by `materializeFromRealm`: a descriptor walk that never invokes accessors and rejects loud everything JSON cannot carry (accessors, exotic prototypes, functions, symbols, cycles, sparse arrays, non-finite numbers, nested `undefined`), copying into host containers via `defineProperty` so a `"__proto__"` key becomes a data property, never a prototype mutation. Values ENTERING the realm (`args`, `agent()` results) are rebuilt INSIDE the realm through the context's own `JSON.parse`, so the script never holds an object whose prototype chain reaches host intrinsics.
## Limits, cancellation, disposal
Per-run: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` aborts every child (a shared `AbortSignal`), rejects waiting `agent()` slots, and makes every future hook call throw `CANCELLED` — the script dies at its next await and the run settles `cancelled`. Once a run settles, stray children a script fired without awaiting are aborted too. Every hook-returned promise carries a no-op rejection consumer, so a dropped promise cannot surface an unhandled rejection (the app boot layer exits the process on those).
**Documented limitations** (the accepted cost of the in-process mechanism; the seam exists so a worker-thread/isolated-vm engine can swap in): vm is NOT a security boundary — scripts are model-written, the same trust level as the model's bash access — and the vm `timeout` covers only the initial synchronous slice, so a pathological synchronous spin after the first await cannot be killed; `dispose()` waits `disposeGraceMs` then ABANDONS such a script (its settlement stays contained, but an abandoned spin would still occupy the event loop).
## Config
| Key | Default | Meaning |
|---|---|---|
| `provider` | `spawn` | The `ctx.subagents` provider children run on. |
| `maxConcurrentAgents` | `0` (auto) | Concurrent `agent()` ceiling; `0` resolves to `min(16, max(1, cores - 2))`. |
| `maxTotalAgents` | `1000` | Total `agent()` calls one run may start (runaway-loop backstop). |
| `maxItemsPerCall` | `4096` | Items accepted by one `parallel()`/`pipeline()` call. |
| `syncTimeoutMs` | `5000` | vm timeout for the initial synchronous slice and the meta evaluation. |
| `disposeGraceMs` | `5000` | How long `dispose()` waits for a cancelled script before abandoning it. |

50
packages/workflow/workflow-vm/package.json Normal file
View File

@@ -0,0 +1,50 @@
{
"name": "@deepseek-ai/dsh-workflow-vm",
"description": "node:vm workflow engine: executes model-written orchestration scripts over ctx.subagents",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-subagent": "^0.0.1",
"@deepseek-ai/dsh-tools": "^0.0.1",
"@deepseek-ai/dsh-workflow": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"dependencies": {
"schemastery": "^3.18.0"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-agent-loop": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/dsh-subagent-spawn": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-workflow": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

166
packages/workflow/workflow-vm/src/index.ts Normal file
View File

@@ -0,0 +1,166 @@
/**
* The `node:vm` workflow engine: the first {@link WorkflowService}
* implementation. Parses the Claude Code-format script (meta + body), runs the
* body in a fresh in-process vm context with the workflow hooks injected, and
* fans `agent()` calls out to `ctx.subagents`.
*
* Engine limitations, documented as the accepted cost of the in-process
* mechanism (the interface/implementation seam exists precisely so a
* worker-thread or isolated-vm engine can swap in if these ever matter):
*
* - vm is NOT a security boundary. Scripts are model-written — the same trust
* level as the model's bash access — and the realm-boundary materialization
* is correctness containment, not a sandbox.
* - The vm `timeout` covers only the initial SYNCHRONOUS slice of the script;
* a pathological synchronous spin after the first await cannot be killed
* in-process. `dispose()` therefore waits a bounded grace and then ABANDONS
* a stuck script: its pending hook promises are already rejected and its
* settlement is contained (no unhandled rejection), but an abandoned
* synchronous spin would still occupy the event loop.
*
* Plugin export shape: a default-exported {@link WorkflowService} subclass
* (the class-based service form, like `dsh-bash-local`).
*
* @module @deepseek-ai/dsh-workflow-vm
*/
import { randomUUID } from 'node:crypto'
import { availableParallelism } from 'node:os'
import type { Context } from 'cordis'
import z from 'schemastery'
import WorkflowService, { WorkflowRunId } from '@deepseek-ai/dsh-workflow'
import type { WorkflowResult, WorkflowRun, WorkflowRunInfo, WorkflowStartRequest } from '@deepseek-ai/dsh-workflow'
import { extractMeta } from './meta.ts'
import { WorkflowExecution, type ExecutionLimits } from './runtime.ts'
export { extractMeta, type ExtractedScript } from './meta.ts'
export { materializeFromRealm, MaterializeError } from './realm.ts'
export { WorkflowExecution, type ExecutionLimits, type ExecutionObserver } from './runtime.ts'
/** Plugin config (all optional — `static Config` supplies the defaults). */
export interface Config {
/** The `ctx.subagents` provider children run on (default `spawn`). */
provider?: string
/** Concurrent `agent()` ceiling; `0` (the default) auto-resolves to `min(16, max(1, cores - 2))`. */
maxConcurrentAgents?: number
/** Total `agent()` calls one run may start — the runaway-loop backstop (default 1000). */
maxTotalAgents?: number
/** Items accepted by a single `parallel()`/`pipeline()` call (default 4096). */
maxItemsPerCall?: number
/** vm timeout for the script's initial synchronous slice AND the meta-literal evaluation (default 5000 ms). */
syncTimeoutMs?: number
/** How long `dispose()` waits for a cancelled script to settle before abandoning it (default 5000 ms). */
disposeGraceMs?: number
}
type ResolvedConfig = Required<Config>
/**
* The vm engine service. `start()` validates the script up front (meta +
* body compile) and returns a {@link WorkflowRun} whose `result` never
* rejects; the `workflow/*` events fire around the run per the seam contract.
*/
export class VmWorkflowEngine extends WorkflowService {
static inject = ['subagents']
static Config: z<Config> = z.object({
provider: z.string().default('spawn'),
maxConcurrentAgents: z.natural().default(0),
maxTotalAgents: z.natural().min(1).default(1000),
maxItemsPerCall: z.natural().min(1).default(4096),
syncTimeoutMs: z.natural().min(1).default(5000),
disposeGraceMs: z.natural().default(5000),
})
private readonly config: ResolvedConfig
constructor(ctx: Context, config: Config) {
super(ctx)
// schemastery (static Config) has already filled the defaulted fields;
// the assertion records that resolution, not a hidden fallback.
this.config = config as ResolvedConfig
}
/**
* Parse and execute a workflow script. Throws {@link WorkflowError}
* synchronously (`SCRIPT_PARSE`/`META_INVALID`) for a script that cannot
* begin; once a run is returned, every failure resolves through
* `result.stopReason` instead.
* @param request - the script, its `args`, the parent agent, and an
* optional cancel signal.
* @returns the live run (its `result` resolves when the script settles).
*/
start(request: WorkflowStartRequest): WorkflowRun {
const { meta, body } = extractMeta(request.script, this.config.syncTimeoutMs)
const id = WorkflowRunId(randomUUID())
// The event payloads and the run handle get SEPARATE meta clones: a
// listener mutating its snapshot must not corrupt the holder's view.
const info: WorkflowRunInfo = { id, meta: structuredClone(meta) }
const limits: ExecutionLimits = {
provider: this.config.provider,
maxConcurrentAgents: this.config.maxConcurrentAgents === 0
? Math.min(16, Math.max(1, availableParallelism() - 2))
: this.config.maxConcurrentAgents,
maxTotalAgents: this.config.maxTotalAgents,
maxItemsPerCall: this.config.maxItemsPerCall,
syncTimeoutMs: this.config.syncTimeoutMs,
}
const execution = new WorkflowExecution(
this.ctx,
meta,
body,
request.parent,
request.args,
request.signal,
limits,
{
phase: (title) => { this.emitWorkflowEvent('workflow/phase', info, title) },
log: (message) => { this.emitWorkflowEvent('workflow/log', info, message) },
agentStart: (agent) => { this.emitWorkflowEvent('workflow/agent-start', info, agent) },
agentEnd: (agent) => { this.emitWorkflowEvent('workflow/agent-end', info, agent) },
},
)
this.emitWorkflowEvent('workflow/start', info)
const result: Promise<WorkflowResult> = execution.drive()
// `workflow/end` fires as the (never-rejecting) result settles, with the
// outcome DATA only — the value stays with the run's holder.
void result.then((settled) => {
this.emitWorkflowEvent('workflow/end', info, {
stopReason: settled.stopReason,
...settled.error !== undefined ? { error: settled.error } : {},
agentsStarted: settled.agentsStarted,
})
})
let disposed: Promise<void> | undefined
return {
id,
meta: structuredClone(meta),
result,
cancel(reason?: string): void {
execution.cancel(reason)
},
dispose: (): Promise<void> => {
// Idempotent: cancel, then wait min(settle, grace). `result` never
// rejects, so the race needs no rejection handling; an unsettled
// script past the grace is abandoned per the module contract.
disposed ??= (async () => {
execution.cancel('workflow disposed')
await Promise.race([result, sleep(this.config.disposeGraceMs)])
})()
return disposed
},
}
}
}
/** A plain timer sleep (the dispose grace); unref'd so it never holds the process open. */
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => {
const timer = setTimeout(resolve, ms)
timer.unref()
})
}
export default VmWorkflowEngine

198
packages/workflow/workflow-vm/src/meta.ts Normal file
View File

@@ -0,0 +1,198 @@
/**
* Meta-block extraction: turn a Claude Code-format workflow script —
* `export const meta = {...}` followed by a plain-JS body — into a validated
* {@link WorkflowMeta} plus the body with the meta statement blanked
* line-preservingly (error stacks keep the script's own line numbers).
*
* The scanner is a small string/comment-aware brace matcher, not a JS parser:
* it only has to find the END of the meta object literal, and the literal is
* contractually PURE (no interpolation, no computed values). Template strings
* are tolerated as plain quotes but `${` inside one is rejected up front —
* interpolation is where "literal" stops being checkable by evaluation. The
* extracted text is then evaluated ALONE in an empty, timed vm context (a
* non-literal reference throws there; an expression can still RUN, so the
* result — not the source — is the contract: it must materialize to plain
* JSON data and pass the shape validation).
*
* @module @deepseek-ai/dsh-workflow-vm/meta
*/
import * as vm from 'node:vm'
import { WorkflowError } from '@deepseek-ai/dsh-workflow'
import type { WorkflowMeta, WorkflowPhase } from '@deepseek-ai/dsh-workflow'
import { materializeFromRealm, MaterializeError } from './realm.ts'
/** The result of {@link extractMeta}: the validated meta and the runnable body. */
export interface ExtractedScript {
meta: WorkflowMeta
/** The script with the meta statement blanked (newlines preserved). */
body: string
}
const META_PREFIX = /^\s*(?:\/\/[^\n]*\n|\/\*[\s\S]*?\*\/\s*|\s+)*export\s+const\s+meta\s*=\s*/
/**
* Scan `source` from `start` (an opening `{`) to its matching `}`, aware of
* string literals (`'`/`"`/backtick, with escapes) and comments. Returns the
* index AFTER the closing brace. Throws `SCRIPT_PARSE` on template
* interpolation (`${` inside a backtick string) or an unterminated literal.
*/
function scanObjectLiteral(source: string, start: number): number {
let depth = 0
let index = start
while (index < source.length) {
const ch = source.charAt(index)
if (ch === '/' && source[index + 1] === '/') {
const end = source.indexOf('\n', index)
index = end === -1 ? source.length : end + 1
continue
}
if (ch === '/' && source[index + 1] === '*') {
const end = source.indexOf('*/', index + 2)
if (end === -1) throw new WorkflowError('meta block has an unterminated comment', 'SCRIPT_PARSE')
index = end + 2
continue
}
if (ch === '\'' || ch === '"' || ch === '`') {
index = scanString(source, index, ch)
continue
}
if (ch === '{' || ch === '[') depth += 1
if (ch === '}' || ch === ']') {
depth -= 1
if (depth === 0) return index + 1
}
index += 1
}
throw new WorkflowError('meta block is not a balanced object literal', 'SCRIPT_PARSE')
}
/** Scan past one string literal starting at `start` (the quote char); returns the index after the closing quote. */
function scanString(source: string, start: number, quote: string): number {
let index = start + 1
while (index < source.length) {
const ch = source.charAt(index)
if (ch === '\\') {
index += 2
continue
}
if (quote === '`' && ch === '$' && source[index + 1] === '{') {
throw new WorkflowError('template interpolation (`${...}`) is not allowed in the meta block — meta must be a pure literal', 'SCRIPT_PARSE')
}
if (ch === quote) return index + 1
index += 1
}
throw new WorkflowError('meta block has an unterminated string literal', 'SCRIPT_PARSE')
}
/** Replace `[from, to)` of `source` with whitespace, preserving every newline (line numbers survive). */
function blankSpan(source: string, from: number, to: number): string {
const blanked = source.slice(from, to).replace(/[^\n]/g, ' ')
return source.slice(0, from) + blanked + source.slice(to)
}
/** Collect shape violations for an evaluated meta value (already materialized to host JSON data). */
function validateMetaShape(meta: unknown): { meta?: WorkflowMeta; violations: string[] } {
const violations: string[] = []
/* v8 ignore next 3 -- defensive: the scanner only extracts a brace-delimited literal, which always evaluates to a plain object */
if (typeof meta !== 'object' || meta === null || Array.isArray(meta)) {
return { violations: ['meta must be an object literal'] }
}
const record = meta as Record<string, unknown>
const known = new Set(['name', 'description', 'whenToUse', 'phases'])
for (const key of Object.keys(record)) {
if (!known.has(key)) violations.push(`meta.${key} is not a recognized field (name/description/whenToUse/phases)`)
}
if (typeof record.name !== 'string' || record.name.length === 0) violations.push('meta.name must be a non-empty string')
if (typeof record.description !== 'string' || record.description.length === 0) violations.push('meta.description must be a non-empty string')
if (record.whenToUse !== undefined && typeof record.whenToUse !== 'string') violations.push('meta.whenToUse must be a string')
const phases: WorkflowPhase[] = []
if (record.phases !== undefined) {
if (!Array.isArray(record.phases)) {
violations.push('meta.phases must be an array')
} else {
record.phases.forEach((phase, index) => {
if (typeof phase !== 'object' || phase === null || Array.isArray(phase)) {
violations.push(`meta.phases[${index}] must be an object`)
return
}
const entry = phase as Record<string, unknown>
for (const key of Object.keys(entry)) {
if (!['title', 'detail', 'model'].includes(key)) violations.push(`meta.phases[${index}].${key} is not a recognized field`)
}
if (typeof entry.title !== 'string' || entry.title.length === 0) violations.push(`meta.phases[${index}].title must be a non-empty string`)
if (entry.detail !== undefined && typeof entry.detail !== 'string') violations.push(`meta.phases[${index}].detail must be a string`)
if (entry.model !== undefined && typeof entry.model !== 'string') violations.push(`meta.phases[${index}].model must be a string`)
if (violations.length === 0) {
phases.push({
title: entry.title as string,
...entry.detail !== undefined ? { detail: entry.detail as string } : {},
...entry.model !== undefined ? { model: entry.model as string } : {},
})
}
})
}
}
if (violations.length > 0) return { violations }
return {
violations,
meta: {
name: record.name as string,
description: record.description as string,
...record.whenToUse !== undefined ? { whenToUse: record.whenToUse as string } : {},
...record.phases !== undefined ? { phases } : {},
},
}
}
/**
* Extract and validate the leading `export const meta = {...}` statement.
* Throws {@link WorkflowError} — `SCRIPT_PARSE` when the statement is missing
* or unscannable, `META_INVALID` when the literal evaluates to something
* outside the meta contract (non-JSON data, wrong shape, unknown fields).
* @param script - the full script text.
* @param evalTimeoutMs - the vm timeout for evaluating the extracted literal.
* @returns the validated meta and the line-preservingly blanked body.
*/
export function extractMeta(script: string, evalTimeoutMs: number): ExtractedScript {
const match = META_PREFIX.exec(script)
if (!match) {
throw new WorkflowError('script must begin with `export const meta = {...}` (leading comments allowed)', 'SCRIPT_PARSE')
}
const literalStart = match[0].length
if (script[literalStart] !== '{') {
throw new WorkflowError('`export const meta =` must be followed by an object literal', 'SCRIPT_PARSE')
}
const literalEnd = scanObjectLiteral(script, literalStart)
const literal = script.slice(literalStart, literalEnd)
let evaluated: unknown
try {
// An EMPTY context: any non-literal reference (a variable, a call) throws
// here. The result — data only — is what the contract checks; a getter or
// IIFE can still run, which is why the timeout and the materialization
// below are part of the same boundary.
evaluated = vm.runInNewContext(`(${literal})`, undefined, { timeout: evalTimeoutMs })
} catch (error: unknown) {
throw new WorkflowError(`meta block failed to evaluate as a pure literal: ${String(error)}`, 'META_INVALID', { cause: error })
}
let data: unknown
try {
data = materializeFromRealm(evaluated, 'meta')
} catch (error: unknown) {
/* v8 ignore next -- defensive rethrow arm: materializeFromRealm only throws MaterializeError */
if (!(error instanceof MaterializeError)) throw error
throw new WorkflowError(`meta block is not pure JSON data — ${error.message}`, 'META_INVALID', { cause: error })
}
const { meta, violations } = validateMetaShape(data)
if (meta === undefined) {
throw new WorkflowError(`invalid meta block: ${violations.join('; ')}`, 'META_INVALID')
}
// Blank the whole statement (including a trailing semicolon, if any) so the
// body compiles standalone with its original line numbers.
let statementEnd = literalEnd
while (statementEnd < script.length && (script[statementEnd] === ' ' || script[statementEnd] === '\t')) statementEnd += 1
if (script[statementEnd] === ';') statementEnd += 1
return { meta, body: blankSpan(script, 0, statementEnd) }
}

142
packages/workflow/workflow-vm/src/realm.ts Normal file
View File

@@ -0,0 +1,142 @@
/**
* Realm-boundary materialization for the vm engine.
*
* Values produced INSIDE the script realm (the meta literal, hook arguments,
* the script's return value) must become plain host-realm JSON data before the
* host touches them. The repo's `isJsonValue` guard cannot run first: it is
* prototype-strict (any cross-realm object fails it) and it INVOKES getters
* (letting realm code run outside the vm's timed window). So this module walks
* own-property DESCRIPTORS — never invoking accessors — and copies data into
* host containers, rejecting loud everything JSON cannot carry:
* accessor properties, non-plain prototypes, functions, symbols (keys or
* values), bigints, non-finite numbers, `undefined` values, cycles, sparse
* arrays, and arrays with non-index own properties.
*
* Host objects are built with `Object.defineProperty` into a fresh `{}` —
* never plain `target[key] =` assignment, which a `"__proto__"` key would turn
* into prototype mutation instead of a data property.
*
* The host→realm direction deliberately does NOT live here: a host object
* handed into the realm would expose host intrinsics through its prototype
* chain, so the engine rebuilds inbound values INSIDE the realm via the
* context's own `JSON.parse` (see the runtime).
*
* @module @deepseek-ai/dsh-workflow-vm/realm
*/
/** Thrown by {@link materializeFromRealm}; the caller wraps it into the right `WorkflowError` code. */
export class MaterializeError extends Error {
constructor(public readonly path: string, public readonly reason: string) {
super(`${path}: ${reason}`)
this.name = 'MaterializeError'
}
}
/**
* Whether an object's prototype chain is data-shaped: `null`, or a prototype
* whose own prototype is `null` (the realm's `Object.prototype` — which we
* cannot compare by identity across realms). A `Date`/`Map`/class instance
* has a longer chain and is rejected.
*/
function hasPlainPrototype(value: object): boolean {
const proto: unknown = Object.getPrototypeOf(value)
if (proto === null) return true
return Object.getPrototypeOf(proto) === null
}
/**
* Copy `value` (typically from the vm realm) into plain host JSON data.
* Throws {@link MaterializeError} naming the offending path for anything JSON
* cannot carry losslessly. Accessors are detected via descriptors and NEVER
* invoked. `undefined` is accepted only at the ROOT (a script with no
* `return` value) — the caller decides what it means; an `undefined` nested
* INSIDE a container is a violation.
* @param value - the realm value to materialize.
* @param root - the path label for the root value (error messages).
* @returns the host-realm copy (plain objects/arrays/scalars only).
*/
export function materializeFromRealm(value: unknown, root = 'value'): unknown {
if (value === undefined) return undefined
return materialize(value, root, new Set())
}
function materialize(value: unknown, path: string, seen: Set<object>): unknown {
switch (typeof value) {
case 'boolean':
case 'string':
return value
case 'number': {
if (!Number.isFinite(value)) throw new MaterializeError(path, 'non-finite numbers are not JSON data')
return value
}
case 'bigint':
throw new MaterializeError(path, 'bigints are not JSON data')
case 'function':
throw new MaterializeError(path, 'functions cannot cross the workflow realm boundary')
case 'symbol':
throw new MaterializeError(path, 'symbols cannot cross the workflow realm boundary')
case 'undefined':
throw new MaterializeError(path, 'undefined is not JSON data')
case 'object':
break
}
if (value === null) return null
const objectValue: object = value
if (seen.has(objectValue)) throw new MaterializeError(path, 'circular references are not JSON data')
seen.add(objectValue)
try {
if (Array.isArray(objectValue)) return materializeArray(objectValue, path, seen)
return materializeObject(objectValue, path, seen)
} finally {
seen.delete(objectValue)
}
}
function materializeArray(value: unknown[], path: string, seen: Set<object>): unknown[] {
const out: unknown[] = []
for (let index = 0; index < value.length; index++) {
const descriptor = Object.getOwnPropertyDescriptor(value, index)
if (descriptor === undefined) throw new MaterializeError(`${path}[${index}]`, 'sparse arrays are not JSON data')
if (!('value' in descriptor)) throw new MaterializeError(`${path}[${index}]`, 'accessor properties cannot cross the workflow realm boundary')
out.push(materialize(descriptor.value, `${path}[${index}]`, seen))
}
// Own enumerable props beyond the indices (e.g. `arr.total = 3`) would be
// silently dropped by JSON — reject them instead.
for (const key of Object.keys(value)) {
const index = Number(key)
if (!Number.isInteger(index) || index < 0 || index >= value.length) {
throw new MaterializeError(`${path}.${key}`, 'arrays with non-index properties are not JSON data')
}
}
if (Object.getOwnPropertySymbols(value).length > 0) {
throw new MaterializeError(path, 'symbol-keyed properties cannot cross the workflow realm boundary')
}
return out
}
function materializeObject(value: object, path: string, seen: Set<object>): Record<string, unknown> {
if (!hasPlainPrototype(value)) {
throw new MaterializeError(path, 'only plain objects and arrays are JSON data (exotic prototype)')
}
if (Object.getOwnPropertySymbols(value).length > 0) {
throw new MaterializeError(path, 'symbol-keyed properties cannot cross the workflow realm boundary')
}
const out: Record<string, unknown> = {}
for (const [key, descriptor] of Object.entries(Object.getOwnPropertyDescriptors(value))) {
// Non-enumerable own props never reach JSON output — skip them, matching
// JSON.stringify's contract exactly (documented in the module doc).
if (!descriptor.enumerable) continue
if (!('value' in descriptor)) {
throw new MaterializeError(`${path}.${key}`, 'accessor properties cannot cross the workflow realm boundary')
}
// defineProperty, never assignment: a "__proto__" key must become an OWN
// data property of the copy, not a prototype mutation.
Object.defineProperty(out, key, {
value: materialize(descriptor.value, `${path}.${key}`, seen),
enumerable: true,
writable: true,
configurable: true,
})
}
return out
}

499
packages/workflow/workflow-vm/src/runtime.ts Normal file
View File

@@ -0,0 +1,499 @@
/**
* Per-run execution state for the vm workflow engine: the script context and
* its injected hooks (`agent`/`parallel`/`pipeline`/`phase`/`log`/`args`), the
* concurrency semaphore and caps, cancellation, and the drive loop that turns
* a script settlement into a {@link WorkflowResult}.
*
* Realm discipline (see also ./realm.ts): values ENTERING the host from the
* script (hook options, schemas, the return value) are materialized via
* descriptor walks; values ENTERING the realm from the host (`args`, agent()
* results) are rebuilt INSIDE the realm through the context's own
* `JSON.parse`, so the script never holds an object whose prototype chain
* reaches host intrinsics. Realm functions (pipeline stages, parallel thunks)
* are called, not materialized — their values stay realm-side.
*
* Failure discipline: fatal {@link WorkflowError}s (bad hook arguments,
* unsupported options/schemas, tripped caps, seam start failures,
* cancellation) ALWAYS propagate through `parallel`/`pipeline`; the per-item
* `null` is reserved for child-run failures and ordinary in-stage script
* errors. Every hook-returned promise gets a no-op rejection consumer
* attached, so a script that drops a promise (fires an `agent()` without
* awaiting it) cannot surface an unhandled rejection when cancellation
* rejects it — the app boot layer exits the process on unhandled rejections.
*
* @module @deepseek-ai/dsh-workflow-vm/runtime
*/
import * as vm from 'node:vm'
import type { Context } from 'cordis'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { Agent } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-subagent'
import { assertSupportedOutputSchema, OutputSchemaError } from '@deepseek-ai/dsh-tools'
import type { StructuredOutputSchema } from '@deepseek-ai/dsh-tools'
import { WorkflowError, isFatalWorkflowError } from '@deepseek-ai/dsh-workflow'
import type {
WorkflowAgentEndInfo,
WorkflowAgentInfo,
WorkflowMeta,
WorkflowResult,
} from '@deepseek-ai/dsh-workflow'
import { materializeFromRealm, MaterializeError } from './realm.ts'
/** The per-run knobs the engine resolves from its Config. */
export interface ExecutionLimits {
/** The `ctx.subagents` provider name to start children on. */
provider: string
/** Concurrent `agent()` ceiling (already auto-resolved; ≥ 1). */
maxConcurrentAgents: number
/** Total `agent()` calls per run (the runaway-loop backstop). */
maxTotalAgents: number
/** Items accepted by one `parallel()`/`pipeline()` call. */
maxItemsPerCall: number
/** vm timeout for the script's initial synchronous slice. */
syncTimeoutMs: number
}
/** The engine-side observers the execution reports progress through. */
export interface ExecutionObserver {
phase(title: string): void
log(message: string): void
agentStart(info: WorkflowAgentInfo): void
agentEnd(info: WorkflowAgentEndInfo): void
}
/** The `agent()` options the script may pass; everything else rejects loud. */
const SUPPORTED_AGENT_OPTIONS = new Set(['label', 'phase', 'schema', 'model'])
/** Deferred Claude Code options we name explicitly in the rejection message. */
const DEFERRED_AGENT_OPTIONS = new Set(['effort', 'isolation', 'agentType'])
/** The in-context prelude that bans the nondeterminism sources (kept even though resume is deferred, so scripts stay resume-compatible). */
const DETERMINISM_PRELUDE = `
{
const banned = (name) => () => {
throw new Error(name + ' is not available in workflow scripts (runs must stay deterministic for future resume support; pass timestamps in via args)')
}
Math.random = banned('Math.random()')
Date.now = banned('Date.now()')
const RealDate = Date
globalThis.Date = new Proxy(RealDate, {
construct(target, args, newTarget) {
if (args.length === 0) banned('argless new Date()')()
return Reflect.construct(target, args, newTarget)
},
apply: banned('Date()'),
})
}
`
/** Flatten a child's final output blocks to text (the non-schema `agent()` result). */
function outputText(blocks: ContentBlock[]): string {
return blocks
.filter((block): block is Extract<ContentBlock, { type: 'text' }> => block.type === 'text')
.map(block => block.text)
.join('')
}
/**
* Render a script failure for the result: prefer the stack (it carries the
* script's own line numbers via the compile lineOffset), then the message.
* STRUCTURAL detection, not `instanceof Error` — a realm-thrown Error is not
* an instance of the host Error class.
*/
function errorText(error: unknown): string {
if (typeof error === 'object' && error !== null) {
const maybe = error as { stack?: unknown; message?: unknown }
if (typeof maybe.stack === 'string' && maybe.stack.length > 0) return maybe.stack
if (typeof maybe.message === 'string') return maybe.message
}
return String(error)
}
/** A short display label derived from the prompt when the script passes none. */
function defaultLabel(prompt: string): string {
const newline = prompt.indexOf('\n')
const line = newline === -1 ? prompt : prompt.slice(0, newline)
return line.length <= 48 ? line : `${line.slice(0, 47)}…`
}
/**
* One live script execution. Constructed per run by the engine; `drive()` is
* called exactly once and NEVER rejects — every failure becomes a
* {@link WorkflowResult} with a non-`completed` stop reason.
*/
export class WorkflowExecution {
/** 1-based count of `agent()` calls started (the `agentsStarted` result field). */
private started = 0
private activeSlots = 0
private readonly slotWaiters: { resolve(): void; reject(error: unknown): void }[] = []
private cancelReason: string | undefined
private cancelError: WorkflowError | undefined
private readonly controller = new AbortController()
private currentPhase: string | undefined
private readonly context: vm.Context
private readonly realmJsonParse: (text: string) => unknown
private readonly compiled: vm.Script
constructor(
private readonly ctx: Context,
meta: WorkflowMeta,
body: string,
private readonly parent: Agent,
args: unknown,
signal: AbortSignal | undefined,
private readonly limits: ExecutionLimits,
private readonly observer: ExecutionObserver,
) {
// Compile FIRST: a body syntax error must throw out of the constructor
// (the engine maps it to SCRIPT_PARSE) before any realm state exists.
// lineOffset compensates for the wrapper line, so stack traces carry the
// script's own line numbers (the meta statement was blanked, not removed).
try {
this.compiled = new vm.Script(`(async () => {\n${body}\n})()`, {
filename: `workflow:${meta.name}`,
lineOffset: -1,
})
} catch (error: unknown) {
throw new WorkflowError(`workflow script does not parse: ${String(error)}`, 'SCRIPT_PARSE', { cause: error })
}
this.context = vm.createContext({}, { name: `workflow:${meta.name}` })
vm.runInContext(DETERMINISM_PRELUDE, this.context)
// The realm's own JSON.parse — the host→realm rebuild channel.
const realmJson = vm.runInContext('JSON', this.context) as { parse(text: string): unknown }
this.realmJsonParse = (text: string) => realmJson.parse(text)
const globals: Record<string, unknown> = {
agent: (prompt: unknown, opts?: unknown) => this.contain(this.agent(prompt, opts)),
parallel: (thunks: unknown) => this.contain(this.parallel(thunks)),
pipeline: (items: unknown, ...stages: unknown[]) => this.contain(this.pipeline(items, stages)),
phase: (title: unknown) => { this.phase(title) },
log: (message: unknown) => { this.log(message) },
args: this.toRealm(args),
}
for (const [key, value] of Object.entries(globals)) {
// Data properties on the contextified global; frozen shape not required —
// a script overwriting its own hooks only sabotages itself.
;(this.context as Record<string, unknown>)[key] = typeof value === 'function' ? Object.freeze(value) : value
}
if (signal?.aborted) {
this.cancel('workflow start signal already aborted')
} else {
signal?.addEventListener('abort', () => { this.cancel('workflow signal aborted') }, { once: true })
}
}
/**
* Whether the run has been cancelled. A METHOD, not an inline property
* read: `cancel()` mutates `cancelReason` concurrently (a signal listener,
* a raced dispose), and an inline read after an `await` gets narrowed by
* control flow into an always-false comparison.
*/
private isCancelled(): boolean {
return this.cancelReason !== undefined
}
/**
* Cancel the run: children abort (the shared signal), waiting `agent()`
* slots reject, and every future hook call throws `CANCELLED` — the script
* dies at its next await. Idempotent; the first reason wins.
*/
cancel(reason?: string): void {
if (this.cancelReason !== undefined) return
this.cancelReason = reason ?? 'workflow cancelled'
this.cancelError = new WorkflowError(`workflow run cancelled: ${this.cancelReason}`, 'CANCELLED')
this.controller.abort(this.cancelReason)
for (const waiter of this.slotWaiters.splice(0)) waiter.reject(this.cancelledError())
}
/**
* Run the script to settlement. Resolves — never rejects — with the run's
* {@link WorkflowResult}: the materialized return value on `completed`, the
* failure message on `error`, and `cancelled` when the script died of
* cancellation. After settlement, any stray children a script fired without
* awaiting are aborted (their `agent()` wrappers dispose them).
*/
async drive(): Promise<WorkflowResult> {
try {
const scriptPromise = this.compiled.runInContext(this.context, { timeout: this.limits.syncTimeoutMs }) as Promise<unknown>
const raw: unknown = await this.contain(Promise.resolve(scriptPromise))
const value = raw === undefined ? null : this.materializeResult(raw)
return { value, stopReason: 'completed', agentsStarted: this.started }
} catch (error: unknown) {
if (error instanceof WorkflowError && error.code === 'CANCELLED') {
return { value: null, stopReason: 'cancelled', error: error.message, agentsStarted: this.started }
}
return { value: null, stopReason: 'error', error: errorText(error), agentsStarted: this.started }
} finally {
// Reap strays: a script that fired agent() calls without awaiting them
// leaves live children behind after settlement — abort them all. (The
// per-call wrappers dispose each child; the contain() consumer keeps
// their rejections from going unhandled.)
if (this.cancelReason === undefined) this.cancel('workflow settled')
}
}
/**
* Attach a no-op rejection consumer WITHOUT changing what the caller
* receives: if the script drops the promise (no await), cancellation cannot
* become an unhandled rejection (the app boot layer exits the process on
* those); if the script does await it, it still observes the rejection.
*/
private contain<T>(promise: Promise<T>): Promise<T> {
promise.catch(() => { /* consumed: see method contract — a dropped hook promise must not surface an unhandled rejection */ })
return promise
}
private cancelledError(): WorkflowError {
// cancel() arms cancelError before any caller can observe isCancelled()
// === true; the fallback guards the type, not a reachable path.
/* v8 ignore next */
return this.cancelError ?? new WorkflowError('workflow run cancelled', 'CANCELLED')
}
/** Rebuild a host value inside the script realm (via the realm's own JSON.parse). */
private toRealm(value: unknown): unknown {
if (value === undefined) return undefined
if (value === null) return null
if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return value
return this.realmJsonParse(JSON.stringify(value))
}
/** Materialize the script's return value; violations become RESULT_UNSERIALIZABLE. */
private materializeResult(raw: unknown): unknown {
try {
return materializeFromRealm(raw, 'workflow result')
} catch (error: unknown) {
/* v8 ignore next -- defensive rethrow arm: materializeFromRealm only throws MaterializeError */
if (!(error instanceof MaterializeError)) throw error
throw new WorkflowError(
`the workflow's return value is not plain JSON data — ${error.message}. Return only JSON-serializable objects/arrays/scalars.`,
'RESULT_UNSERIALIZABLE',
{ cause: error },
)
}
}
/**
* Acquire one concurrency slot (FIFO). Cancellation rejects QUEUED waiters
* (see {@link cancel}); the callers guard their own entry and post-acquire
* windows, so no cancelled-precheck is duplicated here.
*/
private acquireSlot(): Promise<void> {
if (this.activeSlots < this.limits.maxConcurrentAgents) {
this.activeSlots += 1
return Promise.resolve()
}
return new Promise<void>((resolve, reject) => {
this.slotWaiters.push({
resolve: () => {
this.activeSlots += 1
resolve()
},
reject,
})
})
}
private releaseSlot(): void {
this.activeSlots -= 1
const next = this.slotWaiters.shift()
if (next) next.resolve()
}
/** The `agent(prompt, opts)` hook. */
private async agent(rawPrompt: unknown, rawOpts: unknown): Promise<unknown> {
if (this.isCancelled()) throw this.cancelledError()
if (typeof rawPrompt !== 'string' || rawPrompt.length === 0) {
throw new WorkflowError('agent() requires a non-empty prompt string', 'INVALID_ARGUMENT')
}
const opts = this.readAgentOptions(rawOpts)
if (this.started >= this.limits.maxTotalAgents) {
throw new WorkflowError(
`this run reached its total agent cap (${this.limits.maxTotalAgents}) — a runaway-loop backstop; raise maxTotalAgents in the engine config if the scale is intentional`,
'AGENT_CAP',
)
}
this.started += 1
const seq = this.started
const label = opts.label ?? defaultLabel(rawPrompt)
const phase = opts.phase ?? this.currentPhase
await this.acquireSlot()
try {
// No cancelled re-check here: a cancel cannot interleave between a
// waiter's resolution and this continuation (single-threaded, no await
// between them), and a child started moments after a cancel still dies
// via the shared abort signal — the CANCELLED mapping below covers it.
let run
try {
run = this.ctx.subagents.start(this.limits.provider, {
prompt: [{ type: 'text', text: rawPrompt }],
parent: this.parent,
signal: this.controller.signal,
...opts.schema !== undefined ? { outputSchema: opts.schema } : {},
...opts.model !== undefined ? { agentOptions: { model: opts.model } } : {},
})
} catch (error: unknown) {
throw new WorkflowError(`agent() could not start a child on provider "${this.limits.provider}": ${String(error)}`, 'AGENT_START', { cause: error })
}
const info: WorkflowAgentInfo = { seq, label, ...phase !== undefined ? { phase } : {}, childId: run.id }
this.observer.agentStart(info)
try {
const result = await run.result
if (result.stopReason === 'completed') {
if (opts.schema !== undefined) {
// The provider honored outputSchema (capability-gated at start), so
// a completed run without a structured value is a child failure.
if (result.structured === undefined) {
this.observer.agentEnd({ ...info, outcome: 'failed' })
return null
}
this.observer.agentEnd({ ...info, outcome: 'completed' })
return this.toRealm(result.structured)
}
this.observer.agentEnd({ ...info, outcome: 'completed' })
return outputText(result.output)
}
// A cancelled RUN kills the script; a child that failed for its own
// reasons resolves null (scripts .filter(Boolean) per the CC contract).
if (this.isCancelled()) {
this.observer.agentEnd({ ...info, outcome: 'cancelled' })
throw this.cancelledError()
}
this.observer.agentEnd({ ...info, outcome: 'failed' })
return null
} finally {
await run.dispose()
}
} finally {
this.releaseSlot()
}
}
/** Materialize + validate the `agent()` options bag from the realm. */
private readAgentOptions(rawOpts: unknown): { label?: string; phase?: string; model?: string; schema?: StructuredOutputSchema } {
if (rawOpts === undefined) return {}
let opts: unknown
try {
opts = materializeFromRealm(rawOpts, 'agent() options')
} catch (error: unknown) {
/* v8 ignore next -- defensive rethrow arm: materializeFromRealm only throws MaterializeError */
if (!(error instanceof MaterializeError)) throw error
throw new WorkflowError(`agent() options must be plain JSON data — ${error.message}`, 'INVALID_ARGUMENT', { cause: error })
}
if (typeof opts !== 'object' || opts === null || Array.isArray(opts)) {
throw new WorkflowError('agent() options must be an object', 'INVALID_ARGUMENT')
}
const record = opts as Record<string, unknown>
for (const key of Object.keys(record)) {
if (SUPPORTED_AGENT_OPTIONS.has(key)) continue
if (DEFERRED_AGENT_OPTIONS.has(key)) {
throw new WorkflowError(`agent() option "${key}" is deferred and not supported by this engine (supported: label, phase, schema, model)`, 'UNSUPPORTED_OPTION')
}
throw new WorkflowError(`agent() option "${key}" is not recognized (supported: label, phase, schema, model)`, 'UNSUPPORTED_OPTION')
}
for (const key of ['label', 'phase', 'model'] as const) {
if (record[key] !== undefined && typeof record[key] !== 'string') {
throw new WorkflowError(`agent() option "${key}" must be a string`, 'INVALID_ARGUMENT')
}
}
let schema: StructuredOutputSchema | undefined
if (record.schema !== undefined) {
try {
assertSupportedOutputSchema(record.schema)
schema = record.schema
} catch (error: unknown) {
/* v8 ignore next -- defensive rethrow arm: assertSupportedOutputSchema only throws OutputSchemaError */
if (!(error instanceof OutputSchemaError)) throw error
throw new WorkflowError(`agent() schema is outside the supported subset — ${error.message}`, 'UNSUPPORTED_SCHEMA', { cause: error })
}
}
return {
...record.label !== undefined ? { label: record.label as string } : {},
...record.phase !== undefined ? { phase: record.phase as string } : {},
...record.model !== undefined ? { model: record.model as string } : {},
...schema !== undefined ? { schema } : {},
}
}
/** The `parallel(thunks)` hook: each thunk caught → `null`; fatal errors propagate. */
private async parallel(rawThunks: unknown): Promise<unknown[]> {
if (!Array.isArray(rawThunks)) {
throw new WorkflowError('parallel() requires an array of zero-argument functions', 'INVALID_ARGUMENT')
}
this.assertItemCap(rawThunks.length, 'parallel()')
const thunks = rawThunks.map((thunk, index) => {
if (typeof thunk !== 'function') {
throw new WorkflowError(`parallel() item ${index} is not a function`, 'INVALID_ARGUMENT')
}
return thunk as () => unknown
})
return Promise.all(thunks.map(async (thunk) => {
try {
return await thunk()
} catch (error: unknown) {
if (isFatalWorkflowError(error)) throw error
return null
}
}))
}
/** The `pipeline(items, ...stages)` hook: per-item stage chains, NO cross-stage barrier. */
private async pipeline(rawItems: unknown, rawStages: unknown[]): Promise<unknown[]> {
if (!Array.isArray(rawItems)) {
throw new WorkflowError('pipeline() requires an items array', 'INVALID_ARGUMENT')
}
this.assertItemCap(rawItems.length, 'pipeline()')
if (rawStages.length === 0) {
throw new WorkflowError('pipeline() requires at least one stage function', 'INVALID_ARGUMENT')
}
const stages = rawStages.map((stage, index) => {
if (typeof stage !== 'function') {
throw new WorkflowError(`pipeline() stage ${index} is not a function`, 'INVALID_ARGUMENT')
}
return stage as (previous: unknown, item: unknown, index: number) => unknown
})
return Promise.all(rawItems.map(async (item: unknown, index) => {
let value: unknown = item
try {
for (const stage of stages) {
value = await stage(value, item, index)
}
return value
} catch (error: unknown) {
// An ordinary stage throw drops the ITEM to null and skips its
// remaining stages; a fatal error kills the whole script.
if (isFatalWorkflowError(error)) throw error
return null
}
}))
}
private assertItemCap(length: number, hook: string): void {
if (length > this.limits.maxItemsPerCall) {
throw new WorkflowError(
`${hook} received ${length} items — over the per-call cap (${this.limits.maxItemsPerCall}); split the work or raise maxItemsPerCall in the engine config`,
'ITEM_CAP',
)
}
}
/** The `phase(title)` hook: sets the current label for subsequent `agent()` calls and notifies observers. */
private phase(title: unknown): void {
if (typeof title !== 'string' || title.length === 0) {
throw new WorkflowError('phase() requires a non-empty title string', 'INVALID_ARGUMENT')
}
this.currentPhase = title
this.observer.phase(title)
}
/** The `log(message)` hook: narration to observers. */
private log(message: unknown): void {
if (typeof message !== 'string') {
throw new WorkflowError('log() requires a message string', 'INVALID_ARGUMENT')
}
this.observer.log(message)
}
}

89
packages/workflow/workflow-vm/tests/integration.spec.ts Normal file
View File

@@ -0,0 +1,89 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import * as Invariants from '@deepseek-ai/dsh-invariants'
import SubagentService from '@deepseek-ai/dsh-subagent'
import * as spawn from '@deepseek-ai/dsh-subagent-spawn'
import { STRUCTURED_OUTPUT_TOOL } from '@deepseek-ai/dsh-subagent-inprocess'
import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import VmWorkflowEngine from '../src/index.ts'
type Script = ConstructorParameters<typeof MockAdapter>[0]
/**
* The whole in-process stack, keyless: the vm engine drives the REAL spawn
* backend (with its structured runtime) on a real agent loop; the scripted
* mock MODEL is the only mocked boundary. This is the integration guard the
* per-hook unit tests (which stub the subagent seam) structurally cannot give.
*/
async function setup(script: Script) {
const ctx = new Context()
const adapter = new MockAdapter(script)
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(Invariants)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
await ctx.plugin(spawn, { providerName: 'spawn', structuredNudgeRetries: 1 })
await ctx.plugin(VmWorkflowEngine, {})
ctx.llm.registerAdapter(['mock'], adapter)
const parent = ctx.agentLoop.create(AgentId('parent'), { model: 'mock' })
return { ctx, parent, adapter }
}
describe('dsh-workflow-vm over the real in-process stack', () => {
it('runs a two-stage workflow: a plain child, then a schema child through the structured runtime', async () => {
const { ctx, parent } = await setup([
textResponse('the file list is a.ts'),
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { verdict: 'real', confidence: 0.9 }),
])
const childIds: string[] = []
ctx.on('workflow/agent-start', (_info, agent) => { childIds.push(agent.childId) })
const run = ctx.workflows.start({
script: `export const meta = { name: 'integration', description: 'plain + structured children' }
phase('Read')
const prose = await agent('read the repo')
phase('Judge')
const judged = await agent('judge: ' + prose, {
schema: { type: 'object', properties: { verdict: { type: 'string', enum: ['real', 'bogus'] }, confidence: { type: 'number' } }, required: ['verdict'] },
})
return { prose, verdict: judged.verdict, confidence: judged.confidence }`,
parent,
})
const result = await run.result
expect(result.stopReason).toBe('completed')
expect(result.value).toEqual({ prose: 'the file list is a.ts', verdict: 'real', confidence: 0.9 })
expect(result.agentsStarted).toBe(2)
await run.dispose()
// Both children were disposed to quiescence — no live child agents remain.
expect(childIds.length).toBe(2)
for (const childId of childIds) {
expect(ctx.agents.get(AgentId(childId))).toBeUndefined()
}
})
it('a child that fails against its schema (nudges exhausted) reaches the script as null', async () => {
const { ctx, parent } = await setup([
textResponse('prose only'),
textResponse('still prose after the nudge'),
])
const run = ctx.workflows.start({
script: `export const meta = { name: 'null-path', description: 'schema failure maps to null' }
const judged = await agent('judge it', { schema: { type: 'object', properties: { v: { type: 'string' } } } })
return { got: judged === null ? 'null' : 'value' }`,
parent,
})
const result = await run.result
expect(result.stopReason).toBe('completed')
expect(result.value).toEqual({ got: 'null' })
await run.dispose()
})
})

143
packages/workflow/workflow-vm/tests/meta.spec.ts Normal file
View File

@@ -0,0 +1,143 @@
import { describe, expect, it } from 'vitest'
import { WorkflowError } from '@deepseek-ai/dsh-workflow'
import { extractMeta } from '../src/meta.ts'
const TIMEOUT = 1000
/** Extract and expect success. */
function ok(script: string) {
return extractMeta(script, TIMEOUT)
}
/** The WorkflowError a bad script produces (throws if it extracts cleanly). */
function bad(script: string): WorkflowError {
try {
extractMeta(script, TIMEOUT)
} catch (error: unknown) {
if (error instanceof WorkflowError) return error
throw error
}
throw new Error('expected extraction to fail')
}
describe('extractMeta', () => {
it('extracts a full meta block and blanks the statement line-preservingly', () => {
const script = `export const meta = {
name: 'audit-routes',
description: 'Audit every route',
whenToUse: 'when auditing',
phases: [{ title: 'Scan', detail: 'find files' }, { title: 'Fix', model: 'deepseek-v4-pro' }],
}
const x = 1
return x`
const { meta, body } = ok(script)
expect(meta).toEqual({
name: 'audit-routes',
description: 'Audit every route',
whenToUse: 'when auditing',
phases: [{ title: 'Scan', detail: 'find files' }, { title: 'Fix', model: 'deepseek-v4-pro' }],
})
// Same line count; the statement's characters blanked; the body intact.
expect(body.split('\n').length).toBe(script.split('\n').length)
expect(body.split('\n')[6]).toBe('const x = 1')
expect(body).not.toContain('export')
})
it('allows leading line and block comments before the meta statement', () => {
const script = `// a workflow
/* multi
line */
export const meta = { name: 'x', description: 'y' }
return 1`
expect(ok(script).meta.name).toBe('x')
})
it('handles braces inside strings and comments while scanning', () => {
const script = `export const meta = {
name: 'tricky', // } not a close {
/* } also not } */
description: "has { braces } and 'quotes'",
}
return 2`
expect(ok(script).meta.description).toBe("has { braces } and 'quotes'")
})
it('tolerates template-quoted strings WITHOUT interpolation, escapes included', () => {
const script = 'export const meta = { name: `plain`, description: `esc \\` tick` }\nreturn 1'
expect(ok(script).meta.name).toBe('plain')
})
it('consumes a trailing semicolon after the literal, spaces included', () => {
const { body } = ok("export const meta = { name: 'x', description: 'y' };\nreturn 1")
expect(body).not.toContain(';')
expect(body.split('\n')[1]).toBe('return 1')
const spaced = ok("export const meta = { name: 'x', description: 'y' } ;\nreturn 1")
expect(spaced.body).not.toContain(';')
})
it('rejects a script that does not begin with the meta statement (SCRIPT_PARSE)', () => {
expect(bad('const a = 1').code).toBe('SCRIPT_PARSE')
expect(bad('').code).toBe('SCRIPT_PARSE')
expect(bad('export const meta = [1]').code).toBe('SCRIPT_PARSE')
})
it('rejects template interpolation in the meta block as impure (SCRIPT_PARSE)', () => {
const error = bad('export const meta = { name: `w-${1}`, description: "d" }\nreturn 1')
expect(error.code).toBe('SCRIPT_PARSE')
expect(error.message).toContain('pure literal')
})
it('rejects unbalanced literals, unterminated strings, and unterminated comments (SCRIPT_PARSE)', () => {
expect(bad('export const meta = { name: "x", description: "y"').code).toBe('SCRIPT_PARSE')
expect(bad('export const meta = { name: "x').code).toBe('SCRIPT_PARSE')
expect(bad('export const meta = { /* open').code).toBe('SCRIPT_PARSE')
// A line comment running to EOF (no newline) leaves the literal unbalanced.
expect(bad('export const meta = { name: "x" // eof comment').code).toBe('SCRIPT_PARSE')
})
it('rejects a literal referencing variables or calls (META_INVALID via the empty realm)', () => {
const error = bad('export const meta = { name: someVariable, description: "d" }\nreturn 1')
expect(error.code).toBe('META_INVALID')
expect(error.message).toContain('pure literal')
expect(bad('export const meta = { name: compute(), description: "d" }').code).toBe('META_INVALID')
})
it('rejects a literal evaluating to non-JSON data (META_INVALID via materialization)', () => {
const error = bad('export const meta = { name: "x", description: "d", phases: [{ get title() { return "t" } }] }')
expect(error.code).toBe('META_INVALID')
expect(error.message).toContain('JSON data')
})
it('rejects shape violations with EVERY violation listed (META_INVALID)', () => {
const error = bad('export const meta = { description: 7, bogus: 1 }\nreturn 1')
expect(error.code).toBe('META_INVALID')
expect(error.message).toContain('meta.name must be a non-empty string')
expect(error.message).toContain('meta.description must be a non-empty string')
expect(error.message).toContain('meta.bogus is not a recognized field')
})
it('rejects malformed whenToUse and phases shapes precisely', () => {
expect(bad('export const meta = { name: "x", description: "d", whenToUse: 3 }').message)
.toContain('meta.whenToUse must be a string')
expect(bad('export const meta = { name: "x", description: "d", phases: "no" }').message)
.toContain('meta.phases must be an array')
expect(bad('export const meta = { name: "x", description: "d", phases: [3] }').message)
.toContain('meta.phases[0] must be an object')
expect(bad('export const meta = { name: "x", description: "d", phases: [{}] }').message)
.toContain('meta.phases[0].title must be a non-empty string')
expect(bad('export const meta = { name: "x", description: "d", phases: [{ title: "t", extra: 1 }] }').message)
.toContain('meta.phases[0].extra is not a recognized field')
expect(bad('export const meta = { name: "x", description: "d", phases: [{ title: "t", detail: 1 }] }').message)
.toContain('meta.phases[0].detail must be a string')
expect(bad('export const meta = { name: "x", description: "d", phases: [{ title: "t", model: 1 }] }').message)
.toContain('meta.phases[0].model must be a string')
})
it('stops scanning at the balanced literal — trailing expression text stays in the body', () => {
// The scanner extracts exactly `{ valueOf: null }`; the ` && 3` is body
// text (which would fail compilation later, but extraction sees only the
// literal and reports its unknown field).
expect(bad('export const meta = { valueOf: null } && 3').message)
.toContain('meta.valueOf is not a recognized field')
})
})

114
packages/workflow/workflow-vm/tests/realm.spec.ts Normal file
View File

@@ -0,0 +1,114 @@
import { describe, expect, it } from 'vitest'
import * as vm from 'node:vm'
import { materializeFromRealm, MaterializeError } from '../src/realm.ts'
/** Evaluate an expression inside a fresh vm realm and hand back the raw realm value. */
function inRealm(expression: string): unknown {
return vm.runInNewContext(`(${expression})`)
}
/** The MaterializeError message for a value that must be rejected (throws if accepted). */
function rejection(value: unknown): string {
try {
materializeFromRealm(value)
} catch (error: unknown) {
if (error instanceof MaterializeError) return error.message
throw error
}
throw new Error('expected the value to be rejected')
}
describe('materializeFromRealm', () => {
it('copies realm objects/arrays/scalars into host plain data', () => {
const value = inRealm("{ a: 1, b: 'x', c: true, d: null, list: [1, [2, { deep: 'y' }]] }")
const out = materializeFromRealm(value) as Record<string, unknown>
expect(out).toEqual({ a: 1, b: 'x', c: true, d: null, list: [1, [2, { deep: 'y' }]] })
// The copy is HOST data: prototypes are the host intrinsics.
expect(Object.getPrototypeOf(out)).toBe(Object.prototype)
expect(Array.isArray(out.list)).toBe(true)
// And it round-trips through JSON byte-identically (the whole point).
expect(JSON.parse(JSON.stringify(out))).toEqual(out)
})
it('accepts undefined ONLY at the root (a valueless script return)', () => {
expect(materializeFromRealm(undefined)).toBeUndefined()
expect(rejection(inRealm('{ a: undefined }'))).toContain('value.a')
})
it('never invokes accessors: a counting getter is rejected, not read', () => {
const counter = inRealm(`
(() => {
globalThis.reads = 0
return { get x() { globalThis.reads += 1; return 1 } }
})()
`)
expect(rejection(counter)).toContain('accessor properties cannot cross')
// The getter body never ran — descriptor inspection only.
expect((counter as { x?: unknown }).x).toBe(1) // sanity: reading DOES run it…
expect(rejection(counter)).toContain('accessor') // …but materialization still never did
})
it('a "__proto__" key becomes an OWN data property of the copy, never a prototype mutation', () => {
const value: unknown = vm.runInNewContext('JSON.parse(\'{"__proto__": {"polluted": 1}, "ok": 2}\')')
const out = materializeFromRealm(value) as Record<string, unknown>
expect(Object.getPrototypeOf(out)).toBe(Object.prototype)
expect(Object.prototype.hasOwnProperty.call(out, '__proto__')).toBe(true)
expect(out.ok).toBe(2)
// The host Object.prototype was NOT touched.
expect(({} as Record<string, unknown>).polluted).toBeUndefined()
})
it('rejects functions, symbols (keys and values), and bigints with path-qualified messages', () => {
expect(rejection(inRealm('{ fn: () => 1 }'))).toContain('value.fn')
expect(rejection(inRealm("{ [Symbol('k')]: 1 }"))).toContain('symbol-keyed')
expect(rejection(inRealm("{ s: Symbol('v') }"))).toContain('value.s')
expect(rejection(inRealm('{ big: 1n }'))).toContain('value.big')
expect(rejection(inRealm("[Symbol('x')]"))).toContain('value[0]')
const taggedArray = inRealm("(() => { const a = [1]; a[Symbol('t')] = 1; return a })()")
expect(rejection(taggedArray)).toContain('symbol-keyed')
})
it('rejects non-finite numbers and undefined values inside containers', () => {
expect(rejection(inRealm('{ n: NaN }'))).toContain('non-finite')
expect(rejection(inRealm('[Infinity]'))).toContain('non-finite')
})
it('rejects exotic prototypes (Date, Map, class instances) but accepts null-prototype data', () => {
expect(rejection(inRealm('{ d: new Date(0) }'))).toContain('exotic prototype')
expect(rejection(inRealm('new Map()'))).toContain('exotic prototype')
expect(rejection(inRealm('(() => { class C { constructor() { this.x = 1 } } return new C() })()')))
.toContain('exotic prototype')
expect(materializeFromRealm(inRealm('Object.assign(Object.create(null), { a: 1 })'))).toEqual({ a: 1 })
})
it('rejects cycles and accepts the same object reused as a sibling (a DAG)', () => {
expect(rejection(inRealm('(() => { const o = {}; o.self = o; return o })()'))).toContain('circular')
const dag = inRealm('(() => { const leaf = { v: 1 }; return { a: leaf, b: leaf } })()')
expect(materializeFromRealm(dag)).toEqual({ a: { v: 1 }, b: { v: 1 } })
})
it('rejects sparse arrays, accessor elements, and non-index array properties', () => {
expect(rejection(inRealm('[1, , 3]'))).toContain('sparse')
expect(rejection(inRealm('(() => { const a = [1]; Object.defineProperty(a, 0, { get: () => 1 }); return a })()')))
.toContain('accessor')
expect(rejection(inRealm('(() => { const a = [1]; a.total = 3; return a })()')))
.toContain('non-index')
})
it('skips non-enumerable own properties (matching JSON.stringify exactly)', () => {
const value = inRealm(`(() => {
const o = { visible: 1 }
Object.defineProperty(o, 'hidden', { value: () => 1, enumerable: false })
return o
})()`)
expect(materializeFromRealm(value)).toEqual({ visible: 1 })
})
it('works on plain host values too (the boundary is realm-agnostic)', () => {
expect(materializeFromRealm({ a: [1, 'x'] })).toEqual({ a: [1, 'x'] })
expect(materializeFromRealm('str')).toBe('str')
expect(materializeFromRealm(3)).toBe(3)
expect(materializeFromRealm(false)).toBe(false)
expect(materializeFromRealm(null)).toBeNull()
})
})

614
packages/workflow/workflow-vm/tests/workflow-vm.spec.ts Normal file
View File

@@ -0,0 +1,614 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { AgentId } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentCapabilities, SubagentProvider, SubagentResult, SubagentRun, SubagentStartRequest } from '@deepseek-ai/dsh-subagent'
import type { WorkflowResult, WorkflowRunInfo } from '@deepseek-ai/dsh-workflow'
import * as vmEngineModule from '../src/index.ts'
import VmWorkflowEngine, { type Config } from '../src/index.ts'
/** A minimal parent stand-in: the engine only threads it through to the provider. */
function fakeParent(): Agent {
return { id: AgentId('workflow-parent'), options: {} } as unknown as Agent
}
/** One controllable child run: the test (or auto mode) settles it. */
interface ControlledRun {
request: SubagentStartRequest
settle(result: SubagentResult): void
cancelled: string | undefined
disposed: boolean
}
/**
* A scripted in-test provider over the REAL SubagentService registry: `auto`
* settles each run via the reply function on a microtask; `manual` piles runs
* up in `runs` for the test to settle (concurrency/cancellation tests). A run
* aborts (settles `aborted`) when the request signal fires, like the real
* in-process backends.
*/
class StubProvider implements SubagentProvider {
readonly capabilities: SubagentCapabilities = { outputSchema: true, depthLimit: true, toolFilter: true }
readonly runs: ControlledRun[] = []
constructor(
readonly name: string,
private readonly reply?: (request: SubagentStartRequest, index: number) => SubagentResult,
) {}
start(request: SubagentStartRequest): SubagentRun {
let settle!: (result: SubagentResult) => void
const result = new Promise<SubagentResult>((resolve) => { settle = resolve })
const controlled: ControlledRun = { request, settle, cancelled: undefined, disposed: false }
this.runs.push(controlled)
const index = this.runs.length - 1
request.signal?.addEventListener('abort', () => { settle({ output: [], stopReason: 'aborted' }) }, { once: true })
if (this.reply) {
const reply = this.reply
queueMicrotask(() => { settle(reply(request, index)) })
}
return {
id: AgentId(`stub-child-${index}`),
result,
cancel: (reason?: string) => {
controlled.cancelled = reason ?? 'cancelled'
settle({ output: [], stopReason: 'aborted' })
},
dispose: () => {
controlled.disposed = true
return Promise.resolve()
},
}
}
}
/** Text-reply helper for auto providers. */
function text(reply: string): SubagentResult {
return { output: [{ type: 'text', text: reply }], stopReason: 'completed' }
}
interface SetupOptions {
config?: Config
reply?: (request: SubagentStartRequest, index: number) => SubagentResult
manual?: boolean
}
async function setup(options?: SetupOptions) {
const ctx = new Context()
await ctx.plugin(SubagentService)
const provider = new StubProvider('stub', options?.manual ? undefined : options?.reply ?? (() => text('stub reply')))
ctx.subagents.registerProvider(provider)
await ctx.plugin(VmWorkflowEngine, { provider: 'stub', ...options?.config })
return { ctx, provider, parent: fakeParent() }
}
/** Wrap a body in the minimal valid meta header. */
function script(body: string, metaExtra = ''): string {
return `export const meta = { name: 'test-flow', description: 'a test workflow'${metaExtra} }\n${body}`
}
/** Start + await one run, disposing on the way out. */
async function run(ctx: Context, parent: Agent, source: string, args?: unknown): Promise<WorkflowResult> {
const handle = ctx.workflows.start({ script: source, parent, ...args !== undefined ? { args } : {} })
try {
return await handle.result
} finally {
await handle.dispose()
}
}
describe('dsh-workflow-vm', () => {
describe('script execution', () => {
it('runs a script end-to-end: agent() text results, phases, log, args, return value', async () => {
const { ctx, parent, provider } = await setup({ reply: (_request, index) => text(`answer-${index}`) })
const events: [string, unknown[]][] = []
for (const name of ['workflow/start', 'workflow/phase', 'workflow/log', 'workflow/agent-start', 'workflow/agent-end', 'workflow/end'] as const) {
ctx.on(name, (...payload: unknown[]) => { events.push([name, payload]) })
}
const result = await run(ctx, parent, script(`
phase('Scan')
log('starting with ' + args.files.length + ' files')
const answers = await pipeline(args.files, (prev, item) => agent('read ' + item))
phase('Report')
return { answers, count: args.files.length }
`, ", phases: [{ title: 'Scan' }, { title: 'Report' }]"), { files: ['a.ts', 'b.ts'] })
expect(result.stopReason).toBe('completed')
expect(result.agentsStarted).toBe(2)
expect(result.value).toEqual({ answers: ['answer-0', 'answer-1'], count: 2 })
expect(provider.runs.every(r => r.disposed)).toBe(true)
const names = events.map(([name]) => name)
expect(names[0]).toBe('workflow/start')
expect(names).toContain('workflow/phase')
expect(names).toContain('workflow/log')
expect(names.at(-1)).toBe('workflow/end')
const info = events[0]![1][0] as WorkflowRunInfo
expect(info.meta.name).toBe('test-flow')
const end = events.at(-1)![1][1] as Record<string, unknown>
expect(end).toEqual({ stopReason: 'completed', agentsStarted: 2 })
expect('value' in end).toBe(false)
})
it('agent-start/end events carry seq, label (defaulted from the prompt), phase, and outcome', async () => {
const { ctx, parent } = await setup()
const starts: unknown[] = []
const ends: unknown[] = []
ctx.on('workflow/agent-start', (_info, agent) => starts.push(agent))
ctx.on('workflow/agent-end', (_info, agent) => ends.push(agent))
await run(ctx, parent, script(`
phase('Find')
await agent('a prompt that is quite long and will surely get truncated down to a display label\\n'
+ 'with a second line the label must not include')
await agent('short', { label: 'named', phase: 'Custom' })
return null
`))
expect(starts[0]).toMatchObject({ seq: 1, phase: 'Find', childId: 'stub-child-0' })
expect((starts[0] as { label: string }).label.length).toBeLessThanOrEqual(48)
expect((starts[0] as { label: string }).label).not.toContain('second line')
expect(starts[1]).toMatchObject({ seq: 2, label: 'named', phase: 'Custom' })
expect(ends[0]).toMatchObject({ seq: 1, outcome: 'completed' })
})
it('agent({schema}) forwards outputSchema to the provider and returns the structured value into the realm', async () => {
const { ctx, parent, provider } = await setup({
reply: () => ({ output: [], structured: { files: ['x.ts', 'y.ts'] }, stopReason: 'completed' }),
})
const result = await run(ctx, parent, script(`
const found = await agent('list files', { schema: { type: 'object', properties: { files: { type: 'array', items: { type: 'string' } } }, required: ['files'] } })
return { first: found.files[0], count: found.files.length }
`))
expect(result.value).toEqual({ first: 'x.ts', count: 2 })
expect(provider.runs[0]!.request.outputSchema).toEqual({
type: 'object',
properties: { files: { type: 'array', items: { type: 'string' } } },
required: ['files'],
})
})
it('model option maps to agentOptions.model on the start request', async () => {
const { ctx, parent, provider } = await setup()
await run(ctx, parent, script("return await agent('p', { model: 'deepseek-v4-pro' })"))
expect(provider.runs[0]!.request.agentOptions).toEqual({ model: 'deepseek-v4-pro' })
})
it('a failed child resolves null (scripts filter), never throwing into the script', async () => {
const { ctx, parent } = await setup({
reply: (_request, index) => index === 0 ? { output: [], stopReason: 'error' } : text('ok'),
})
const result = await run(ctx, parent, script(`
const results = await parallel([() => agent('one'), () => agent('two')])
return results
`))
expect(result.value).toEqual([null, 'ok'])
})
it('a schema run that completes WITHOUT a structured value is a child failure (null + failed outcome)', async () => {
const { ctx, parent } = await setup({ reply: () => text('prose, no structure') })
const ends: unknown[] = []
ctx.on('workflow/agent-end', (_info, agent) => ends.push(agent))
const result = await run(ctx, parent, script(`
return await agent('p', { schema: { type: 'object' } })
`))
expect(result.value).toBeNull()
expect(ends[0]).toMatchObject({ outcome: 'failed' })
})
it('a script with no return value resolves value: null', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script("await agent('p')"))
expect(result.stopReason).toBe('completed')
expect(result.value).toBeNull()
})
})
describe('combinator semantics', () => {
it('pipeline has NO cross-stage barrier: a fast item finishes stage 2 while a slow item holds stage 1', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const handle = ctx.workflows.start({
script: script(`
const out = await pipeline(['slow', 'fast'],
(prev, item) => agent('s1 ' + item),
(prev, item) => agent('s2 ' + item + ' after ' + prev),
)
return out
`),
parent: fakeParent(),
})
// Both items enter stage 1 concurrently.
await vi.waitFor(() => { expect(provider.runs.length).toBe(2) })
// Settle only the FAST item's stage 1 → its stage 2 starts with no barrier.
provider.runs[1]!.settle(text('fast-1'))
await vi.waitFor(() => { expect(provider.runs.length).toBe(3) })
expect((provider.runs[2]!.request.prompt[0] as { text: string }).text).toBe('s2 fast after fast-1')
// The slow item is still sitting in stage 1.
provider.runs[2]!.settle(text('fast-2'))
provider.runs[0]!.settle(text('slow-1'))
await vi.waitFor(() => { expect(provider.runs.length).toBe(4) })
provider.runs[3]!.settle(text('slow-2'))
const result = await handle.result
expect(result.value).toEqual(['slow-2', 'fast-2'])
await handle.dispose()
void parent
})
it('pipeline stage callbacks receive (prev, item, index); an ordinary stage throw nulls the ITEM and skips its remaining stages', async () => {
const { ctx, parent, provider } = await setup({ reply: request => text(`ok:${(request.prompt[0] as { text: string }).text}`) })
const result = await run(ctx, parent, script(`
const out = await pipeline([10, 20],
(prev, item, index) => {
if (item === 10) throw new Error('ordinary failure')
return agent('stage1-' + item + '-' + index)
},
(prev) => agent('stage2 saw ' + prev),
)
return out
`))
expect(result.stopReason).toBe('completed')
const prompts = provider.runs.map(r => (r.request.prompt[0] as { text: string }).text)
// Item 10 never reached stage 1's agent nor stage 2.
expect(prompts).toEqual(['stage1-20-1', 'stage2 saw ok:stage1-20-1'])
expect(result.value).toEqual([null, 'ok:stage2 saw ok:stage1-20-1'])
})
it('parallel maps a throwing thunk to null and never rejects for ordinary errors', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script(`
return await parallel([
() => { throw new Error('boom') },
() => agent('fine'),
() => 'plain value',
])
`))
expect(result.value).toEqual([null, 'stub reply', 'plain value'])
})
it('FATAL errors propagate through parallel AND pipeline instead of dissolving into null', async () => {
const { ctx, parent } = await setup()
const viaParallel = await run(ctx, parent, script(`
return await parallel([() => agent('x', { isolation: 'worktree' })])
`))
expect(viaParallel.stopReason).toBe('error')
expect(viaParallel.error).toContain('"isolation" is deferred')
const viaPipeline = await run(ctx, parent, script(`
return await pipeline([1], () => agent('x', { bogus: true }))
`))
expect(viaPipeline.stopReason).toBe('error')
expect(viaPipeline.error).toContain('"bogus" is not recognized')
})
it('validates combinator arguments loudly (non-array, non-function, missing stages)', async () => {
const { ctx, parent } = await setup()
expect((await run(ctx, parent, script("return await parallel('no')"))).error).toContain('parallel() requires an array')
expect((await run(ctx, parent, script('return await parallel([3])'))).error).toContain('item 0 is not a function')
expect((await run(ctx, parent, script("return await pipeline('no', () => 1)"))).error).toContain('pipeline() requires an items array')
expect((await run(ctx, parent, script('return await pipeline([1])'))).error).toContain('at least one stage')
expect((await run(ctx, parent, script("return await pipeline([1], 'x')"))).error).toContain('stage 0 is not a function')
})
})
describe('caps and option validation', () => {
it('trips the total-agent cap with a message naming the config knob', async () => {
const { ctx, parent } = await setup({ config: { provider: 'stub', maxTotalAgents: 2 } })
const result = await run(ctx, parent, script(`
await agent('1'); await agent('2'); await agent('3')
return 'unreachable'
`))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('total agent cap (2)')
expect(result.error).toContain('maxTotalAgents')
expect(result.agentsStarted).toBe(2)
})
it('trips the per-call item cap for parallel and pipeline', async () => {
const { ctx, parent } = await setup({ config: { provider: 'stub', maxItemsPerCall: 2 } })
expect((await run(ctx, parent, script('return await parallel([() => 1, () => 2, () => 3])'))).error)
.toContain('over the per-call cap (2)')
expect((await run(ctx, parent, script('return await pipeline([1, 2, 3], (x) => x)'))).error)
.toContain('maxItemsPerCall')
})
it('enforces the concurrency ceiling: never more than maxConcurrentAgents children in flight', async () => {
const { ctx, parent, provider } = await setup({ manual: true, config: { provider: 'stub', maxConcurrentAgents: 2 } })
const handle = ctx.workflows.start({
script: script("return await parallel([1, 2, 3, 4, 5].map((n) => () => agent('job ' + n)))"),
parent,
})
// Only 2 children may exist until one settles.
await vi.waitFor(() => { expect(provider.runs.length).toBe(2) })
await new Promise(resolve => setTimeout(resolve, 20))
expect(provider.runs.length).toBe(2)
// Settle children in arrival order; after each settle at most ONE more
// child may enter — the window never exceeds the ceiling.
for (let index = 0; index < 5; index++) {
await vi.waitFor(() => { expect(provider.runs.length).toBeGreaterThan(index) })
expect(provider.runs.length).toBeLessThanOrEqual(Math.min(index + 2, 5))
provider.runs[index]!.settle(text(`r${index}`))
}
const result = await handle.result
expect(result.stopReason).toBe('completed')
expect(result.agentsStarted).toBe(5)
expect(result.value).toEqual(['r0', 'r1', 'r2', 'r3', 'r4'])
await handle.dispose()
})
it('rejects malformed agent() arguments and option types loudly', async () => {
const { ctx, parent } = await setup()
expect((await run(ctx, parent, script('return await agent(42)'))).error).toContain('non-empty prompt string')
expect((await run(ctx, parent, script("return await agent('')"))).error).toContain('non-empty prompt string')
expect((await run(ctx, parent, script("return await agent('p', 'opts')"))).error).toContain('options must be an object')
expect((await run(ctx, parent, script("return await agent('p', { label: 3 })"))).error).toContain('"label" must be a string')
expect((await run(ctx, parent, script("return await agent('p', { effort: 'high' })"))).error).toContain('"effort" is deferred')
})
it('rejects options that are not plain JSON data (an accessor smuggled into opts)', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script("return await agent('p', { get label() { return 'x' } })"))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('options must be plain JSON data')
})
it('validates phase() and log() arguments loudly', async () => {
const { ctx, parent } = await setup()
expect((await run(ctx, parent, script('phase(3)'))).error).toContain('phase() requires a non-empty title string')
expect((await run(ctx, parent, script("phase('')"))).error).toContain('phase() requires a non-empty title string')
expect((await run(ctx, parent, script('log(3)'))).error).toContain('log() requires a message string')
})
it('rejects an unsupported schema via the shared subset assertion (UNSUPPORTED_SCHEMA)', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script("return await agent('p', { schema: { type: 'object', oneOf: [] } })"))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('outside the supported subset')
expect(result.error).toContain('oneOf')
})
it('wraps a provider start failure as a fatal AGENT_START error (a missing provider cannot dissolve into null)', async () => {
const { ctx, parent } = await setup({ config: { provider: 'nonexistent' } })
const result = await run(ctx, parent, script("return await pipeline([1], () => agent('p'))"))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('could not start a child on provider "nonexistent"')
})
})
describe('determinism bans and realm isolation', () => {
it('Date.now, Math.random, and argless new Date throw; parameterized Date stays usable', async () => {
const { ctx, parent } = await setup()
expect((await run(ctx, parent, script('return Date.now()'))).error).toContain('Date.now() is not available')
expect((await run(ctx, parent, script('return Math.random()'))).error).toContain('Math.random() is not available')
expect((await run(ctx, parent, script('return new Date().toISOString()'))).error).toContain('argless new Date()')
const ok = await run(ctx, parent, script('return new Date(0).getTime()'))
expect(ok.value).toBe(0)
})
it('args cross into the realm as data: mutating them (or their prototype chain) cannot reach host intrinsics', async () => {
const { ctx, parent } = await setup()
const hostArgs = { files: ['a.ts'], nested: { deep: [1, 2] } }
const result = await run(ctx, parent, script(`
args.files.push('b.ts')
Object.getPrototypeOf(args).polluted = 'realm-only'
return { count: args.files.length, deep: args.nested.deep[1] }
`), hostArgs)
expect(result.value).toEqual({ count: 2, deep: 2 })
// The host copy is untouched, and the HOST Object.prototype was never reachable.
expect(hostArgs.files).toEqual(['a.ts'])
expect(({} as Record<string, unknown>).polluted).toBeUndefined()
})
it('scalar/null args pass through directly; absent args leave the global undefined', async () => {
const { ctx, parent } = await setup()
expect((await run(ctx, parent, script('return args * 2'), 21)).value).toBe(42)
expect((await run(ctx, parent, script('return args === null'), null)).value).toBe(true)
expect((await run(ctx, parent, script('return typeof args'))).value).toBe('undefined')
})
it('a non-JSON return value fails loud as RESULT_UNSERIALIZABLE', async () => {
const { ctx, parent } = await setup()
const withDate = await run(ctx, parent, script('return { when: new Date(0) }'))
expect(withDate.stopReason).toBe('error')
expect(withDate.error).toContain('not plain JSON data')
const withFn = await run(ctx, parent, script('return { fn: () => 1 }'))
expect(withFn.error).toContain('not plain JSON data')
})
it('kills a synchronous spin in the initial slice via the vm timeout', async () => {
const { ctx, parent } = await setup({ config: { provider: 'stub', syncTimeoutMs: 50 } })
const result = await run(ctx, parent, script('while (true) {}'))
expect(result.stopReason).toBe('error')
expect(result.error?.toLowerCase()).toContain('timed out')
})
})
describe('lifecycle: parse errors, cancellation, disposal', () => {
it('start() throws synchronously for an unparseable script or invalid meta', async () => {
const { ctx, parent } = await setup()
expect(() => ctx.workflows.start({ script: 'const x = 1', parent })).toThrow(/must begin with/)
expect(() => ctx.workflows.start({ script: script('return ((('), parent })).toThrow(/does not parse/)
})
it('cancel() aborts in-flight children and settles the run cancelled', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const handle = ctx.workflows.start({ script: script("return await agent('long job')"), parent })
await vi.waitFor(() => { expect(provider.runs.length).toBe(1) })
handle.cancel('user stopped it')
const result = await handle.result
expect(result.stopReason).toBe('cancelled')
expect(result.error).toContain('user stopped it')
expect(provider.runs[0]!.disposed).toBe(true)
await handle.dispose()
})
it('an already-aborted request signal cancels before any child starts', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const controller = new AbortController()
controller.abort()
const handle = ctx.workflows.start({ script: script("return await agent('never')"), parent, signal: controller.signal })
const result = await handle.result
expect(result.stopReason).toBe('cancelled')
expect(provider.runs.length).toBe(0)
await handle.dispose()
})
it('the signal aborting mid-run cancels like cancel()', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const controller = new AbortController()
const handle = ctx.workflows.start({ script: script("return await agent('job')"), parent, signal: controller.signal })
await vi.waitFor(() => { expect(provider.runs.length).toBe(1) })
controller.abort()
const result = await handle.result
expect(result.stopReason).toBe('cancelled')
await handle.dispose()
})
it('reports a non-Error script throw (a thrown string) faithfully', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script("throw 'plain string failure'"))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('plain string failure')
})
it('a script Error surfaces its stack, carrying the script line numbers (lineOffset)', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script("throw new Error('with stack')"))
expect(result.stopReason).toBe('error')
// Line 1 is the blanked meta statement; the throw sits on line 2.
expect(result.error).toContain('workflow:test-flow:2')
})
it('an object throw with neither stack nor message stringifies', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script('throw { code: 42 }'))
expect(result.stopReason).toBe('error')
expect(result.error).toBe('[object Object]')
})
it('falls back to the message for an Error whose stack was stripped', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, script(`
const e = new Error('stackless failure')
e.stack = undefined
throw e
`))
expect(result.stopReason).toBe('error')
expect(result.error).toBe('stackless failure')
})
it('a waiter resumed by a release RACING a cancel still dies at the post-acquire check', async () => {
const { ctx, parent, provider } = await setup({ manual: true, config: { provider: 'stub', maxConcurrentAgents: 1 } })
const handle = ctx.workflows.start({
script: script("return await parallel([() => agent('a'), () => agent('b')])"),
parent,
})
await vi.waitFor(() => { expect(provider.runs.length).toBe(1) })
// Same synchronous block: the release resolves b's waiter, then the
// cancel lands BEFORE b's continuation runs — b must not start a child.
provider.runs[0]!.settle(text('a-done'))
handle.cancel('raced')
const result = await handle.result
expect(result.stopReason).toBe('cancelled')
expect(provider.runs.length).toBe(1)
await handle.dispose()
})
it('a dropped agent() promise cannot become an unhandled rejection when cancellation lands', async () => {
const unhandled: unknown[] = []
const onUnhandled = (reason: unknown): void => { unhandled.push(reason) }
process.on('unhandledRejection', onUnhandled)
try {
const { ctx, parent, provider } = await setup({ manual: true })
const handle = ctx.workflows.start({
script: script(`
agent('dropped, never awaited')
return await agent('awaited')
`),
parent,
})
await vi.waitFor(() => { expect(provider.runs.length).toBe(2) })
handle.cancel()
await handle.result
await handle.dispose()
// Let any stray rejection reach the process hook before asserting.
await new Promise(resolve => setTimeout(resolve, 20))
expect(unhandled).toEqual([])
} finally {
process.off('unhandledRejection', onUnhandled)
}
})
it('dispose() abandons a stuck script after the grace instead of hanging (result stays pending)', async () => {
const { ctx, parent } = await setup({ config: { provider: 'stub', disposeGraceMs: 30 } })
const handle = ctx.workflows.start({
// No hooks involved: an unsettleable await the engine cannot reject.
script: script("await new Promise(() => {})\nreturn 'unreachable'"),
parent,
})
const before = Date.now()
await handle.dispose()
expect(Date.now() - before).toBeLessThan(1000)
const settled = await Promise.race([handle.result.then(() => 'settled'), Promise.resolve('pending')])
expect(settled).toBe('pending')
})
it('dispose() is idempotent and settles cleanly after a completed run', async () => {
const { ctx, parent } = await setup()
const handle = ctx.workflows.start({ script: script('return 1'), parent })
await handle.result
await handle.dispose()
await handle.dispose()
})
it('strays: children fired without await are aborted once the script settles', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const handle = ctx.workflows.start({
script: script(`
agent('stray')
return 'done without awaiting'
`),
parent,
})
const result = await handle.result
expect(result.stopReason).toBe('completed')
await vi.waitFor(() => {
expect(provider.runs.length).toBe(1)
expect(provider.runs[0]!.disposed).toBe(true)
})
await handle.dispose()
})
})
describe('service surface', () => {
it('run ids are unique per start; the run handle and event payloads hold SEPARATE meta clones', async () => {
const { ctx, parent } = await setup()
let eventMeta: WorkflowRunInfo | undefined
ctx.on('workflow/start', (info) => { eventMeta = info })
const first = ctx.workflows.start({ script: script('return 1'), parent })
const second = ctx.workflows.start({ script: script('return 2'), parent })
expect(first.id).not.toBe(second.id)
// Mutating a listener's snapshot cannot corrupt the holder's view.
eventMeta!.meta.name = 'corrupted'
expect(second.meta.name).toBe('test-flow')
await Promise.all([first.result, second.result])
await first.dispose()
await second.dispose()
})
it('unregisters ctx.workflows when the engine fiber is disposed (HMR safety)', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
const fiber = await ctx.plugin(VmWorkflowEngine, {})
expect(ctx.get('workflows')).toBeDefined()
await fiber.dispose()
expect(ctx.get('workflows')).toBeUndefined()
})
it('has the class-plugin export shape (default = the engine service class)', () => {
expect(vmEngineModule.default).toBe(VmWorkflowEngine)
const loader = Object.create(Loader.prototype) as Loader
const unwrapped: unknown = loader.unwrapExports(vmEngineModule)
expect(unwrapped).toBe(VmWorkflowEngine)
})
})
})

131
packages/workflow/workflow-vm/tests/workflow.e2e.ts Normal file
View File

@@ -0,0 +1,131 @@
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import SubagentService from '@deepseek-ai/dsh-subagent'
import * as Spawn from '@deepseek-ai/dsh-subagent-spawn'
import * as ToolWorkflow from '@deepseek-ai/dsh-tool-workflow'
import { CallId } from '@deepseek-ai/dsh-llm'
import VmWorkflowEngine from '../src/index.ts'
/**
* With-key e2e for the workflow engine: a REAL script drives REAL spawn
* children against the live DeepSeek API — one plain child and one schema'd
* child through the real structured-output runtime — and the run's value,
* events, and child sessions are asserted from the outside (never the
* script's self-report alone). Key-gated (self-skips without
* DEEPSEEK_API_KEY).
*/
let ctx: Context | undefined
afterEach(async () => {
await ctx?.fiber.dispose()
ctx = undefined
})
async function harness(): Promise<Context> {
const built = new Context()
await built.plugin(LlmService)
await built.plugin(SessionStore)
await built.plugin(SystemPrompt)
await built.plugin(ToolRegistry)
await built.plugin(AgentRegistry)
await built.plugin(AgentLoop, { agents: [] })
await built.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] })
await built.plugin(SubagentService)
await built.plugin(Spawn, { providerName: 'spawn', structuredNudgeRetries: 1 })
await built.plugin(VmWorkflowEngine, { provider: 'spawn' })
await built.plugin(ToolWorkflow, {})
return built
}
const SCRIPT = `export const meta = {
name: 'e2e-arithmetic',
description: 'two real children: one prose, one structured',
phases: [{ title: 'Ask' }, { title: 'Judge' }],
}
phase('Ask')
log('asking the prose child')
const prose = await agent('Reply with exactly one short sentence: what is 2 + 2?')
phase('Judge')
const judged = await agent(
'Here is an answer to the question "what is 2+2": ' + prose
+ ' — report whether it contains the number 4 and your confidence between 0 and 1.',
{ schema: { type: 'object', properties: { containsFour: { type: 'boolean' }, confidence: { type: 'number' } }, required: ['containsFour'] } },
)
return { prose, containsFour: judged === null ? null : judged.containsFour }`
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('workflow engine with-key e2e', () => {
it('runs a two-phase script over real children, one through the structured runtime', async () => {
ctx = await harness()
const parentHandle = ctx.agents.create({
agentId: AgentId('wf-e2e-parent'),
sessionId: 'wf-e2e-session' as never,
agentOptions: { model: 'deepseek-v4-flash' },
})
const events: string[] = []
const childIds: string[] = []
for (const name of ['workflow/start', 'workflow/phase', 'workflow/log', 'workflow/agent-start', 'workflow/agent-end', 'workflow/end'] as const) {
ctx.on(name, (...payload: unknown[]) => {
events.push(name)
if (name === 'workflow/agent-start') childIds.push((payload[1] as { childId: string }).childId)
})
}
const run = ctx.workflows.start({ script: SCRIPT, parent: parentHandle.agent })
const result = await run.result
await run.dispose()
expect(result.stopReason).toBe('completed')
expect(result.agentsStarted).toBe(2)
const value = result.value as { prose: string; containsFour: boolean | null }
// World checks: the prose child really answered (a real completion), and
// the structured child judged it against the REAL schema-forced tool.
expect(value.prose.length).toBeGreaterThan(0)
expect(value.containsFour).toBe(true)
expect(events[0]).toBe('workflow/start')
expect(events.at(-1)).toBe('workflow/end')
expect(events.filter(name => name === 'workflow/phase').length).toBe(2)
expect(events.filter(name => name === 'workflow/agent-start').length).toBe(2)
expect(childIds.length).toBe(2)
// The children were disposed to quiescence after collection.
for (const childId of childIds) {
expect(ctx.agents.get(AgentId(childId))).toBeUndefined()
}
await parentHandle.dispose()
}, 240_000)
it('the workflow TOOL runs the same path through the real registry pipeline', async () => {
ctx = await harness()
const parentHandle = ctx.agents.create({
agentId: AgentId('wf-e2e-tool-parent'),
sessionId: 'wf-e2e-tool-session' as never,
agentOptions: { model: 'deepseek-v4-flash' },
})
const result = await ctx.tools.execute({
callId: CallId('wf-e2e-call'),
name: 'workflow',
arguments: {
script: `export const meta = { name: 'e2e-tool', description: 'one real child via the tool' }
const answer = await agent('Reply with exactly one word: the capital of France.')
return { answer }`,
},
agent: parentHandle.agent,
})
expect(result.isError).toBe(false)
const text = (result.content[0] as { text: string }).text
expect(text).toContain('workflow "e2e-tool" completed (1 agent)')
expect(text.toLowerCase()).toContain('paris')
await parentHandle.dispose()
}, 240_000)
})

39
packages/workflow/workflow-vm/tsconfig.json Normal file
View File

@@ -0,0 +1,39 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../../vendor/schemastery"
},
{
"path": "../../core/agent"
},
{
"path": "../../util/brand"
},
{
"path": "../../llm/llm"
},
{
"path": "../../subagent/subagent"
},
{
"path": "../../core/tools"
},
{
"path": "../workflow"
}
]
}

29
packages/workflow/workflow/README.md Normal file
View File

@@ -0,0 +1,29 @@
# @deepseek-ai/dsh-workflow
The **workflow seam** (`ctx.workflows`): an abstract service defining WHAT a workflow engine does — execute a model-written orchestration script that fans out subagents — without saying HOW. The bash-shaped third of the [workflow family](../README.md): implementations subclass `WorkflowService` and register as the `workflows` service (one per context); [`dsh-workflow-vm`](../workflow-vm/README.md) is the first, and [`dsh-tool-workflow`](../tool-workflow/README.md) is the model-facing consumer.
## Service: `WorkflowService` (abstract)
`start(request: WorkflowStartRequest): WorkflowRun` — parse and execute a script. Throws synchronously (`SCRIPT_PARSE`/`META_INVALID`) for a script that cannot begin; once a run is returned, its `result` NEVER rejects — every failure resolves with `stopReason: 'error'` (or `'cancelled'`). `dispose()` must reach quiescence within a bounded grace (cancel → wait → abandon), never hanging its caller.
The protected `emitWorkflowEvent` helper dispatches the `workflow/*` events with PER-LISTENER containment (a throwing subscriber is logged, never propagated, and cannot starve later listeners) — the same guarantee as the subagent seam's lifecycle emits.
## Vocabulary
- `WorkflowStartRequest` — `{ script, args?, parent: Agent, signal? }`. `parent` is REQUIRED: every child the script spawns is attributed to it. `args` must be plain host-realm JSON data.
- `WorkflowMeta` / `WorkflowPhase` — the script's validated `export const meta` block (Claude Code format: required `name`/`description`, optional `whenToUse`/`phases`).
- `WorkflowRun` — `{ id, meta, result, cancel(reason?), dispose() }`; the consumer awaits `result` and MUST `dispose` on every path.
- `WorkflowResult` — `{ value, stopReason: 'completed'|'cancelled'|'error', error?, agentsStarted }`; `value` is the script's materialized return (plain JSON data; `null` for no return).
- `WorkflowError` — `HarnessError` with a `WorkflowErrorCode` and a `fatal` flag driving the combinator discipline: a fatal error (bad hook arguments, unsupported options/schemas, tripped caps, seam start failures, cancellation) always propagates through `parallel()`/`pipeline()` instead of dissolving into a per-item `null`. `isFatalWorkflowError(error)` is the catch-site predicate.
## Events
All observe-only emits carrying DATA SNAPSHOTS (`WorkflowRunInfo` = id + meta) — never the live `WorkflowRun`, so a listener cannot gain `cancel`/`dispose`; control stays with the `start()` caller:
- `workflow/start`(info) / `workflow/end`(info, resultInfo) — run lifecycle; `resultInfo` deliberately omits the value.
- `workflow/phase`(info, title) / `workflow/log`(info, message) — script narration.
- `workflow/agent-start`(info, agent) / `workflow/agent-end`(info, agent + outcome) — one pair per `agent()` call, correlated by `seq`.
## Non-goals (this cut)
Background collection, journaling/resume, saved workflows, nested `workflow()`, token budgets — see the [RFC's deferred section](../../../docs/rfc/implemented/feature/2026-07-05-dynamic-workflows.md).

37
packages/workflow/workflow/package.json Normal file
View File

@@ -0,0 +1,37 @@
{
"name": "@deepseek-ai/dsh-workflow",
"description": "Workflow capability seam: ctx.workflows service, run vocabulary, and workflow/* events",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

224
packages/workflow/workflow/src/index.ts Normal file
View File

@@ -0,0 +1,224 @@
/**
* The workflow capability seam (`ctx.workflows`): an abstract service defining
* WHAT a workflow engine does — execute a model-written orchestration script
* that fans out subagents — without saying HOW. Implementations subclass
* {@link WorkflowService} and register as the `workflows` service (one
* implementation per context, cordis' standard duplicate-service behavior);
* `@deepseek-ai/dsh-workflow-vm` (an in-process `node:vm` engine) is the
* first. Future engines (a worker-thread or isolated-vm sandbox) swap in
* without touching the model-facing tool that consumes them
* (`@deepseek-ai/dsh-tool-workflow`).
*
* The `workflow/*` lifecycle events are OBSERVE-ONLY data snapshots: they
* carry {@link WorkflowRunInfo} (id + meta), never the live {@link WorkflowRun}
* — a listener must not gain `cancel`/`dispose`; control stays with the
* `start()` caller holding the run. Every emit is per-listener contained (a
* throwing subscriber is logged, never propagated), so one bad observer can
* neither strand a live run nor starve later listeners.
*
* @module @deepseek-ai/dsh-workflow
*/
import { Context, Service } from 'cordis'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import type {
WorkflowAgentEndInfo,
WorkflowAgentInfo,
WorkflowResultInfo,
WorkflowRun,
WorkflowRunInfo,
WorkflowStartRequest,
} from './types.ts'
export { WorkflowRunId } from './types.ts'
export type {
WorkflowAgentEndInfo,
WorkflowAgentInfo,
WorkflowAgentOutcome,
WorkflowMeta,
WorkflowPhase,
WorkflowResult,
WorkflowResultInfo,
WorkflowRun,
WorkflowRunInfo,
WorkflowStartRequest,
WorkflowStopReason,
} from './types.ts'
declare module 'cordis' {
interface Context {
workflows: WorkflowService
}
interface Events {
/**
* A workflow run started — the script's meta block validated, the body
* about to execute. Paired with {@link Events['workflow/end']}.
* @param info - the run's identity snapshot (id + meta).
* @mode emit
*/
'workflow/start'(info: WorkflowRunInfo): void
/**
* The script entered a phase (a `phase(title)` call) — progress grouping
* for observers; no execution semantics.
* @param info - the run's identity snapshot.
* @param title - the phase title, verbatim.
* @mode emit
*/
'workflow/phase'(info: WorkflowRunInfo, title: string): void
/**
* The script emitted a narration line (a `log(message)` call).
* @param info - the run's identity snapshot.
* @param message - the logged message, verbatim.
* @mode emit
*/
'workflow/log'(info: WorkflowRunInfo, message: string): void
/**
* One `agent()` call started a child run. Paired with
* {@link Events['workflow/agent-end']} by `agent.seq`.
* @param info - the run's identity snapshot.
* @param agent - the call's sequence number, label, phase, and child id.
* @mode emit
*/
'workflow/agent-start'(info: WorkflowRunInfo, agent: WorkflowAgentInfo): void
/**
* One `agent()` call settled (clean result, child failure, or run
* cancellation). Paired with {@link Events['workflow/agent-start']}.
* @param info - the run's identity snapshot.
* @param agent - the call identity plus its outcome.
* @mode emit
*/
'workflow/agent-end'(info: WorkflowRunInfo, agent: WorkflowAgentEndInfo): void
/**
* A workflow run settled (any stop reason). Fired when
* {@link WorkflowRun.result} resolves. Paired with
* {@link Events['workflow/start']}.
* @param info - the run's identity snapshot.
* @param result - the outcome data (stop reason, error, agent count) —
* deliberately WITHOUT the result value (see {@link WorkflowResultInfo}).
* @mode emit
*/
'workflow/end'(info: WorkflowRunInfo, result: WorkflowResultInfo): void
}
}
/** The full set of `workflow/*` event names {@link WorkflowService.emitWorkflowEvent} dispatches. */
export type WorkflowEventName =
| 'workflow/start'
| 'workflow/phase'
| 'workflow/log'
| 'workflow/agent-start'
| 'workflow/agent-end'
| 'workflow/end'
/**
* The workflow-seam error codes. Every one of these is FATAL when it reaches
* a script (see {@link WorkflowError.fatal}): the combinators re-throw it
* instead of dissolving it into an ordinary per-item `null`.
*
* - `SCRIPT_PARSE` — the script (or its meta statement) does not parse.
* - `META_INVALID` — the meta block evaluated but fails the shape contract.
* - `INVALID_ARGUMENT` — a hook was called with malformed arguments.
* - `UNSUPPORTED_OPTION` — an `agent()` option this engine does not support
* (deferred: `effort`/`isolation`/`agentType`) or does not know.
* - `UNSUPPORTED_SCHEMA` — an `agent()` schema outside the structured-output
* subset (see dsh-tools).
* - `AGENT_CAP` / `ITEM_CAP` — the run/agent caps tripped.
* - `AGENT_START` — the subagent seam refused to start a child.
* - `RESULT_UNSERIALIZABLE` — a value crossing the realm boundary is not
* plain JSON data.
* - `CANCELLED` — the run was cancelled; pending and future hooks reject
* with this (the script-kill mechanism).
*/
export type WorkflowErrorCode =
| 'SCRIPT_PARSE'
| 'META_INVALID'
| 'INVALID_ARGUMENT'
| 'UNSUPPORTED_OPTION'
| 'UNSUPPORTED_SCHEMA'
| 'AGENT_CAP'
| 'ITEM_CAP'
| 'AGENT_START'
| 'RESULT_UNSERIALIZABLE'
| 'CANCELLED'
/**
* Typed error for workflow-seam failures. Extends {@link HarnessError}, so the
* `code` is machine-routable taxonomy. `fatal` drives the combinator
* discipline: `parallel()`/`pipeline()` re-throw a fatal error (a typo'd
* option or a tripped cap must kill the script loudly), and reserve the
* per-item `null` for child-run failures and ordinary in-stage script errors.
* Every {@link WorkflowErrorCode} is fatal in this cut; the flag exists so the
* distinction is explicit at every catch site rather than implied.
*/
export class WorkflowError extends HarnessError {
/** Whether combinators must propagate this error instead of nulling the item. */
readonly fatal: boolean
constructor(message: string, code: WorkflowErrorCode, options?: ErrorOptions & { fatal?: boolean }) {
super(message, code, options)
this.name = 'WorkflowError'
this.fatal = options?.fatal ?? true
}
}
/** Whether combinators must re-throw `error` instead of mapping the item to `null`. */
export function isFatalWorkflowError(error: unknown): boolean {
return error instanceof WorkflowError && error.fatal
}
/**
* Abstract workflow execution service. Subclass, implement {@link start}, and
* load the subclass as a plugin — it registers as `ctx.workflows` (one
* implementation per context; loading a second throws, cordis' standard
* duplicate-service behavior).
*
* Semantics every implementation must honor:
* - {@link start} throws synchronously for a request that cannot begin (an
* unparseable script, an invalid meta block). Once it returns a
* {@link WorkflowRun}, `result` NEVER rejects — every failure resolves with
* `stopReason: 'error'` (or `'cancelled'`).
* - The `workflow/*` events fire through {@link emitWorkflowEvent} (data
* snapshots, per-listener containment); `workflow/end` fires exactly once
* per started run, after `result` is settled or as it settles.
* - `dispose()` reaches quiescence within a bounded grace: it cancels, waits
* for the script to settle, and abandons a stuck script rather than
* hanging its caller (the engine documents what abandonment leaves behind).
*/
export abstract class WorkflowService extends Service {
constructor(ctx: Context) {
super(ctx, 'workflows')
}
/**
* Parse and execute a workflow script.
* @param request - the script, its `args`, the parent agent, and an
* optional cancel signal.
* @returns the live run; its `result` resolves when the script settles.
*/
abstract start(request: WorkflowStartRequest): WorkflowRun
/**
* Emit one `workflow/*` lifecycle event with PER-LISTENER containment:
* dispatch each subscriber individually and log (never propagate) a thrown
* one, so one bad subscriber can neither fail the engine mid-run, surface as
* an unhandled rejection on a detached settle hook, nor starve the listeners
* registered after it (cordis `emit` halts on the first throw — same
* guarantee as the subagent seam's lifecycle emits).
* @param name - the `workflow/*` event to dispatch.
* @param args - the event's payload, matching its declared signature.
*/
protected emitWorkflowEvent(name: WorkflowEventName, ...args: unknown[]): void {
for (const callback of this.ctx.events.dispatch('emit', [name, ...args])) {
try {
// The declared workflow/* signatures are all void-returning emits; the
// dispatch callback applies the payload tuple.
;(callback as (...payload: unknown[]) => void)(...args)
} catch (error: unknown) {
this.ctx.logger.warn(`workflow: ${name} listener threw: ${String(error)}`)
}
}
}
}
export default WorkflowService

154
packages/workflow/workflow/src/types.ts Normal file
View File

@@ -0,0 +1,154 @@
/**
* Workflow seam vocabulary: the request/run/result types a workflow engine
* consumes and produces, plus the payload shapes of the `workflow/*` events.
* Types only (plus the id-brand factory), per the package convention.
*
* @module @deepseek-ai/dsh-workflow/types
*/
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { Agent, AgentId } from '@deepseek-ai/dsh-agent'
/** Identifies one workflow run. */
export type WorkflowRunId = Branded<'WorkflowRunId'>
/** Brand a string as a {@link WorkflowRunId}. */
export function WorkflowRunId(id: string): WorkflowRunId {
return id as WorkflowRunId
}
/**
* One phase declared in a script's `meta.phases` (progress vocabulary only —
* phases group agents in observers/UIs; they impose no execution structure).
*/
export interface WorkflowPhase {
/** The phase title; `phase()` calls match against it by exact string. */
title: string
/** Optional one-line description of what the phase does. */
detail?: string
/** Optional model override this phase is expected to use (informational). */
model?: string
}
/**
* The script's `export const meta` block, validated by the engine before the
* body runs. `name`/`description` are required; the rest is optional
* annotation. Matches the Claude Code dynamic-workflows script format.
*/
export interface WorkflowMeta {
/** Short kebab-case workflow name (display + persistence key). */
name: string
/** One-line description of what the workflow does. */
description: string
/** Optional guidance on when this workflow applies (shown in listings). */
whenToUse?: string
/** Optional phase declarations matched by `phase()` calls. */
phases?: WorkflowPhase[]
}
/**
* What a caller asks for when starting a workflow run. `parent` is REQUIRED —
* every `agent()` the script spawns is attributed to it (cwd, lineage, depth
* flow through the subagent seam). `args` must be plain host-realm JSON data;
* the engine exposes it to the script as the `args` global.
*/
export interface WorkflowStartRequest {
/** The full script text: `export const meta = {...}` + a plain-JS body. */
script: string
/** Optional input exposed verbatim to the script as the `args` global. */
args?: unknown
/** The agent on whose behalf the run executes (parent of every child). */
parent: Agent
/** Cancels the run when aborted (the tool's `exec.signal`). */
signal?: AbortSignal
}
/**
* Why a run settled. CLOSED union (engine-owned, consumers may exhaust):
* `completed` = the script ran to its final `return`; `cancelled` = the run
* was cancelled (caller `cancel()`/signal); `error` = the script threw, a
* fatal `WorkflowError` propagated, or the result failed materialization.
*/
export type WorkflowStopReason = 'completed' | 'cancelled' | 'error'
/**
* The outcome of one run, resolved by {@link WorkflowRun.result}. `value` is
* the script's materialized return value (plain host-realm JSON data; `null`
* when the script returned `undefined`) — meaningful only for `completed`.
* A non-`completed` reason carries the failure in `error`; the consumer maps
* it to an `isError` tool result rather than reporting partial output.
*/
export interface WorkflowResult {
/** The script's return value (host JSON data; `null` for no return). */
value: unknown
/** Why the run settled. */
stopReason: WorkflowStopReason
/** The failure message (present iff `stopReason` is not `completed`). */
error?: string
/** How many `agent()` calls the run started (across its whole lifetime). */
agentsStarted: number
}
/**
* The handle the consumer holds while a script executes. The consumer awaits
* `result`, may `cancel` mid-flight, and MUST `dispose` on every path.
* `result` does NOT reject — a script failure resolves with `stopReason:
* 'error'` — so the consumer maps a non-`completed` reason to an `isError`
* result. `dispose()` cancels, then waits a bounded grace for the script to
* settle before abandoning it (the engine documents the abandonment
* semantics); it never hangs on a stuck script.
*/
export interface WorkflowRun {
readonly id: WorkflowRunId
/** The validated meta block (available before the body runs). */
readonly meta: WorkflowMeta
readonly result: Promise<WorkflowResult>
/** Cancel the run: children abort, pending hooks reject, the script dies at its next await. */
cancel(reason?: string): void
/** Cancel + bounded-grace settle; safe to call on every path (idempotent). */
dispose(): Promise<void>
}
/** Identifying detail for a run, carried by every `workflow/*` event (a data snapshot, never the live run). */
export interface WorkflowRunInfo {
/** The run's id. */
id: WorkflowRunId
/** The run's validated meta block. */
meta: WorkflowMeta
}
/** One `agent()` call's identity within a run (the `workflow/agent-start` payload). */
export interface WorkflowAgentInfo {
/** 1-based sequence number of this `agent()` call within the run. */
seq: number
/** The display label (the `label` option, or a prompt snippet). */
label: string
/** The phase this agent belongs to (the `phase` option, else the current `phase()` title). */
phase?: string
/** The child agent's id on the subagent seam. */
childId: AgentId
}
/** How one `agent()` call settled: clean result, child failure (script sees `null`), or run cancellation. */
export type WorkflowAgentOutcome = 'completed' | 'failed' | 'cancelled'
/** One `agent()` call's settlement (the `workflow/agent-end` payload). */
export interface WorkflowAgentEndInfo extends WorkflowAgentInfo {
/** How the call settled. */
outcome: WorkflowAgentOutcome
}
/**
* A settled run's outcome as event data (the `workflow/end` payload): the
* {@link WorkflowResult} minus `value` (a listener observing outcomes must not
* receive a mutable alias of the caller's result value; a consumer that needs
* the value holds the run and awaits `result`).
*/
export interface WorkflowResultInfo {
/** Why the run settled. */
stopReason: WorkflowStopReason
/** The failure message (present iff `stopReason` is not `completed`). */
error?: string
/** How many `agent()` calls the run started. */
agentsStarted: number
}

86
packages/workflow/workflow/tests/workflow.spec.ts Normal file
View File

@@ -0,0 +1,86 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import WorkflowServiceDefault, {
isFatalWorkflowError,
WorkflowError,
WorkflowRunId,
WorkflowService,
} from '../src/index.ts'
import type { WorkflowRun, WorkflowRunInfo, WorkflowStartRequest } from '../src/index.ts'
/** A minimal concrete subclass exposing the protected emit helper for tests. */
class StubEngine extends WorkflowService {
start(request: WorkflowStartRequest): WorkflowRun {
void request
throw new Error('not under test')
}
emit(name: Parameters<WorkflowService['emitWorkflowEvent']>[0], ...args: unknown[]): void {
this.emitWorkflowEvent(name, ...args)
}
}
const INFO: WorkflowRunInfo = { id: WorkflowRunId('run-1'), meta: { name: 'w', description: 'd' } }
describe('dsh-workflow (interface)', () => {
it('WorkflowRunId brands a string (identity at runtime)', () => {
expect(WorkflowRunId('abc')).toBe('abc')
})
it('WorkflowError carries code + fatal (default true) and reads as a HarnessError', () => {
const error = new WorkflowError('cap hit', 'AGENT_CAP')
expect(error.code).toBe('AGENT_CAP')
expect(error.fatal).toBe(true)
expect(error.name).toBe('WorkflowError')
const soft = new WorkflowError('advisory', 'ITEM_CAP', { fatal: false })
expect(soft.fatal).toBe(false)
})
it('isFatalWorkflowError: true only for a fatal WorkflowError', () => {
expect(isFatalWorkflowError(new WorkflowError('x', 'CANCELLED'))).toBe(true)
expect(isFatalWorkflowError(new WorkflowError('x', 'CANCELLED', { fatal: false }))).toBe(false)
expect(isFatalWorkflowError(new Error('plain'))).toBe(false)
expect(isFatalWorkflowError('string')).toBe(false)
})
it('registers as ctx.workflows and unregisters when its fiber is disposed (HMR safety)', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(StubEngine)
expect(ctx.get('workflows')).toBeInstanceOf(StubEngine)
await fiber.dispose()
expect(ctx.get('workflows')).toBeUndefined()
})
it('emitWorkflowEvent dispatches to every listener with the payload tuple', async () => {
const ctx = new Context()
await ctx.plugin(StubEngine)
const seen: unknown[][] = []
ctx.on('workflow/log', (info, message) => { seen.push([info, message]) })
ctx.on('workflow/agent-start', (info, agent) => { seen.push([info, agent]) })
const engine = ctx.workflows as StubEngine
engine.emit('workflow/log', INFO, 'hello')
engine.emit('workflow/agent-start', INFO, { seq: 1, label: 'l', childId: 'c' })
expect(seen).toEqual([
[INFO, 'hello'],
[INFO, { seq: 1, label: 'l', childId: 'c' }],
])
})
it('contains a throwing listener PER LISTENER: later listeners still run, nothing propagates', async () => {
const ctx = new Context()
await ctx.plugin(StubEngine)
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => ctx.logger)
const reached: string[] = []
ctx.on('workflow/phase', () => { throw new Error('bad listener') })
ctx.on('workflow/phase', (_info, title) => { reached.push(title) })
const engine = ctx.workflows as StubEngine
expect(() => { engine.emit('workflow/phase', INFO, 'Scan') }).not.toThrow()
expect(reached).toEqual(['Scan'])
expect(warn).toHaveBeenCalledOnce()
expect(String(warn.mock.calls[0]![0])).toContain('workflow/phase listener threw')
})
it('has the expected export surface (default = the abstract service class)', () => {
expect(WorkflowServiceDefault).toBe(WorkflowService)
})
})

27
packages/workflow/workflow/tsconfig.json Normal file
View File

@@ -0,0 +1,27 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../core/agent"
},
{
"path": "../../util/brand"
},
{
"path": "../../llm/llm"
}
]
}