Merge remote-tracking branch 'origin/master' into codex/pr224-rfc-rewrite
# Conflicts: # docs/architecture.md # docs/event-producer-consumer.md # docs/module-graph.md # scripts/doc-budgets.manifest.json # scripts/gen-doc-graphs.ts
This commit is contained in:
Tianyi Cui
154 files changed
+13524
-5857
No files matched your search
@@ -0,0 +1,57 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import { rm, writeFile } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { execFile } from 'node:child_process'
|
||||
import { promisify } from 'node:util'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
const packageRoot = fileURLToPath(new URL('..', import.meta.url))
|
||||
const builtIndex = join(packageRoot, 'lib', 'index.js')
|
||||
const builtWorker = join(packageRoot, 'lib', 'worker.js')
|
||||
const run = promisify(execFile)
|
||||
|
||||
/**
|
||||
* The BUILT-output guard for the worker entry: every other suite runs
|
||||
* unbuilt (src/ + tsx), so nothing else proves that `lib/index.js` resolves
|
||||
* its sibling `lib/worker.js` and that the bundle boots a worker under plain
|
||||
* node (no tsx loader). Keyless — a zero-agent script needs no provider —
|
||||
* and self-skips until `pnpm run build` has produced the bundles.
|
||||
*/
|
||||
describe.skipIf(!existsSync(builtIndex) || !existsSync(builtWorker))('built worker entry (lib/worker.js)', () => {
|
||||
it('the built engine spawns its built worker under plain node and completes a run', async () => {
|
||||
// ESM resolves bare specifiers from the IMPORTING FILE's location, so the
|
||||
// driver must live inside the package for its node_modules to apply — a
|
||||
// temp-named file at the package root, removed on the way out.
|
||||
const driver = join(packageRoot, `.built-worker-driver-${process.pid}.mjs`)
|
||||
try {
|
||||
await writeFile(driver, `
|
||||
import { Context } from 'cordis'
|
||||
import SubagentService from '@deepseek-ai/dsh-subagent'
|
||||
import WorkerWorkflowEngine from '@deepseek-ai/dsh-workflow-workerthread'
|
||||
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
await ctx.plugin(WorkerWorkflowEngine, {})
|
||||
const run = ctx.workflows.start({
|
||||
script: 'return 6 * 7',
|
||||
meta: { name: 'built-smoke', description: 'built worker smoke' },
|
||||
// A zero-agent script never touches the provider, so a bare id suffices.
|
||||
parent: { id: 'built-smoke-parent', options: {} },
|
||||
})
|
||||
const result = await run.result
|
||||
await run.dispose()
|
||||
if (result.stopReason !== 'completed' || result.value !== 42) {
|
||||
console.error('unexpected result: ' + JSON.stringify(result))
|
||||
process.exit(1)
|
||||
}
|
||||
console.log('built-worker-smoke-ok')
|
||||
`, 'utf8')
|
||||
// Plain node — no tsx loader anywhere; the bundle must stand on its own.
|
||||
const { stdout } = await run(process.execPath, [driver], { cwd: packageRoot, timeout: 60_000 })
|
||||
expect(stdout).toContain('built-worker-smoke-ok')
|
||||
} finally {
|
||||
await rm(driver, { force: true })
|
||||
}
|
||||
}, 120_000)
|
||||
})
|
||||
@@ -0,0 +1,91 @@
|
||||
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 WorkerWorkflowEngine from '../src/index.ts'
|
||||
|
||||
type Script = ConstructorParameters<typeof MockAdapter>[0]
|
||||
|
||||
/**
|
||||
* The whole in-process stack, keyless, with the script in a REAL worker
|
||||
* thread: the 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 guard the unit suites structurally
|
||||
* cannot give — the MessageChannel suite fakes the host, and the host suite
|
||||
* stubs the subagent seam.
|
||||
*/
|
||||
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' })
|
||||
await ctx.plugin(WorkerWorkflowEngine, {})
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const parent = ctx.agentLoop.create(AgentId('parent'), { model: 'mock' })
|
||||
return { ctx, parent, adapter }
|
||||
}
|
||||
|
||||
describe('dsh-workflow-workerthread 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({
|
||||
meta: { name: 'integration', description: 'plain + structured children' },
|
||||
script: `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({
|
||||
meta: { name: 'null-path', description: 'schema failure maps to null' },
|
||||
script: `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()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,88 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { WorkflowError } from '@deepseek-ai/dsh-workflow'
|
||||
import { validateMeta } from '../src/meta.ts'
|
||||
|
||||
/** Assert a META_INVALID throw whose message matches every given fragment. */
|
||||
function expectInvalid(value: unknown, ...fragments: string[]): void {
|
||||
let thrown: unknown
|
||||
try {
|
||||
validateMeta(value)
|
||||
} catch (error: unknown) {
|
||||
thrown = error
|
||||
}
|
||||
expect(thrown).toBeInstanceOf(WorkflowError)
|
||||
expect((thrown as WorkflowError).code).toBe('META_INVALID')
|
||||
for (const fragment of fragments) {
|
||||
expect((thrown as WorkflowError).message).toContain(fragment)
|
||||
}
|
||||
}
|
||||
|
||||
describe('validateMeta', () => {
|
||||
it('accepts a minimal meta and returns a normalized copy (no aliasing of the input)', () => {
|
||||
const input = { name: 'audit', description: 'audit the repo' }
|
||||
const meta = validateMeta(input)
|
||||
expect(meta).toEqual({ name: 'audit', description: 'audit the repo' })
|
||||
expect(meta).not.toBe(input)
|
||||
input.name = 'mutated'
|
||||
expect(meta.name).toBe('audit')
|
||||
})
|
||||
|
||||
it('accepts the full shape and rebuilds phases entry by entry', () => {
|
||||
const meta = validateMeta({
|
||||
name: 'migrate',
|
||||
description: 'migrate call sites',
|
||||
whenToUse: 'large mechanical sweeps',
|
||||
phases: [
|
||||
{ title: 'Discover' },
|
||||
{ title: 'Transform', detail: 'one agent per file', model: 'deepseek-v4-pro' },
|
||||
],
|
||||
})
|
||||
expect(meta).toEqual({
|
||||
name: 'migrate',
|
||||
description: 'migrate call sites',
|
||||
whenToUse: 'large mechanical sweeps',
|
||||
phases: [
|
||||
{ title: 'Discover' },
|
||||
{ title: 'Transform', detail: 'one agent per file', model: 'deepseek-v4-pro' },
|
||||
],
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects non-object values loud', () => {
|
||||
expectInvalid(undefined, 'meta must be an object')
|
||||
expectInvalid('a string', 'meta must be an object')
|
||||
expectInvalid(null, 'meta must be an object')
|
||||
expectInvalid([{ name: 'x', description: 'd' }], 'meta must be an object')
|
||||
})
|
||||
|
||||
it('rejects unknown fields by name (accepted-then-ignored is banned)', () => {
|
||||
expectInvalid({ name: 'x', description: 'd', color: 'red' }, 'meta.color is not a recognized field')
|
||||
})
|
||||
|
||||
it('rejects missing or mistyped name/description/whenToUse', () => {
|
||||
expectInvalid({ description: 'd' }, 'meta.name must be a non-empty string')
|
||||
expectInvalid({ name: '', description: 'd' }, 'meta.name must be a non-empty string')
|
||||
expectInvalid({ name: 'x' }, 'meta.description must be a non-empty string')
|
||||
expectInvalid({ name: 'x', description: 42 }, 'meta.description must be a non-empty string')
|
||||
expectInvalid({ name: 'x', description: 'd', whenToUse: 3 }, 'meta.whenToUse must be a string')
|
||||
})
|
||||
|
||||
it('rejects malformed phases, entry by entry', () => {
|
||||
expectInvalid({ name: 'x', description: 'd', phases: 'Scan' }, 'meta.phases must be an array')
|
||||
expectInvalid({ name: 'x', description: 'd', phases: ['Scan'] }, 'meta.phases[0] must be an object')
|
||||
expectInvalid({ name: 'x', description: 'd', phases: [{ title: '' }] }, 'meta.phases[0].title must be a non-empty string')
|
||||
expectInvalid({ name: 'x', description: 'd', phases: [{ title: 'Scan', order: 1 }] }, 'meta.phases[0].order is not a recognized field')
|
||||
expectInvalid({ name: 'x', description: 'd', phases: [{ title: 'Scan', detail: 9 }] }, 'meta.phases[0].detail must be a string')
|
||||
expectInvalid({ name: 'x', description: 'd', phases: [{ title: 'Scan', model: 9 }] }, 'meta.phases[0].model must be a string')
|
||||
})
|
||||
|
||||
it('names EVERY violation in one throw, not just the first', () => {
|
||||
expectInvalid(
|
||||
{ description: 7, extra: true, phases: [{ title: 'Scan' }, 'bad'] },
|
||||
'meta.extra is not a recognized field',
|
||||
'meta.name must be a non-empty string',
|
||||
'meta.description must be a non-empty string',
|
||||
'meta.phases[1] must be an object',
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,145 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import * as vm from 'node:vm'
|
||||
import { materializeFromRealm, MaterializeError, renderThrown } 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('invokes getters ordinarily — the getter RESULT is what crosses (trust premise)', () => {
|
||||
const counter = inRealm(`
|
||||
(() => {
|
||||
globalThis.reads = 0
|
||||
return { get x() { globalThis.reads += 1; return globalThis.reads } }
|
||||
})()
|
||||
`)
|
||||
expect(materializeFromRealm(counter)).toEqual({ x: 1 })
|
||||
})
|
||||
|
||||
it('a getter that THROWS surfaces as a MaterializeError carrying the rendered failure', () => {
|
||||
const hostile = inRealm("{ get x() { throw new Error('read failed') } }")
|
||||
const message = rejection(hostile)
|
||||
expect(message).toContain('reading the value threw')
|
||||
expect(message).toContain('read failed')
|
||||
})
|
||||
|
||||
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 and non-index array properties; an array getter element materializes its value', () => {
|
||||
expect(rejection(inRealm('[1, , 3]'))).toContain('sparse')
|
||||
expect(rejection(inRealm('(() => { const a = [1]; a.total = 3; return a })()')))
|
||||
.toContain('non-index')
|
||||
expect(materializeFromRealm(inRealm('(() => { const a = [1]; Object.defineProperty(a, 0, { get: () => 7, enumerable: true }); return a })()')))
|
||||
.toEqual([7])
|
||||
})
|
||||
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
||||
describe('renderThrown', () => {
|
||||
it('prefers the stack, for host and realm errors alike', () => {
|
||||
const host = renderThrown(new Error('host failure'))
|
||||
expect(host).toContain('host failure')
|
||||
expect(host).toContain('at ') // a real stack, not just the message
|
||||
const realmError: unknown = vm.runInNewContext('(() => { try { throw new Error("realm failure") } catch (e) { return e } })()')
|
||||
expect(renderThrown(realmError)).toContain('realm failure')
|
||||
})
|
||||
|
||||
it('falls back from stack to message to String()', () => {
|
||||
expect(renderThrown({ stack: 'custom data stack' })).toBe('custom data stack')
|
||||
const stackless = new Error('stackless failure')
|
||||
delete stackless.stack
|
||||
expect(renderThrown(stackless)).toBe('stackless failure')
|
||||
expect(renderThrown({ code: 42 })).toBe('[object Object]')
|
||||
expect(renderThrown('plain')).toBe('plain')
|
||||
expect(renderThrown(42)).toBe('42')
|
||||
expect(renderThrown(undefined)).toBe('undefined')
|
||||
expect(renderThrown(null)).toBe('null')
|
||||
})
|
||||
|
||||
it('is total: a value whose accessors/toString throw renders as a fixed label', () => {
|
||||
expect(renderThrown({ get stack() { throw new Error('nope') } })).toBe('[unrenderable thrown value]')
|
||||
expect(renderThrown({ [Symbol.toPrimitive]() { throw new Error('nope') } })).toBe('[unrenderable thrown value]')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,504 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { MessageChannel } from 'node:worker_threads'
|
||||
import type { MessagePort } from 'node:worker_threads'
|
||||
import { HostToWorkerType, WorkerToHostType } from '../src/protocol.ts'
|
||||
import type { HostToWorkerMessage, WorkerToHostMessage } from '../src/protocol.ts'
|
||||
import { requireParentPort, runWorkerSession } from '../src/session.ts'
|
||||
import type { ChildResult, WorkerInit } from '../src/types.ts'
|
||||
|
||||
/** Default limits for in-process sessions (concurrency pinned; auto is machine-derived). */
|
||||
function limits(overrides?: Partial<WorkerInit['limits']>): WorkerInit['limits'] {
|
||||
return { maxConcurrentAgents: 8, maxTotalAgents: 1000, maxItemsPerCall: 4096, syncTimeoutMs: 5000, ...overrides }
|
||||
}
|
||||
|
||||
/** Wrap a body in the minimal valid meta header (the session receives it pre-extracted). */
|
||||
function init(body: string, args?: unknown, limitOverrides?: Partial<WorkerInit['limits']>): WorkerInit {
|
||||
return {
|
||||
meta: { name: 'test-flow', description: 'a test workflow' },
|
||||
body,
|
||||
...args !== undefined ? { args } : {},
|
||||
limits: limits(limitOverrides),
|
||||
}
|
||||
}
|
||||
|
||||
/** One scripted host over the other end of a MessageChannel. */
|
||||
interface FakeHost {
|
||||
port: MessagePort
|
||||
messages: WorkerToHostMessage[]
|
||||
/** Messages of one type, as they arrive. */
|
||||
ofType<T extends WorkerToHostMessage['type']>(type: T): Extract<WorkerToHostMessage, { type: T }>[]
|
||||
send(message: HostToWorkerMessage): void
|
||||
/** Resolves with the terminal result message. */
|
||||
result(): Promise<Extract<WorkerToHostMessage, { type: 'result' }>['result']>
|
||||
close(): void
|
||||
}
|
||||
|
||||
interface FakeHostOptions {
|
||||
/** Auto-respond to child-start: reply started + settled per child index. Omit a reply to leave the child pending. */
|
||||
reply?: (request: { prompt: string; schema?: unknown; model?: string }, index: number) => ChildResult | undefined
|
||||
/** Reject the start instead (child-start-error) when returning a string. */
|
||||
refuse?: (index: number) => string | undefined
|
||||
/** Auto-send `go` on `ready` (default true). */
|
||||
go?: boolean
|
||||
/** Manual mode: do NOT auto-answer child-start at all (the test scripts the replies). */
|
||||
manual?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Drive runWorkerSession IN-PROCESS over a MessageChannel: this is where the
|
||||
* worker-side files earn their coverage — code inside a real Worker is
|
||||
* invisible to main-process coverage. The fake host mirrors the real host's
|
||||
* protocol discipline (one started/start-error per start; settled/disposed
|
||||
* follow).
|
||||
*/
|
||||
function fakeHost(options?: FakeHostOptions): FakeHost {
|
||||
const channel = new MessageChannel()
|
||||
const messages: WorkerToHostMessage[] = []
|
||||
const resultGate = Promise.withResolvers<Extract<WorkerToHostMessage, { type: 'result' }>['result']>()
|
||||
let childIndex = 0
|
||||
channel.port1.on('message', (message: WorkerToHostMessage) => {
|
||||
messages.push(message)
|
||||
switch (message.type) {
|
||||
case WorkerToHostType.Ready:
|
||||
if (options?.go !== false) channel.port1.postMessage({ type: HostToWorkerType.Go } satisfies HostToWorkerMessage)
|
||||
break
|
||||
case WorkerToHostType.ChildStart: {
|
||||
if (options?.manual) break
|
||||
const index = childIndex
|
||||
childIndex += 1
|
||||
const refusal = options?.refuse?.(index)
|
||||
if (refusal !== undefined) {
|
||||
channel.port1.postMessage(
|
||||
{ type: HostToWorkerType.ChildStartError, callId: message.callId, rendered: refusal } satisfies HostToWorkerMessage,
|
||||
)
|
||||
break
|
||||
}
|
||||
channel.port1.postMessage({ type: HostToWorkerType.ChildStarted, callId: message.callId, childId: `child-${index}` } satisfies HostToWorkerMessage)
|
||||
const reply = options?.reply?.(message.request, index)
|
||||
if (reply !== undefined) {
|
||||
channel.port1.postMessage(
|
||||
{ type: HostToWorkerType.ChildSettled, callId: message.callId, result: reply } satisfies HostToWorkerMessage,
|
||||
)
|
||||
}
|
||||
break
|
||||
}
|
||||
case WorkerToHostType.ChildDispose:
|
||||
channel.port1.postMessage({ type: HostToWorkerType.ChildDisposed, callId: message.callId } satisfies HostToWorkerMessage)
|
||||
break
|
||||
case WorkerToHostType.Result:
|
||||
resultGate.resolve(message.result)
|
||||
break
|
||||
default:
|
||||
break
|
||||
}
|
||||
})
|
||||
return {
|
||||
port: channel.port2,
|
||||
messages,
|
||||
ofType: type => messages.filter((message): message is never => message.type === type),
|
||||
send: (message) => { channel.port1.postMessage(message) },
|
||||
result: () => resultGate.promise,
|
||||
close: () => { channel.port1.close() },
|
||||
}
|
||||
}
|
||||
|
||||
/** A completed text child result. */
|
||||
function text(reply: string): ChildResult {
|
||||
return { output: [{ type: 'text', text: reply }], stopReason: 'completed' }
|
||||
}
|
||||
|
||||
describe('runWorkerSession over an in-process MessageChannel', () => {
|
||||
it('runs a script end to end: ready/go handshake, phases, log, agents, result', async () => {
|
||||
const host = fakeHost({ reply: (_request, index) => text(`answer-${index}`) })
|
||||
const session = runWorkerSession(host.port, init(`
|
||||
phase('Scan')
|
||||
log('starting with ' + args.files.length + ' files')
|
||||
const answers = await pipeline(args.files, (prev, item) => agent('read ' + item))
|
||||
return { answers }
|
||||
`, { files: ['a.ts', 'b.ts'] }))
|
||||
const result = await host.result()
|
||||
await session
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.agentsStarted).toBe(2)
|
||||
expect(result.value).toEqual({ answers: ['answer-0', 'answer-1'] })
|
||||
expect(host.messages[0]!.type).toBe('ready')
|
||||
expect(host.ofType(WorkerToHostType.Phase).map(m => m.title)).toEqual(['Scan'])
|
||||
expect(host.ofType(WorkerToHostType.Log).map(m => m.message)).toEqual(['starting with 2 files'])
|
||||
expect(host.ofType(WorkerToHostType.AgentStart).map(m => m.info.childId)).toEqual(['child-0', 'child-1'])
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd).every(m => m.info.outcome === 'completed')).toBe(true)
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('agent({schema}) forwards the schema on the start request and returns the structured value', async () => {
|
||||
const host = fakeHost({ reply: () => ({ output: [], structured: { files: ['x.ts'] }, stopReason: 'completed' }) })
|
||||
void runWorkerSession(host.port, init(`
|
||||
const found = await agent('list files', { schema: { type: 'object', properties: { files: { type: 'array', items: { type: 'string' } } } }, model: 'deepseek-v4-pro' })
|
||||
return { first: found.files[0] }
|
||||
`))
|
||||
const result = await host.result()
|
||||
expect(result.value).toEqual({ first: 'x.ts' })
|
||||
const start = host.ofType(WorkerToHostType.ChildStart)[0]!
|
||||
expect(start.request.schema).toEqual({ type: 'object', properties: { files: { type: 'array', items: { type: 'string' } } } })
|
||||
expect(start.request.model).toBe('deepseek-v4-pro')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a schema child completing WITHOUT a structured value resolves null with a failed outcome', async () => {
|
||||
const host = fakeHost({ reply: () => text('prose, no structure') })
|
||||
void runWorkerSession(host.port, init("return await agent('p', { schema: { type: 'object' } })"))
|
||||
const result = await host.result()
|
||||
expect(result.value).toBeNull()
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('failed')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a child settling non-completed resolves null (scripts filter), never throwing into the script', async () => {
|
||||
const host = fakeHost({ reply: (_request, index) => index === 0 ? { output: [], stopReason: 'error' } : text('ok') })
|
||||
void runWorkerSession(host.port, init("return await parallel([() => agent('one'), () => agent('two')])"))
|
||||
const result = await host.result()
|
||||
expect(result.value).toEqual([null, 'ok'])
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd).map(m => m.info.outcome)).toEqual(expect.arrayContaining(['failed', 'completed']))
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a start refusal (child-start-error) is a fatal AGENT_START that kills the script through a combinator', async () => {
|
||||
const host = fakeHost({ refuse: () => 'no provider here' })
|
||||
void runWorkerSession(host.port, init("return await pipeline([1], () => agent('p'))"))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('agent() could not start a child')
|
||||
expect(result.error).toContain('no provider here')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a child-failed message (infrastructure rejection) is fatal AGENT_RESULT with the paired failed outcome', async () => {
|
||||
const host = fakeHost()
|
||||
void runWorkerSession(host.port, init(`
|
||||
try { await agent('p'); return 'unreachable' } catch (e) { return { name: e.name, code: e.code, fatal: e.fatal } }
|
||||
`))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
|
||||
host.send({ type: HostToWorkerType.ChildFailed, callId, rendered: 'backend exploded' })
|
||||
const result = await host.result()
|
||||
expect(result.value).toMatchObject({ name: 'WorkflowError', code: 'AGENT_RESULT', fatal: true })
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('failed')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('cancel before go: the body never runs at all and the result is cancelled (a second cancel is a no-op)', async () => {
|
||||
const host = fakeHost({ go: false })
|
||||
const session = runWorkerSession(host.port, init("log('ran')\nreturn 123"))
|
||||
await vi.waitFor(() => { expect(host.messages.some(m => m.type === WorkerToHostType.Ready)).toBe(true) })
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'aborted before start' })
|
||||
// Idempotence: the first reason wins; a duplicate cancel changes nothing.
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'a later reason that must lose' })
|
||||
const result = await host.result()
|
||||
await session
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('aborted before start')
|
||||
expect(result.error).not.toContain('must lose')
|
||||
expect(result.value).toBeNull()
|
||||
expect(host.ofType(WorkerToHostType.Log)).toEqual([])
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a script with no return value resolves value: null', async () => {
|
||||
const host = fakeHost({ reply: () => text('ok') })
|
||||
void runWorkerSession(host.port, init("await agent('p')"))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toBeNull()
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('cancel mid-run: in-flight children get cancel RPCs, hooks throw at entry, the run reports cancelled', async () => {
|
||||
const host = fakeHost()
|
||||
void runWorkerSession(host.port, init(`
|
||||
phase('before')
|
||||
try { await agent('x') } catch (e) {}
|
||||
try { phase('after') } catch (e) {}
|
||||
try { log('after') } catch (e) {}
|
||||
try { await parallel([() => 'ran']) } catch (e) {}
|
||||
try { await pipeline(['item'], p => p) } catch (e) {}
|
||||
return 'survived by catching'
|
||||
`))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'stop everything' })
|
||||
// The real host settles the aborted child; mirror it.
|
||||
host.send({ type: HostToWorkerType.ChildSettled, callId, result: { output: [], stopReason: 'aborted' } })
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('stop everything')
|
||||
expect(host.ofType(WorkerToHostType.ChildCancel).map(m => m.callId)).toContain(callId)
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('cancelled')
|
||||
// No post-cancel narration left the runtime (the hooks threw at entry).
|
||||
expect(host.ofType(WorkerToHostType.Phase).map(m => m.title)).toEqual(['before'])
|
||||
expect(host.ofType(WorkerToHostType.Log)).toEqual([])
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('cancellation between a queued waiter and its slot: the waiter rejects without a child-start', async () => {
|
||||
const host = fakeHost({ go: true })
|
||||
void runWorkerSession(host.port, init(
|
||||
"return await parallel([() => agent('a'), () => agent('b')])",
|
||||
undefined,
|
||||
{ maxConcurrentAgents: 1 },
|
||||
))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'raced' })
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
// Only the first agent ever reached the host.
|
||||
expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1)
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a stray (never-awaited) agent is reaped after settlement: cancel + dispose RPCs flow, no unhandled rejection', async () => {
|
||||
const unhandled: unknown[] = []
|
||||
const onUnhandled = (reason: unknown): void => { unhandled.push(reason) }
|
||||
process.on('unhandledRejection', onUnhandled)
|
||||
try {
|
||||
const host = fakeHost()
|
||||
void runWorkerSession(host.port, init(`
|
||||
agent('stray, never awaited')
|
||||
return 'done without awaiting'
|
||||
`))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('completed')
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
|
||||
host.send({ type: HostToWorkerType.ChildSettled, callId, result: { output: [], stopReason: 'aborted' } })
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildDispose).map(m => m.callId)).toContain(callId) })
|
||||
await new Promise(resolve => setTimeout(resolve, 20))
|
||||
expect(unhandled).toEqual([])
|
||||
host.close()
|
||||
} finally {
|
||||
process.off('unhandledRejection', onUnhandled)
|
||||
}
|
||||
})
|
||||
|
||||
it('an unparseable body settles an error result instead of dying without one (host pre-parse skew guard)', async () => {
|
||||
const host = fakeHost()
|
||||
await runWorkerSession(host.port, init('return ((('))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('does not parse')
|
||||
expect(result.agentsStarted).toBe(0)
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a synchronous spin in the initial slice dies by the in-worker vm timeout', async () => {
|
||||
const host = fakeHost()
|
||||
void runWorkerSession(host.port, init('while (true) {}', undefined, { syncTimeoutMs: 50 }))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error?.toLowerCase()).toContain('timed out')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a non-JSON return value fails loud as RESULT_UNSERIALIZABLE', async () => {
|
||||
const host = fakeHost()
|
||||
void runWorkerSession(host.port, init('return { when: new Date(0) }'))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('not plain JSON data')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('tolerates replies for unknown callIds (a teardown race): nothing crashes, the run completes', async () => {
|
||||
const host = fakeHost({ reply: () => text('fine') })
|
||||
void runWorkerSession(host.port, init("return await agent('p')"))
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId: 999, childId: 'ghost' })
|
||||
host.send({ type: HostToWorkerType.ChildStartError, callId: 999, rendered: 'ghost' })
|
||||
host.send({ type: HostToWorkerType.ChildSettled, callId: 999, result: text('ghost') })
|
||||
host.send({ type: HostToWorkerType.ChildFailed, callId: 999, rendered: 'ghost' })
|
||||
host.send({ type: HostToWorkerType.ChildDisposed, callId: 999 })
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toBe('fine')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('caps and malformed hook arguments reject loud (the runtime runs unchanged inside the session)', async () => {
|
||||
const cases: [string, string][] = [
|
||||
['return await agent(42)', 'non-empty prompt string'],
|
||||
["return await agent('')", 'non-empty prompt string'],
|
||||
["return await agent('p', 'opts')", 'options must be an object'],
|
||||
["return await agent('p', { label: 3 })", '"label" must be a string'],
|
||||
["return await agent('p', { get label() { throw new Error('read failed') } })", 'options must be plain JSON data'],
|
||||
["return await agent('p', { bogus: true })", '"bogus" is not recognized'],
|
||||
["return await agent('p', { effort: 'high' })", '"effort" is deferred'],
|
||||
["return await agent('p', { schema: { type: 'object', oneOf: [] } })", 'outside the supported subset'],
|
||||
['return await parallel([() => 1, () => 2, () => 3])', 'over the per-call cap (2)'],
|
||||
['return await pipeline([1, 2, 3], (x) => x)', 'maxItemsPerCall'],
|
||||
["return await parallel('no')", 'parallel() requires an array'],
|
||||
['return await parallel([3])', 'item 0 is not a function'],
|
||||
["return await pipeline('no', () => 1)", 'pipeline() requires an items array'],
|
||||
['return await pipeline([1])', 'at least one stage'],
|
||||
["return await pipeline([1], 'x')", 'stage 0 is not a function'],
|
||||
["phase('')", 'phase() requires a non-empty title string'],
|
||||
['log(3)', 'log() requires a message string'],
|
||||
]
|
||||
for (const [body, expected] of cases) {
|
||||
const host = fakeHost({ reply: () => text('ok') })
|
||||
void runWorkerSession(host.port, init(body, undefined, { maxItemsPerCall: 2 }))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain(expected)
|
||||
host.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('combinator semantics: thunk/stage throws null the item; a forged fatal-shaped object stays null; real fatals propagate', async () => {
|
||||
const host = fakeHost({ reply: () => text('fine') })
|
||||
void runWorkerSession(host.port, init(`
|
||||
const viaParallel = await parallel([
|
||||
() => { throw new Error('boom') },
|
||||
() => agent('fine'),
|
||||
() => 'plain value',
|
||||
() => { throw { name: 'WorkflowError', fatal: true, message: 'forged fatal' } },
|
||||
])
|
||||
const viaPipeline = await pipeline([10, 20],
|
||||
(prev, item, index) => { if (item === 10) throw new Error('ordinary failure'); return 'kept-' + item + '-' + index },
|
||||
)
|
||||
return { viaParallel, viaPipeline }
|
||||
`))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toEqual({
|
||||
viaParallel: [null, 'fine', 'plain value', null],
|
||||
viaPipeline: [null, 'kept-20-1'],
|
||||
})
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('trips the total-agent cap with a message naming the config knob', async () => {
|
||||
const host = fakeHost({ reply: () => text('ok') })
|
||||
void runWorkerSession(host.port, init("await agent('1'); await agent('2'); await agent('3')", undefined, { maxTotalAgents: 2 }))
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('total agent cap (2)')
|
||||
expect(result.agentsStarted).toBe(2)
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('queued agents proceed through the concurrency semaphore in FIFO order', async () => {
|
||||
const host = fakeHost({ reply: request => text(`ok:${request.prompt}`) })
|
||||
void runWorkerSession(host.port, init(
|
||||
"return await parallel([1, 2, 3].map((n) => () => agent('job ' + n)))",
|
||||
undefined,
|
||||
{ maxConcurrentAgents: 1 },
|
||||
))
|
||||
const result = await host.result()
|
||||
expect(result.value).toEqual(['ok:job 1', 'ok:job 2', 'ok:job 3'])
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('labels default from the prompt first line, truncated; explicit label/phase options win', async () => {
|
||||
const host = fakeHost({ reply: () => text('ok') })
|
||||
void runWorkerSession(host.port, init(`
|
||||
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
|
||||
`))
|
||||
await host.result()
|
||||
const starts = host.ofType(WorkerToHostType.AgentStart).map(m => m.info)
|
||||
expect(starts[0]).toMatchObject({ seq: 1, phase: 'Find' })
|
||||
expect(starts[0]!.label.length).toBeLessThanOrEqual(48)
|
||||
expect(starts[0]!.label).not.toContain('second line')
|
||||
expect(starts[1]).toMatchObject({ seq: 2, label: 'named', phase: 'Custom' })
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('non-text output blocks are filtered out of the text result', async () => {
|
||||
const host = fakeHost({
|
||||
reply: () => ({
|
||||
output: [
|
||||
{ type: 'text', text: 'first ' },
|
||||
{ type: 'tool_call', id: 'c1', name: 'x', arguments: {} } as never,
|
||||
{ type: 'text', text: 'second' },
|
||||
],
|
||||
stopReason: 'completed',
|
||||
}),
|
||||
})
|
||||
void runWorkerSession(host.port, init("return await agent('p')"))
|
||||
const result = await host.result()
|
||||
expect(result.value).toBe('first second')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a cancel landing DURING the start round-trip winds the fresh child down (cancel + dispose) and dies cancelled', async () => {
|
||||
const host = fakeHost({ manual: true })
|
||||
void runWorkerSession(host.port, init("return await agent('p')"))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
// Cancel FIRST, then the (stale) started reply: the worker processes them
|
||||
// in order, so the agent() continuation resumes already-cancelled — the
|
||||
// window the real host cannot produce (it refuses starts once cancelled)
|
||||
// but a teardown race can.
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'raced the start' })
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
await vi.waitFor(() => {
|
||||
expect(host.ofType(WorkerToHostType.ChildCancel).map(m => m.callId)).toContain(callId)
|
||||
expect(host.ofType(WorkerToHostType.ChildDispose).map(m => m.callId)).toContain(callId)
|
||||
})
|
||||
// The child never became an agent-start: it was wound down pre-lifecycle.
|
||||
expect(host.ofType(WorkerToHostType.AgentStart)).toEqual([])
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a start refusal arriving after a cancel reads as the cancellation, not a broken seam', async () => {
|
||||
const host = fakeHost({ manual: true })
|
||||
void runWorkerSession(host.port, init(`
|
||||
try { await agent('p'); return 'unreachable' } catch (e) { return { code: e.code } }
|
||||
`))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'stopping' })
|
||||
host.send({ type: HostToWorkerType.ChildStartError, callId, rendered: 'workflow run cancelled: stopping' })
|
||||
const result = await host.result()
|
||||
// The run reports cancelled (the script died of CANCELLED, not AGENT_START).
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
host.close()
|
||||
})
|
||||
|
||||
it('a child result rejection while cancelled pairs a cancelled agent-end, and the run reports cancelled', async () => {
|
||||
const host = fakeHost({ manual: true })
|
||||
void runWorkerSession(host.port, init("return await agent('doomed')"))
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
|
||||
const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
|
||||
host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
|
||||
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.AgentStart).length).toBe(1) })
|
||||
host.send({ type: HostToWorkerType.Cancel, reason: 'user aborted' })
|
||||
host.send({ type: HostToWorkerType.ChildFailed, callId, rendered: 'backend crashed on abort' })
|
||||
const result = await host.result()
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('cancelled')
|
||||
host.close()
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
describe('the worker bootstrap', () => {
|
||||
it('requireParentPort narrows a real port and throws on the main thread', () => {
|
||||
const channel = new MessageChannel()
|
||||
expect(requireParentPort(channel.port1)).toBe(channel.port1)
|
||||
channel.port1.close()
|
||||
expect(() => requireParentPort(null)).toThrow(/inside a worker thread/)
|
||||
})
|
||||
|
||||
it('the entry module itself throws when loaded on the main thread (no parentPort)', async () => {
|
||||
// This import EXECUTES ../src/worker.ts on the main thread, which is what
|
||||
// covers the bootstrap file: requireParentPort throws before
|
||||
// runWorkerSession is reached.
|
||||
await expect(import('../src/worker.ts')).rejects.toThrow(/inside a worker thread/)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,102 @@
|
||||
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 WorkerWorkflowEngine from '../src/index.ts'
|
||||
|
||||
/**
|
||||
* With-key e2e: a REAL script in a REAL worker thread
|
||||
* 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' })
|
||||
await built.plugin(WorkerWorkflowEngine, { provider: 'spawn' })
|
||||
return built
|
||||
}
|
||||
|
||||
const META = {
|
||||
name: 'e2e-worker-arithmetic',
|
||||
description: 'two real children through a worker thread: one prose, one structured',
|
||||
phases: [{ title: 'Ask' }, { title: 'Judge' }],
|
||||
}
|
||||
const SCRIPT = `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)('worker workflow engine with-key e2e', () => {
|
||||
it('runs a two-phase script in a worker thread over real children, one through the structured runtime', async () => {
|
||||
ctx = await harness()
|
||||
const parentHandle = ctx.agents.create({
|
||||
agentId: AgentId('wf-worker-e2e-parent'),
|
||||
sessionId: 'wf-worker-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, meta: META, 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)
|
||||
})
|
||||
@@ -0,0 +1,940 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
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 { WorkflowMeta, WorkflowResult, WorkflowResultInfo, WorkflowRunInfo } from '@deepseek-ai/dsh-workflow'
|
||||
import * as workerEngineModule from '../src/index.ts'
|
||||
import WorkerWorkflowEngine, { 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
|
||||
}
|
||||
|
||||
// Worker-thread startup is CPU-bound (a fresh thread compiles the runtime on
|
||||
// every start): on a contended CI runner it regularly blows past vitest's 5s
|
||||
// default test timeout, observed repeatedly on the coverage lane.
|
||||
vi.setConfig({ testTimeout: 30_000 })
|
||||
|
||||
/**
|
||||
* `vi.waitFor` with a contention-proof default timeout: the 1s default
|
||||
* flaked repeatedly on the CI coverage lane, where worker-thread cold start
|
||||
* (CPU-bound — a fresh thread compiles the runtime) competes with three
|
||||
* sibling vitest workers for CPU. The 10s default is for exactly those
|
||||
* races — waiting for a worker to start, run its first script line, or
|
||||
* deliver an async child-registration message to the host. It is NOT for a
|
||||
* wait that asserts the HOST reacted PROMPTLY to something that already
|
||||
* happened (a settled result, an observed worker death): those keep an
|
||||
* explicit tight override below, or the generous default would silently
|
||||
* accept a multi-second regression in host-side reap latency as passing
|
||||
* (proven by injecting a 6s delay into one such reap and watching the
|
||||
* un-overridden version of this helper still pass in ~6s).
|
||||
* @param assertion - retried until it stops throwing or the timeout elapses.
|
||||
* @param timeout - override for a wait that must stay deliberately tight.
|
||||
* @returns resolves when the assertion passes.
|
||||
*/
|
||||
function waitFor(assertion: () => void, timeout = 10_000): Promise<void> {
|
||||
return vi.waitFor(assertion, { timeout, interval: 50 })
|
||||
}
|
||||
|
||||
/** The vm-context escape hatch, spelled once: real Worker tests use it to make the WORKER misbehave. */
|
||||
const ESCAPE = "globalThis.constructor.constructor('return process')()"
|
||||
|
||||
/** One controllable child run: the test (or auto mode) settles it. */
|
||||
interface ControlledRun {
|
||||
request: SubagentStartRequest
|
||||
settle(result: SubagentResult): void
|
||||
cancelled: string | undefined
|
||||
disposed: boolean
|
||||
disposeCalls: number
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. 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, persona: false }
|
||||
readonly inheritsParentContext = false
|
||||
readonly runs: ControlledRun[] = []
|
||||
|
||||
constructor(
|
||||
readonly name: string,
|
||||
private readonly reply?: (request: SubagentStartRequest, index: number) => SubagentResult,
|
||||
private readonly disposeDelayMs = 0,
|
||||
) {}
|
||||
|
||||
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, disposeCalls: 0 }
|
||||
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.disposeCalls += 1
|
||||
if (this.disposeDelayMs === 0) {
|
||||
controlled.disposed = true
|
||||
return Promise.resolve()
|
||||
}
|
||||
return new Promise<void>((resolve) => {
|
||||
setTimeout(() => {
|
||||
controlled.disposed = true
|
||||
resolve()
|
||||
}, this.disposeDelayMs)
|
||||
})
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 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
|
||||
disposeDelayMs?: number
|
||||
}
|
||||
|
||||
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')),
|
||||
options?.disposeDelayMs ?? 0,
|
||||
)
|
||||
ctx.subagents.registerProvider(provider)
|
||||
// A fixed concurrency ceiling: the auto-resolved default is machine-derived
|
||||
// (cores - 2, floored at 1), so tests that expect N children in flight
|
||||
// would wedge on small CI runners.
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'stub', maxConcurrentAgents: 8, ...options?.config })
|
||||
return { ctx, provider, parent: fakeParent() }
|
||||
}
|
||||
|
||||
/** The standard test meta plus a body, spread into a start request. */
|
||||
function scripted(body: string, metaExtra?: Partial<WorkflowMeta>): { script: string; meta: WorkflowMeta } {
|
||||
return { script: body, meta: { name: 'test-flow', description: 'a test workflow', ...metaExtra } }
|
||||
}
|
||||
|
||||
/** Start + await one run, disposing on the way out. */
|
||||
async function run(ctx: Context, parent: Agent, source: { script: string; meta: WorkflowMeta }, args?: unknown): Promise<WorkflowResult> {
|
||||
const handle = ctx.workflows.start({ ...source, parent, ...args !== undefined ? { args } : {} })
|
||||
try {
|
||||
return await handle.result
|
||||
} finally {
|
||||
await handle.dispose()
|
||||
}
|
||||
}
|
||||
|
||||
describe('dsh-workflow-workerthread', () => {
|
||||
describe('script execution over a real worker thread', () => {
|
||||
it('runs a script end-to-end: agent() text results, phases, log, args, return value, events', 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, scripted(`
|
||||
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({schema, model}) forwards outputSchema and agentOptions to the provider across the thread', async () => {
|
||||
const { ctx, parent, provider } = await setup({
|
||||
reply: () => ({ output: [], structured: { files: ['x.ts', 'y.ts'] }, stopReason: 'completed' }),
|
||||
})
|
||||
const result = await run(ctx, parent, scripted(`
|
||||
const found = await agent('list files', { model: 'deepseek-v4-pro', 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'],
|
||||
})
|
||||
expect(provider.runs[0]!.request.agentOptions).toEqual({ model: 'deepseek-v4-pro' })
|
||||
expect(provider.runs[0]!.request.parent).toBeDefined()
|
||||
})
|
||||
|
||||
it('a fatal hook error inside the worker kills the script and reports the error', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const result = await run(ctx, parent, scripted("return await parallel([() => agent('x', { isolation: 'worktree' })])"))
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('"isolation" is deferred')
|
||||
})
|
||||
|
||||
it('a provider start failure crosses back as a fatal AGENT_START error', async () => {
|
||||
const { ctx, parent } = await setup({ config: { provider: 'nonexistent' } })
|
||||
const result = await run(ctx, parent, scripted("return await pipeline([1], () => agent('p'))"))
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('agent() could not start a child')
|
||||
})
|
||||
|
||||
it('a child result REJECTION crosses back as a fatal AGENT_RESULT error (a broken provider is not a failed child)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const provider: SubagentProvider = {
|
||||
name: 'rejecting',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('reject-child'),
|
||||
result: Promise.reject(new Error('backend exploded')),
|
||||
cancel: () => { /* nothing in flight */ },
|
||||
dispose: () => Promise.resolve(),
|
||||
}),
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'rejecting', maxConcurrentAgents: 2 })
|
||||
const result = await run(ctx, fakeParent(), scripted(`
|
||||
try { await agent('p'); return 'unreachable' } catch (e) { return { name: e.name, code: e.code, fatal: e.fatal, message: e.message } }
|
||||
`))
|
||||
expect(result.value).toMatchObject({ name: 'WorkflowError', code: 'AGENT_RESULT', fatal: true })
|
||||
expect((result.value as { message: string }).message).toContain('backend exploded')
|
||||
})
|
||||
|
||||
it('a child whose dispose() rejects cannot wedge the script (the host acks anyway)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const provider: SubagentProvider = {
|
||||
name: 'bad-dispose',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('bad-dispose-child'),
|
||||
result: Promise.resolve({ output: [{ type: 'text', text: 'fine' }], stopReason: 'completed' }),
|
||||
cancel: () => { /* settled already */ },
|
||||
dispose: () => Promise.reject(new Error('dispose exploded')),
|
||||
}),
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'bad-dispose', maxConcurrentAgents: 2 })
|
||||
const result = await run(ctx, fakeParent(), scripted("return await agent('p')"))
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toBe('fine')
|
||||
})
|
||||
|
||||
it('a child dispose() rejecting an UNRENDERABLE value still acks — the containment warn is total', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const provider: SubagentProvider = {
|
||||
name: 'coercion-trap-dispose',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('trap-child'),
|
||||
result: Promise.resolve({ output: [{ type: 'text', text: 'fine' }], stopReason: 'completed' }),
|
||||
cancel: () => { /* settled already */ },
|
||||
// The rejection VALUE's own coercion throws: a warn built with bare
|
||||
// String(error) would itself throw, skipping the ChildDisposed ack
|
||||
// and wedging the script's finally until the grace/terminate path.
|
||||
// eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- the non-Error rejection IS the scenario under test
|
||||
dispose: () => Promise.reject({ toString: () => { throw new Error('coercion trap') } }),
|
||||
}),
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'coercion-trap-dispose', maxConcurrentAgents: 2 })
|
||||
const result = await run(ctx, fakeParent(), scripted("return await agent('p')"))
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toBe('fine')
|
||||
})
|
||||
|
||||
it('the worker spawns with an EMPTY environment: an escaped script finds no ambient credentials', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
// A canary in the HARNESS process's env: with an inherited environment
|
||||
// the escape below would read it back (exactly how DEEPSEEK_API_KEY
|
||||
// would leak); env: {} in the spawn options is what keeps it out.
|
||||
process.env.WORKFLOW_ENV_CANARY = 'leak me'
|
||||
try {
|
||||
const result = await run(ctx, parent, scripted(`
|
||||
const proc = ${ESCAPE}
|
||||
return { canary: proc.env.WORKFLOW_ENV_CANARY ?? null, keys: Object.keys(proc.env).length }
|
||||
`))
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toEqual({ canary: null, keys: 0 })
|
||||
} finally {
|
||||
delete process.env.WORKFLOW_ENV_CANARY
|
||||
}
|
||||
})
|
||||
|
||||
it('the unbuilt worker forwards exactly TSX_TSCONFIG_PATH through the scrub: the paths-map pin survives, secrets do not', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
// The ACP snapshot harness runs the parent with its cwd OUTSIDE the
|
||||
// repo and pins the repo tsconfig through this variable; the worker
|
||||
// must inherit the pin (or its dsh-* imports silently resolve to
|
||||
// unbuilt lib/ bundles) while every other variable stays scrubbed.
|
||||
const tsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url))
|
||||
process.env.TSX_TSCONFIG_PATH = tsconfig
|
||||
process.env.WORKFLOW_ENV_CANARY = 'leak me'
|
||||
try {
|
||||
const result = await run(ctx, parent, scripted(`
|
||||
const proc = ${ESCAPE}
|
||||
return { keys: Object.keys(proc.env), tsconfig: proc.env.TSX_TSCONFIG_PATH }
|
||||
`))
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toEqual({ keys: ['TSX_TSCONFIG_PATH'], tsconfig })
|
||||
} finally {
|
||||
delete process.env.TSX_TSCONFIG_PATH
|
||||
delete process.env.WORKFLOW_ENV_CANARY
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('lifecycle: parse errors, cancellation, termination, disposal', () => {
|
||||
it('start() throws synchronously for invalid meta data or an unparseable body (host-side pre-checks)', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
// Meta is DATA — shape violations reject loud, every one named.
|
||||
expect(() => ctx.workflows.start({ script: 'return 1', meta: { name: '', description: 'd' }, parent })).toThrow(/meta\.name must be a non-empty string/)
|
||||
expect(() => ctx.workflows.start({ script: 'return 1', meta: { name: 'x', description: 'd', extra: 1 } as unknown as WorkflowMeta, parent })).toThrow(/META_INVALID|not a recognized field/)
|
||||
expect(() => ctx.workflows.start({ ...scripted('return ((('), parent })).toThrow(/does not parse/)
|
||||
// The likeliest authoring slip — a Claude Code-style meta header in the
|
||||
// body — gets a pointed message, not a bare SyntaxError.
|
||||
expect(() => ctx.workflows.start({ ...scripted("export const meta = { name: 'x', description: 'd' }\nreturn 1"), parent })).toThrow(/meta rides the `meta` request field/)
|
||||
})
|
||||
|
||||
it('cancel() aborts in-flight children (signal AND cancel RPC) and settles the run cancelled', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const ends: unknown[] = []
|
||||
ctx.on('workflow/agent-end', (_info, agent) => { ends.push(agent) })
|
||||
const runEnds: WorkflowResultInfo[] = []
|
||||
ctx.on('workflow/end', (_info, result) => { runEnds.push(result) })
|
||||
const handle = ctx.workflows.start({ ...scripted("return await agent('long job')"), parent })
|
||||
await 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')
|
||||
await handle.dispose()
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
expect(ends).toEqual([expect.objectContaining({ seq: 1, outcome: 'cancelled' })])
|
||||
// workflow/end is an observer's only death signal: it fires for a
|
||||
// cancelled run too, mirroring the settled outcome data.
|
||||
expect(runEnds).toEqual([{ stopReason: 'cancelled', error: result.error, agentsStarted: result.agentsStarted }])
|
||||
})
|
||||
|
||||
it('an already-aborted request signal cancels before the body ever runs (the go handshake holds it)', async () => {
|
||||
const { ctx, parent, provider } = await setup()
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
const logs: string[] = []
|
||||
ctx.on('workflow/log', (_info, message) => { logs.push(message) })
|
||||
const handle = ctx.workflows.start({ ...scripted("log('ran')\nreturn 123"), parent, signal: controller.signal })
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.value).toBeNull()
|
||||
expect(logs).toEqual([])
|
||||
expect(provider.runs.length).toBe(0)
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('cancel() right after start() cancels before the body runs; the signal aborting mid-run cancels like cancel()', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const first = ctx.workflows.start({ ...scripted("return await agent('never')"), parent })
|
||||
// No-reason cancel: the canonical default reason must ride the result.
|
||||
first.cancel()
|
||||
const firstResult = await first.result
|
||||
expect(firstResult.stopReason).toBe('cancelled')
|
||||
expect(firstResult.error).toContain('workflow cancelled')
|
||||
expect(provider.runs.length).toBe(0)
|
||||
await first.dispose()
|
||||
|
||||
const controller = new AbortController()
|
||||
const second = ctx.workflows.start({ ...scripted("return await agent('job')"), parent, signal: controller.signal })
|
||||
await waitFor(() => { expect(provider.runs.length).toBe(1) })
|
||||
controller.abort()
|
||||
expect((await second.result).stopReason).toBe('cancelled')
|
||||
await second.dispose()
|
||||
})
|
||||
|
||||
it('a child-start racing the host cancel is refused: no child starts after cancellation', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
// Cancel from INSIDE the log listener: the worker has already posted
|
||||
// its child-start (queued right behind the log message), so the host
|
||||
// processes it with cancelReason set — the refusal arm no real-world
|
||||
// timing can hit reliably. (The closure runs only after `handle` below
|
||||
// is initialized — the listener fires on the worker's first message.)
|
||||
ctx.on('workflow/log', () => { handle.cancel('cancelled from the log listener') })
|
||||
const handle = ctx.workflows.start({ ...scripted("log('mark')\nreturn await agent('late')"), parent })
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(provider.runs.length).toBe(0)
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('post-cancel narration is suppressed host-side, and completion racing a cancel reports cancelled', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const narration: string[] = []
|
||||
ctx.on('workflow/log', (_info, message) => { narration.push(message) })
|
||||
ctx.on('workflow/phase', (_info, title) => { narration.push(`phase:${title}`) })
|
||||
const handle = ctx.workflows.start({
|
||||
// The sync spin keeps the worker's loop busy so the cancel message
|
||||
// cannot be processed before the script settles `completed` — the
|
||||
// worker posts a completed result that must LOSE to the in-flight
|
||||
// host cancellation. The trailing narration exercises host-side
|
||||
// suppression: posted pre-cancel-processing worker-side, arriving
|
||||
// post-cancel host-side.
|
||||
...scripted(`
|
||||
log('started')
|
||||
const end = Date.now() + 1000
|
||||
while (Date.now() < end) {}
|
||||
phase('late phase')
|
||||
log('late log')
|
||||
return 'done'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(narration).toContain('started') })
|
||||
handle.cancel('raced the completion')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('raced the completion')
|
||||
expect(narration).toEqual(['started'])
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('cancel() force-settles a script parked on a promise no hook owns, and TERMINATES its worker', async () => {
|
||||
const { ctx, parent } = await setup({ config: { provider: 'stub', disposeGraceMs: 50 } })
|
||||
const runEnds: WorkflowResultInfo[] = []
|
||||
ctx.on('workflow/end', (_info, result) => { runEnds.push(result) })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted("await new Promise(() => {})\nreturn 'unreachable'"),
|
||||
parent,
|
||||
})
|
||||
handle.cancel('user aborted')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('user aborted')
|
||||
// The grace force-settle fires workflow/end exactly like an ordinary
|
||||
// settlement — a terminated script's death still reaches observers.
|
||||
expect(runEnds).toEqual([{ stopReason: 'cancelled', error: result.error, agentsStarted: 0 }])
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('dispose() on a stuck script returns within the grace instead of hanging (result settles cancelled)', async () => {
|
||||
const { ctx, parent } = await setup({ config: { provider: 'stub', disposeGraceMs: 50 } })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted("await new Promise(() => {})\nreturn 'unreachable'"),
|
||||
parent,
|
||||
})
|
||||
const before = Date.now()
|
||||
await handle.dispose()
|
||||
expect(Date.now() - before).toBeLessThan(2000)
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
})
|
||||
|
||||
it('dispose() is idempotent and settles cleanly after a completed run', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const handle = ctx.workflows.start({ ...scripted('return 1'), parent })
|
||||
await handle.result
|
||||
await handle.dispose()
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('a settled run arms NO grace timer: disposing a completed run must not pin it for disposeGraceMs', async () => {
|
||||
// A distinctive grace so the spy can tell the cancel-path grace timer
|
||||
// apart from every other timeout in flight.
|
||||
const GRACE = 44_444
|
||||
const { ctx, parent } = await setup({ config: { provider: 'stub', disposeGraceMs: GRACE } })
|
||||
const handle = ctx.workflows.start({ ...scripted('return 1'), parent })
|
||||
await handle.result
|
||||
const spy = vi.spyOn(globalThis, 'setTimeout')
|
||||
try {
|
||||
await handle.dispose()
|
||||
// dispose()'s own bounded-wait sleep is the ONLY grace-sized timer
|
||||
// allowed here; before the settled guard, cancel() armed a second one
|
||||
// that nothing would ever clear (the run was already settled), keeping
|
||||
// the WorkerRun/Worker closure alive until the grace expired.
|
||||
const graceTimers = spy.mock.calls.filter(call => call[1] === GRACE)
|
||||
expect(graceTimers.length).toBe(1)
|
||||
} finally {
|
||||
spy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('strays: children fired without await are aborted once the script settles, and dispose() waits for their disposal', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true, disposeDelayMs: 40 })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted(`
|
||||
agent('stray')
|
||||
return 'done without awaiting'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('completed')
|
||||
await waitFor(() => { expect(provider.runs.length).toBe(1) })
|
||||
await handle.dispose()
|
||||
// Not a waitFor: by the time dispose() returns, the slow child disposal
|
||||
// must already be complete (host-side registry quiescence).
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
})
|
||||
|
||||
it('the settle-reap fires the request signal too: a provider honoring ONLY the signal winds its stray down promptly', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const aborted: string[] = []
|
||||
const provider: SubagentProvider = {
|
||||
name: 'signal-only',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: (request) => {
|
||||
let settle!: (result: SubagentResult) => void
|
||||
const result = new Promise<SubagentResult>((resolve) => { settle = resolve })
|
||||
request.signal?.addEventListener('abort', () => {
|
||||
aborted.push(String(request.signal?.reason))
|
||||
settle({ output: [], stopReason: 'aborted' })
|
||||
}, { once: true })
|
||||
return {
|
||||
id: AgentId('signal-only-child'),
|
||||
result,
|
||||
// The seam leaves a provider free to honor EITHER cancel channel;
|
||||
// this one deliberately ignores run.cancel() — only the request
|
||||
// signal can wind it down.
|
||||
cancel: () => { /* signal-only by design */ },
|
||||
dispose: () => Promise.resolve(),
|
||||
}
|
||||
},
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'signal-only', maxConcurrentAgents: 2 })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted(`
|
||||
agent('stray, never awaited')
|
||||
return 'done'
|
||||
`),
|
||||
parent: fakeParent(),
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('completed')
|
||||
// BEFORE dispose(): the settlement itself must have aborted the signal —
|
||||
// without it this child would stay live until dispose's terminate. This
|
||||
// is a HOST-PROMPTNESS claim, not a cold-start race — a tight explicit
|
||||
// bound (unlike the file default) so a multi-second reap regression
|
||||
// cannot pass by outlasting the wait.
|
||||
await waitFor(() => { expect(aborted).toEqual(['workflow settled']) }, 1000)
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it("cancel() drives each child's explicit cancel() host-side: a wedged worker cannot delay it", async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
let starts = 0
|
||||
const cancelled: string[] = []
|
||||
const provider: SubagentProvider = {
|
||||
name: 'cancel-only',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => {
|
||||
starts += 1
|
||||
return {
|
||||
id: AgentId('cancel-only-child'),
|
||||
result: new Promise(() => { /* only cancel() ends this child */ }),
|
||||
// Deliberately ignores the request signal — the seam leaves a
|
||||
// provider free to honor ONLY the explicit cancel() channel.
|
||||
cancel: (reason?: string) => { cancelled.push(reason ?? 'cancelled') },
|
||||
dispose: () => Promise.resolve(),
|
||||
}
|
||||
},
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
// A deliberately huge grace: if only the grace/terminate reap could
|
||||
// reach this child, the assertion below would time out first.
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'cancel-only', maxConcurrentAgents: 2, disposeGraceMs: 30_000 })
|
||||
const handle = ctx.workflows.start({
|
||||
// The stray child's start RPC reaches the host, then the script wedges
|
||||
// its own worker in a synchronous spin: the worker cannot process the
|
||||
// Cancel message, so it can relay NO ChildCancel RPC — only the host's
|
||||
// own children loop can deliver the explicit cancel in time. The
|
||||
// microtask yields let the agent() continuation POST its child-start
|
||||
// before the spin seizes the worker's loop (the posted message needs
|
||||
// no further worker-loop turns to reach the host).
|
||||
...scripted(`
|
||||
agent('wedged child')
|
||||
for (let i = 0; i < 20; i++) await null
|
||||
const end = Date.now() + 1500
|
||||
while (Date.now() < end) {}
|
||||
return 'raced'
|
||||
`),
|
||||
parent: fakeParent(),
|
||||
})
|
||||
await waitFor(() => { expect(starts).toBe(1) })
|
||||
handle.cancel('stop now')
|
||||
await waitFor(() => { expect(cancelled).toEqual(['stop now']) }, 800)
|
||||
// The wedged worker's own completion loses to the in-flight cancel.
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('dispose() on a wedged worker host-drives child disposal inside the grace: it returns with the children DISPOSED, not with their teardown still in flight', async () => {
|
||||
const { ctx, parent, provider } = await setup({
|
||||
manual: true,
|
||||
disposeDelayMs: 40,
|
||||
config: { provider: 'stub', maxConcurrentAgents: 8, disposeGraceMs: 400 },
|
||||
})
|
||||
const handle = ctx.workflows.start({
|
||||
// Same shape as the wedged-cancel test above: the child's start RPC
|
||||
// reaches the host, then the script seizes its worker's loop, so the
|
||||
// worker can relay NO dispose RPC — the host's own dispose() drive is
|
||||
// the only thing that can start (and finish) this child's disposal
|
||||
// before the grace runs out.
|
||||
...scripted(`
|
||||
agent('wedged child')
|
||||
for (let i = 0; i < 20; i++) await null
|
||||
const end = Date.now() + 1500
|
||||
while (Date.now() < end) {}
|
||||
return 'raced'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(provider.runs.length).toBe(1) })
|
||||
const before = Date.now()
|
||||
await handle.dispose()
|
||||
// Bounded by the grace (plus the terminate), never by the 1.5s spin.
|
||||
expect(Date.now() - before).toBeLessThan(1200)
|
||||
// Not a waitFor: dispose() resolving IS the quiescence claim — the slow
|
||||
// child disposal must be complete, not merely started (before the
|
||||
// host-driven drive, disposal only STARTED at the post-terminate reap,
|
||||
// so dispose() returned with it still in flight).
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
}, 15_000)
|
||||
|
||||
it('a live child disposed by the dispose() drive is disposed ONCE, and the worker\'s late dispose RPC still gets its ack (the script settles, not the grace)', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted(`
|
||||
await agent('long child')
|
||||
return 'unreachable'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(provider.runs.length).toBe(1) })
|
||||
const handleDispose = handle.dispose()
|
||||
const result = await handle.result
|
||||
// The script itself settled (the wrapper's own dispose RPC found the
|
||||
// child already reaped host-side and was acked) — a missing ack would
|
||||
// wedge the wrapper's finally until the 5s default grace force-settle.
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('workflow disposed')
|
||||
await handleDispose
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
// The memo: the host drive and the worker's RPC share one disposal.
|
||||
expect(provider.runs[0]!.disposeCalls).toBe(1)
|
||||
})
|
||||
|
||||
it('the grace force-settle pairs every stranded start: a host-synthesized cancelled agent-end lands before workflow/end', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true, config: { provider: 'stub', maxConcurrentAgents: 8, disposeGraceMs: 300 } })
|
||||
const ends: { seq: number; outcome: string }[] = []
|
||||
const order: string[] = []
|
||||
ctx.on('workflow/agent-start', (_info, agent) => { order.push(`start:${agent.seq}`) })
|
||||
ctx.on('workflow/agent-end', (_info, agent) => {
|
||||
ends.push({ seq: agent.seq, outcome: agent.outcome })
|
||||
order.push(`end:${agent.seq}`)
|
||||
})
|
||||
ctx.on('workflow/end', () => { order.push('run-end') })
|
||||
const handle = ctx.workflows.start({
|
||||
// 'slow' starts and its agent-start crosses to observers (the awaited
|
||||
// 'fast' call keeps the worker loop turning), then the script seizes
|
||||
// the loop: the wedged worker can never author slow's agent-end —
|
||||
// only the host's ledger can close the pair.
|
||||
...scripted(`
|
||||
const p = agent('slow')
|
||||
await agent('fast')
|
||||
const end = Date.now() + 1500
|
||||
while (Date.now() < end) {}
|
||||
return 'raced'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(order.filter(entry => entry.startsWith('start:')).length).toBe(2) })
|
||||
const fast = provider.runs.find(run => (run.request.prompt[0] as { text?: string }).text === 'fast')!
|
||||
fast.settle(text('fast done'))
|
||||
handle.cancel('stop now')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
// fast's end is the worker's own report; slow's is host-synthesized at
|
||||
// the force-settle — exactly one end per started seq, no third event.
|
||||
expect(ends).toEqual([
|
||||
{ seq: 2, outcome: 'completed' },
|
||||
{ seq: 1, outcome: 'cancelled' },
|
||||
])
|
||||
// Both ends reached observers BEFORE workflow/end: a progress consumer
|
||||
// can finalize its state at run-end without dangling agents.
|
||||
expect(order.indexOf('run-end')).toBe(order.length - 1)
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('graceful cancellation keeps pairing worker-authored: exactly one agent-end per start, nothing synthesized on top', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const ends: { seq: number; outcome: string }[] = []
|
||||
const order: string[] = []
|
||||
ctx.on('workflow/agent-end', (_info, agent) => {
|
||||
ends.push({ seq: agent.seq, outcome: agent.outcome })
|
||||
order.push(`end:${agent.seq}`)
|
||||
})
|
||||
ctx.on('workflow/end', () => { order.push('run-end') })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted("await parallel([() => agent('a'), () => agent('b')])\nreturn 'unreachable'"),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(provider.runs.length).toBe(2) })
|
||||
handle.cancel('user stop')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
// The live worker reported both pairs itself; the ledger must not add
|
||||
// a synthesized duplicate on any path that settles inside the grace.
|
||||
expect(ends.map(end => end.outcome)).toEqual(['cancelled', 'cancelled'])
|
||||
expect(new Set(ends.map(end => end.seq)).size).toBe(2)
|
||||
expect(order.indexOf('run-end')).toBe(order.length - 1)
|
||||
await handle.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('worker death', () => {
|
||||
it('a worker that exits before settling reports an error result and reaps its children', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
// The child's dispose() REJECTS on top of the worker death: the reap
|
||||
// must contain it (warn, not crash) while still emptying the registry.
|
||||
const cancelled: string[] = []
|
||||
const provider: SubagentProvider = {
|
||||
name: 'doomed',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('doomed-child'),
|
||||
result: new Promise(() => { /* never settles; the reap is the teardown */ }),
|
||||
cancel: (reason?: string) => { cancelled.push(reason ?? 'cancelled') },
|
||||
dispose: () => Promise.reject(new Error('dispose exploded during reap')),
|
||||
}),
|
||||
}
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(WorkerWorkflowEngine, { provider: 'doomed', maxConcurrentAgents: 2 })
|
||||
const runEnds: WorkflowResultInfo[] = []
|
||||
ctx.on('workflow/end', (_info, result) => { runEnds.push(result) })
|
||||
const handle = ctx.workflows.start({
|
||||
// The stray child's start RPC reaches the host, then the script kills
|
||||
// its own worker through the documented vm escape — the host must
|
||||
// settle `error` with the exit diagnostics and wind the child down.
|
||||
...scripted(`
|
||||
agent('doomed')
|
||||
const proc = ${ESCAPE}
|
||||
const st = globalThis.constructor.constructor('return setTimeout')()
|
||||
await new Promise(resolve => st(resolve, 200))
|
||||
proc.exit(7)
|
||||
`),
|
||||
parent: fakeParent(),
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('exit code 7')
|
||||
expect(result.agentsStarted).toBe(1)
|
||||
// A worker death is a stop reason like any other: workflow/end fires
|
||||
// with the error outcome — for a bus observer it is the only obituary.
|
||||
expect(runEnds).toEqual([{ stopReason: 'error', error: result.error, agentsStarted: 1 }])
|
||||
// Result already settled — this is the reap's promptness, not a
|
||||
// cold-start race; tight explicit bound (see the helper's doc comment).
|
||||
await waitFor(() => { expect(cancelled.length).toBe(1) }, 1000)
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('an uncaught exception inside the worker surfaces as an error result and reaps the in-flight child', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted(`
|
||||
agent('in flight when the worker dies')
|
||||
const proc = ${ESCAPE}
|
||||
const st = globalThis.constructor.constructor('return setTimeout')()
|
||||
await new Promise(resolve => st(resolve, 200))
|
||||
proc.nextTick(() => { throw new Error('worker blew up') })
|
||||
await new Promise(() => {})
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('worker blew up')
|
||||
// The reap wound the stray child down (cancel + a CLEAN dispose).
|
||||
// Result already settled — this is the reap's promptness, not a
|
||||
// cold-start race; tight explicit bound (see the helper's doc comment).
|
||||
await waitFor(() => {
|
||||
expect(provider.runs.length).toBe(1)
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
}, 1000)
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('a worker death pairs every stranded start: the synthesized cancelled agent-end precedes the error workflow/end', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const ends: { seq: number; outcome: string }[] = []
|
||||
const order: string[] = []
|
||||
ctx.on('workflow/agent-start', (_info, agent) => { order.push(`start:${agent.seq}`) })
|
||||
ctx.on('workflow/agent-end', (_info, agent) => {
|
||||
ends.push({ seq: agent.seq, outcome: agent.outcome })
|
||||
order.push(`end:${agent.seq}`)
|
||||
})
|
||||
ctx.on('workflow/end', () => { order.push('run-end') })
|
||||
const handle = ctx.workflows.start({
|
||||
// Same choreography as the force-settle pairing test, but the worker
|
||||
// DIES (the documented vm escape) instead of being terminated: the
|
||||
// exit path must close slow's pair from the ledger too. The escaped
|
||||
// setTimeout lets the already-posted messages flush before the kill.
|
||||
...scripted(`
|
||||
const p = agent('slow')
|
||||
await agent('fast')
|
||||
const proc = ${ESCAPE}
|
||||
const st = globalThis.constructor.constructor('return setTimeout')()
|
||||
await new Promise(resolve => st(resolve, 150))
|
||||
proc.exit(7)
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await waitFor(() => { expect(order.filter(entry => entry.startsWith('start:')).length).toBe(2) })
|
||||
const fast = provider.runs.find(run => (run.request.prompt[0] as { text?: string }).text === 'fast')!
|
||||
fast.settle(text('fast done'))
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('exit code 7')
|
||||
expect(ends).toEqual([
|
||||
{ seq: 2, outcome: 'completed' },
|
||||
{ seq: 1, outcome: 'cancelled' },
|
||||
])
|
||||
expect(order.indexOf('run-end')).toBe(order.length - 1)
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('a dispose ack racing the worker death is dropped, not crashed (post after exit)', async () => {
|
||||
// Slow child disposal: the ack resolves only AFTER the worker died, so
|
||||
// it has nowhere to go and must be dropped silently (the workerGone
|
||||
// guard in post()).
|
||||
const { ctx, parent, provider } = await setup({ disposeDelayMs: 300 })
|
||||
const handle = ctx.workflows.start({
|
||||
// The STRAY child settles instantly, so its wrapper starts the slow
|
||||
// host-side disposal concurrently while the script goes on to kill
|
||||
// its own worker — the ack then resolves into a dead thread.
|
||||
...scripted(`
|
||||
agent('stray, never awaited')
|
||||
const proc = ${ESCAPE}
|
||||
const st = globalThis.constructor.constructor('return setTimeout')()
|
||||
await new Promise(resolve => st(resolve, 150))
|
||||
proc.exit(5)
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('exit code 5')
|
||||
// Result already settled — this is the reap's promptness (bounded
|
||||
// above the mock's fixed 300ms dispose delay, not a cold-start race);
|
||||
// tight explicit bound (see the helper's doc comment).
|
||||
await waitFor(() => { expect(provider.runs[0]!.disposed).toBe(true) }, 1000)
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
|
||||
it('a worker death AFTER a cancel reports cancelled, not error', async () => {
|
||||
const { ctx, parent } = await setup({ config: { provider: 'stub', disposeGraceMs: 60_000 } })
|
||||
const handle = ctx.workflows.start({
|
||||
...scripted(`
|
||||
const proc = ${ESCAPE}
|
||||
const st = globalThis.constructor.constructor('return setTimeout')()
|
||||
log('armed')
|
||||
await new Promise(resolve => st(resolve, 400))
|
||||
proc.exit(3)
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
const logs: string[] = []
|
||||
ctx.on('workflow/log', (_info, message) => { logs.push(message) })
|
||||
await waitFor(() => { expect(logs).toContain('armed') })
|
||||
handle.cancel('stop it')
|
||||
// The grace is deliberately huge: only the worker's own death (exit 3,
|
||||
// unreachable by the cancel — the script ignores hooks) settles this.
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.error).toContain('stop it')
|
||||
await handle.dispose()
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
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({ ...scripted('return 1'), parent })
|
||||
const second = ctx.workflows.start({ ...scripted('return 2'), parent })
|
||||
expect(first.id).not.toBe(second.id)
|
||||
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), and default config runs (auto concurrency)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const fiber = await ctx.plugin(WorkerWorkflowEngine, {})
|
||||
expect(ctx.get('workflows')).toBeDefined()
|
||||
// A zero-agent run through the DEFAULT config exercises the auto
|
||||
// concurrency resolution (cores - 2, capped) in start().
|
||||
const result = await run(ctx, fakeParent(), scripted('return 6 * 7'))
|
||||
expect(result.value).toBe(42)
|
||||
await fiber.dispose()
|
||||
expect(ctx.get('workflows')).toBeUndefined()
|
||||
})
|
||||
|
||||
it('has the class-plugin export shape (default = the engine service class)', () => {
|
||||
expect(workerEngineModule.default).toBe(WorkerWorkflowEngine)
|
||||
const loader = Object.create(Loader.prototype) as Loader
|
||||
const unwrapped: unknown = loader.unwrapExports(workerEngineModule)
|
||||
expect(unwrapped).toBe(WorkerWorkflowEngine)
|
||||
})
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user