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

fix(workflow): harden terminal cleanup races

Browse Source 
Queue worker results before settlement cleanup, claim terminal and death boundaries before provider callbacks, and close late-message admission.

Make child cancellation and disposal reentrancy-safe across the workflow bridge and generic subagent wrapper, with adversarial regression coverage and RFC documentation.
This commit is contained in:
Tianyi Cui
2026-07-12 10:17:31 +08:00
parent c5b1a7941f
commit 9fc2260bb6
10 changed files with 803 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
packages/workflow/workflow-workerthread/README.md
View File

@@ -25,7 +25,7 @@ What the seam guarantees regardless, because benign scripts hit these constantly
Inside the worker, `runWorkerSession` builds the execution core (hooks, combinators, concurrency semaphore, caps, fatal-error discipline) over a **child port**. `agent()` sends `child-start`, and the host starts the child through the holder-bound `SubagentService` handle captured synchronously by `start()`, with parent attribution, the shared per-run abort signal, and `outputSchema`/`model` pass-through. This capture is part of the seam's holder-owned lifetime: unloading the engine removes `ctx.workflows` for new calls but does not invalidate an already returned run whose worker starts another child afterward.
The host observes `run.result` immediately but buffers its snapshotted wire projection until `run.started` fulfills. It then replies `child-started` with the child id before forwarding settlement, so `workflow/agent-start` always names a ready child and precedes its end. A readiness rejection replies `child-start-error`, emits no workflow agent pair, and makes the host dispose the attempt because the worker never received a handle; the worker classifies it as fatal `AGENT_START` unless cancellation already owns the run. If readiness fulfills, an infrastructure result rejection crosses as `child-failed`/`AGENT_RESULT` regardless of whether that rejection settled before or after readiness. Child disposal acknowledgements complete the RPC.
The host observes `run.result` immediately but buffers its snapshotted wire projection until `run.started` fulfills. Provider `start()` is arbitrary code and can synchronously reenter workflow cancellation before its returned run reaches the host registry, so the host registers the run, attaches both promise observers, and re-checks admission after `start()` returns and again at readiness. A closed boundary never admits or announces the run to the worker: while the exact run remains registered, the host invokes explicit cancel once and disposes it; `child-start-error` is sent only while worker-message admission remains open. If the run was already retired, the identity guard sends no cleanup through the deleted call ID. An ordinary readiness rejection sends `child-start-error` while possible and disposes the provider attempt without adding an explicit cancellation. Otherwise the host replies `child-started` with the child id before forwarding settlement, so `workflow/agent-start` always names a ready child and precedes its end. The worker classifies a start error as fatal `AGENT_START` unless cancellation already owns the run. If readiness fulfills, an infrastructure result rejection crosses as `child-failed`/`AGENT_RESULT` regardless of whether that rejection settled before or after readiness. Child disposal acknowledgements complete the RPC.
Observer narration (`phase`/`log`/`agent-start`/`agent-end`) crosses as messages and re-emits as the seam's `workflow/*` events. A **ready→go handshake** gates the body: a cancellation racing worker boot arrives before `go`, so a run cancelled before start never executes the body at all.
@@ -35,9 +35,15 @@ Values LEAVING the script (hook options/schemas, the script's return) are materi
## Cancellation, death, disposal
Per-run limits: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` posts the cancel to the worker (its hooks start throwing `CANCELLED`; the script dies at its next await) and cancels every host-side child NOW on **both seam channels** — the shared request signal aborts AND each registered child's explicit `cancel()` is called host-side, because the seam leaves a provider free to honor either channel and a worker wedged in a synchronous spin could not relay its own per-child cancel RPCs (those later land as idempotent no-ops). Each provider-owned cancel callback is exception-contained independently, so a broken child cannot prevent peer cancellation or workflow settlement. The caller's optional start-signal callback is retained by exact identity only while the run is live and removed at the first settlement or teardown, so a long-lived signal cannot retain completed `WorkerRun` instances. The grace then arms: a run still unsettled `disposeGraceMs` later force-settles `cancelled` and the worker is **terminated**. A cancellation that lands before the body runs (the ready→go handshake) reports `cancelled` without executing anything; a worker `result` racing an in-flight host cancellation reports `cancelled` too (first-wins settlement — the seam-visible result had not settled when cancellation was requested); post-cancel `phase`/`log` narration is suppressed host-side, while cancelled children still deliver their paired `agent-end`.
Cancellation is bounded and host-driven. Per-run limits are a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` first records its reason, then posts to the worker (its hooks start throwing `CANCELLED`; the script dies at its next await) and cancels every host-side child NOW on **both seam channels**: the shared request signal aborts and each registered child's explicit `cancel()` runs host-side. The seam leaves a provider free to honor either channel, and a worker wedged in a synchronous spin could not relay its own per-child cancel RPCs. A host-side per-call gate turns the worker's later explicit-cancel relay into a no-op, because the seam does not require `SubagentRun.cancel()` to be idempotent. Each explicit child `cancel()` callback is exception-contained independently, post-cancel `phase`/`log` narration is suppressed host-side, and cancelled children still deliver paired `agent-end` events. The caller's optional start-signal callback is retained by exact identity only while the run is live and removed at the first settlement or teardown.
A worker that dies unexpectedly (an OOM, a script reaching `process.exit` through the documented vm escape) settles the run `stopReason: 'error'` with the exit diagnostics — or `'cancelled'` when a cancel was in flight — and the host-side child registry is what winds every surviving child down. `dispose()` = cancel + immediate host-driven disposal of every registered child (a wedged worker can relay no dispose RPC, so child teardown overlaps the grace instead of starting after it; the worker's own dispose RPCs join the same per-child disposal) + bounded wait (result, then child-registry quiescence, capped by the grace) + unconditional `worker.terminate()`: the thread never outlives its run. Before an ordinary run settlement becomes observable, the host cancels every stray child on both channels too—even a fire-and-forget run still waiting on `started`, for which the worker has no handle yet—and `dispose()` then waits for their disposal (bounded by the grace) before returning. `agent-start`/`agent-end` pairing is host-guaranteed the same way: forwarded starts live in a ledger, worker-reported ends pair them on the graceful paths, and the termination paths (grace force-settle, worker death) synthesize the missing ends (outcome `cancelled`) before the run settles — a start still in flight across the force-settle can surface after `workflow/end`, immediately paired the same way.
Terminal arbitration is first-wins at explicit host-side claim points. A cancellation before ready→go reports `cancelled` without executing the body. For a later race, the worker queues Result before its settlement-reap `ChildCancel` messages; external `cancel()` records its reason before its fanout, while Result receipt snapshots any earlier cancellation and records the terminal outcome before settlement-cleanup fanout. Same-port FIFO and those claim points mean earlier caller/signal/dispose cancellation overrides a non-cancelled report, while an arrived report cannot be rewritten by a cleanup callback. Once Result has won, a losing reentrant `cancel()` has no state, message, child-fanout, or grace-timer effect. If no earlier terminal source settles the run, the grace callback claims `cancelled`, synthesizes missing lifecycle ends, settles the result, and terminates the worker after `disposeGraceMs`.
Worker death separates outcome ownership, message admission, and resource cleanup. An unexpected OOM, `error`, message failure, or premature exit claims `stopReason: 'error'` with diagnostics—or preserves an external cancellation already in flight—before reaping children or synthesizing observer events. Reentrant provider cancellation therefore cannot turn a death-first error into cancellation. The first death signal also closes worker-message admission because Node may deliver a queued `message` between `error` and `exit`; late protocol data cannot start a child, emit narration, or compete with the outcome. If Result or grace already claimed the outcome, death preserves it while still reaping promptly. The eventual `exit` then performs a final disposal-only sweep, joining any in-flight disposal without repeating explicit child cancellation. This separation lets grace settlement become observable before `worker.terminate()` reports exit without leaking the host-side registry.
Disposal is the holder's bounded resource guarantee: cancel, begin host-driven disposal of every registered child immediately, wait for result plus child-registry quiescence up to the same grace, and unconditionally terminate the worker. `handle.dispose()` claims its public promise before that traversal invokes cancellation or disposal callbacks. Independently, every `disposeChild` path claims the call ID's promise before invoking the wrapped child disposer. Public-first reentry therefore returns the existing holder promise; worker-first reentry may begin holder disposal, whose child traversal joins the already-claimed call ID promise. Neither order can start a second provider disposal. A wedged worker can relay no dispose RPC, so host-driven teardown overlaps the grace; any later worker RPC joins the same per-child disposal. Before ordinary settlement becomes observable, the host also cancels every stray on both channels, including a fire-and-forget run still waiting on readiness. That work is settlement-only cleanup after the terminal claim, so provider reentry cannot rewrite the chosen result; `dispose()` then waits for its completion within the bound.
Lifecycle pairing is host-guaranteed independently of outcome arbitration. Forwarded starts live in a ledger and worker-reported ends pair them on graceful paths. When death or grace is the terminal source, the host synthesizes missing ends with outcome `cancelled` before `workflow/end`. If Result settled first, later death cleanup may synthesize a survivor's end afterward; a start already crossing force-settlement may likewise surface after `workflow/end`. The same ledger still pairs every forwarded start exactly once.
**Engine-specific limitations**: worker startup is paid per run; on a termination path `agentsStarted` reports the HOST-observed count (accepted `child-start`s — calls still queued worker-side for a concurrency slot are unknowable then); and a returned promise or thenable resolves per JavaScript semantics BEFORE materialization — that is what makes an un-awaited `return agent('x')` work — with the value-boundary guard applying to the resolution.

264
packages/workflow/workflow-workerthread/src/host.ts
View File

@@ -7,19 +7,31 @@
* run.
*
* The run's `result` promise settles exactly once, from whichever of these
* lands first: the worker's `result` message (a host-side cancellation in
* flight overrides a non-cancelled report — the seam-visible result had not
* settled when cancellation was requested), an unexpected worker death
* (`error`/`messageerror`/premature `exit` → `stopReason: 'error'`, or
* `'cancelled'` when a cancel was in flight), or the post-cancel grace timer
* (a script that never settles is force-settled `cancelled` and its worker
* terminated — the real kill an in-process engine could not perform).
* lands first: receipt of the worker's `result` message, an unexpected worker
* death (`error`/`messageerror`/premature `exit` → `stopReason: 'error'`, or
* `'cancelled'` when a cancel was in flight), or the post-cancel grace timer (a
* script that never settles is force-settled `cancelled` and its worker
* terminated — the real kill an in-process engine could not perform). At
* `result` receipt the host snapshots whether caller/signal/dispose
* cancellation is already in flight: an earlier cancellation overrides a
* non-cancelled report; otherwise the report wins before settlement-only child
* cleanup invokes arbitrary provider callbacks. Worker death uses the same
* boundary: it claims `error` (or a previously requested `cancelled`) before
* reaping children, so cleanup callbacks cannot rewrite the outcome. That
* first signal also closes inbound message admission: Node may emit `error`,
* then deliver queued messages, then emit `exit`, but those late messages may
* neither create work nor narrate after settlement. If Result or grace already
* owns the outcome, death preserves it while still cleaning resources; the
* eventual exit performs a final disposal-only sweep without repeating child
* cancellation.
*
* Children live in a host-side registry (callId → run) as soon as the provider
* accepts them, so cancellation reaches even a pre-publication attempt. Both
* explicit run cancellation and the shared request signal are driven when the
* workflow is cancelled OR normally settles, so a fire-and-forget child cannot
* survive merely by honoring only one channel. The host observes `result`
* survive merely by honoring only one channel. A per-call gate invokes each
* explicit provider `cancel()` at most once even though host fanout and the
* worker's later relay can both request it. The host observes `result`
* immediately but acknowledges the child to the worker only after `started`
* fulfills; readiness failure is a start error and the host disposes the
* attempt because the worker never received a handle. The
@@ -31,10 +43,12 @@
* paths share ONE disposal per child (memoized by callId; the seam's
* dispose() is idempotent anyway, the memo keeps the bookkeeping and the
* containment warn single). Lifecycle pairing is host-guaranteed the same
* way: every forwarded `agent-start` lives in a ledger, and a start the
* dead or terminated worker never paired is closed by a synthesized
* `agent-end` (outcome `'cancelled'`) before the run settles. On a
* termination path `agentsStarted` reports the
* way: every forwarded `agent-start` lives in a ledger, and a start the dead
* or terminated worker never paired is closed exactly once by a synthesized
* `agent-end` (outcome `'cancelled'`). When death or grace is the terminal
* source, already-known pairs close before the run settles; cleanup after an
* earlier Result can close a survivor afterward. On a termination path
* `agentsStarted` reports the
* HOST-observed count (accepted `child-start` messages) — `agent()` calls
* still queued worker-side for a concurrency slot are unknowable then; the
* worker's own count rides the result message on every graceful path.
@@ -132,6 +146,10 @@ export class WorkerRun implements WorkflowRun {
readonly result: Promise<WorkflowResult>
private settleResolve!: (result: WorkflowResult) => void
private settled = false
/** A Result/death/grace outcome atomically won before teardown callbacks. */
private terminalClaimed = false
/** The first death signal closes worker-message admission and owns failure-time cleanup. */
private workerDeathObserved = false
private cancelReason: string | undefined
private graceTimer: NodeJS.Timeout | undefined
private readonly worker: Worker
@@ -143,6 +161,8 @@ export class WorkerRun implements WorkflowRun {
private readonly children = new Map<number, SubagentRun>()
/** In-flight child disposals by callId — the memo that gives every path (worker RPC, dispose(), reap) ONE shared disposal per child. */
private readonly childDisposals = new Map<number, Promise<void>>()
/** callIds whose explicit provider cancel callback has already been invoked. */
private readonly childCancellations = new Set<number>()
/** Started-but-not-ended agents by seq — the pairing ledger the HOST guarantees (see {@link endAgent}). */
private readonly liveAgents = new Map<number, WorkflowAgentInfo>()
private readonly quiescenceWaiters: (() => void)[] = []
@@ -172,12 +192,12 @@ export class WorkerRun implements WorkflowRun {
const { entry, options } = resolveWorkerSpawn(init)
this.worker = new Worker(entry, options)
this.worker.on('message', (message: WorkerToHostMessage) => { this.onMessage(message) })
this.worker.on('error', (error) => { this.onWorkerDeath(`workflow worker failed: ${renderThrown(error)}`) })
this.worker.on('error', (error) => { this.onWorkerDeath(`workflow worker failed: ${renderThrown(error)}`, false) })
/* v8 ignore next -- messageerror: not constructible from the engine's own protocol (every payload is JSON data) */
this.worker.on('messageerror', (error) => { this.onWorkerDeath(`workflow worker message failed to deserialize: ${renderThrown(error)}`) })
this.worker.on('messageerror', (error) => { this.onWorkerDeath(`workflow worker message failed to deserialize: ${renderThrown(error)}`, false) })
this.worker.on('exit', (code) => {
this.workerGone = true
this.onWorkerDeath(`workflow worker exited before the run settled (exit code ${code})`)
this.onWorkerDeath(`workflow worker exited before the run settled (exit code ${code})`, true)
})
if (signal?.aborted) {
this.cancel('workflow start signal already aborted')
@@ -205,18 +225,24 @@ export class WorkerRun implements WorkflowRun {
* @param reason - human-readable cause (default `'workflow cancelled'`).
*/
cancel(reason?: string): void {
// A settled run has nothing left to cancel: without this guard the
// A settled run has nothing left to cancel, and a terminal source claimed
// before its cleanup callbacks must exclude cancellation reentered by one
// of those callbacks. Without the settled guard the
// ordinary consumer path (await result, then dispose -> cancel) would arm
// a grace timer nothing ever clears, pinning the run and its Worker
// closure until the grace expires - a bounded leak per completed run.
if (this.settled || this.cancelReason !== undefined) return
if (this.settled || this.terminalClaimed || this.cancelReason !== undefined) return
this.cancelReason = reason ?? 'workflow cancelled'
this.post(HostToWorkerType.Cancel, { reason: this.cancelReason })
// The explicit channel is driven host-side, not left to the worker: a
// provider honoring only run.cancel() must not wait on a wedged worker's
// ChildCancel relay (those later RPCs land as idempotent no-ops).
// ChildCancel relay (the per-call cancellation gate makes those later
// RPCs no-ops without imposing idempotence on the provider).
this.cancelChildren(this.cancelReason)
this.graceTimer = setTimeout(() => {
// Cancellation already owns the race through cancelReason; close the
// terminal boundary explicitly before observer teardown callbacks.
this.terminalClaimed = true
// The worker may no longer speak (it is about to be terminated): pair
// every stranded start before the run settles, so ends precede
// workflow/end.
@@ -244,7 +270,13 @@ export class WorkerRun implements WorkflowRun {
* @returns resolves when the run's resources are released or abandoned.
*/
dispose(): Promise<void> {
this.disposed ??= (async () => {
if (this.disposed !== undefined) return this.disposed
// Claim the public transaction BEFORE its body invokes child/provider
// disposal. A raw provider callback can reenter handle.dispose(); it must
// join this promise rather than start a second traversal.
const claimed = Promise.withResolvers<undefined>()
this.disposed = claimed.promise
void (async () => {
this.detachInputSignal()
this.cancel('workflow disposed')
for (const [callId, run] of [...this.children]) void this.disposeChild(callId, run)
@@ -257,13 +289,17 @@ export class WorkerRun implements WorkflowRun {
])
await this.worker.terminate()
this.reapChildren('workflow disposed')
})()
})().then(
() => { claimed.resolve(undefined) },
/* v8 ignore next -- result/quiescence never reject and Worker.terminate is the only external promise */
(error: unknown) => { claimed.reject(error) },
)
return this.disposed
}
/** Post one message to the worker (payload looked up from the tag's map entry), tolerating a thread that is already gone. */
private post<T extends HostToWorkerType>(type: T, payload: HostToWorkerPayloads[T]): void {
if (this.workerGone) return
if (this.workerGone || this.workerDeathObserved) return
try {
this.worker.postMessage({ type, ...payload })
} catch (error: unknown) {
@@ -276,6 +312,11 @@ export class WorkerRun implements WorkflowRun {
}
private onMessage(message: WorkerToHostMessage): void {
// Node may emit `error`, then deliver an already-queued `message`, then
// emit `exit`. The first death signal is the host's logical delivery
// barrier: nothing arriving afterward may create a child, narrate after
// workflow/end, or compete with the chosen outcome.
if (this.workerDeathObserved) return
switch (message.type) {
case WorkerToHostType.Ready:
this.post(HostToWorkerType.Go, {})
@@ -308,7 +349,7 @@ export class WorkerRun implements WorkflowRun {
case WorkerToHostType.ChildCancel:
{
const run = this.children.get(message.callId)
if (run !== undefined) this.cancelChild(run, message.reason)
if (run !== undefined) this.cancelChild(message.callId, run, message.reason)
}
break
case WorkerToHostType.ChildDispose:
@@ -323,12 +364,27 @@ export class WorkerRun implements WorkflowRun {
}
}
private onChildStart(callId: number, request: ChildStartRequest): void {
/** Why a child may no longer cross the provider readiness boundary. */
private childAdmissionFailure(): { reason: string; rendered: string } | undefined {
if (this.cancelReason !== undefined) {
// The worker's start raced our cancel: refuse — a child must never
// start on an already-aborted signal (a provider subscribing only to
// future abort events would never observe it).
this.post(HostToWorkerType.ChildStartError, { callId, rendered: `workflow run cancelled: ${this.cancelReason}` })
return { reason: this.cancelReason, rendered: `workflow run cancelled: ${this.cancelReason}` }
}
if (this.workerDeathObserved) {
return { reason: 'workflow worker gone', rendered: 'workflow worker is no longer available' }
}
if (this.terminalClaimed) {
return { reason: 'workflow settled', rendered: 'workflow run already settled' }
}
return undefined
}
private onChildStart(callId: number, request: ChildStartRequest): void {
const initialFailure = this.childAdmissionFailure()
if (initialFailure !== undefined) {
// Refuse after a terminal boundary: a child must never start on an
// already-aborted signal (a provider subscribing only to future abort
// events would never observe it).
this.post(HostToWorkerType.ChildStartError, { callId, rendered: initialFailure.rendered })
return
}
this.hostStarted += 1
@@ -382,22 +438,54 @@ export class WorkerRun implements WorkflowRun {
},
)
// The provider owns the publication boundary. Only acknowledge the child
// after it is real, then flush any result that settled unusually early. A
// readiness rejection is a START failure, not AGENT_RESULT: the worker
// never receives a handle, so the host must also dispose the registered
// attempt. A concurrent host disposal may already have removed it; the
// identity guard preserves the one-disposal memo in that race.
// The provider owns the publication boundary. Observe both promises before
// invoking cancellation/disposal below: provider.start() itself is
// arbitrary code and may have reentered handle.cancel() before the returned
// run reached our registry. Exactly one branch answers this ChildStart.
let startReplySent = false
const refusePublication = (failure: { reason: string; rendered: string }): void => {
startReplySent = true
this.post(HostToWorkerType.ChildStartError, { callId, rendered: failure.rendered })
// A prior dispose/death can finish and remove this run while readiness
// is still pending. In that case teardown already owned cancellation and
// disposal; touching the retired callId would repeat cancel and orphan a
// fresh gate entry after finishChild deleted it.
if (this.children.get(callId) !== run) return
this.cancelChild(callId, run, failure.reason)
void this.disposeChild(callId, run)
}
// Only acknowledge the child after it is real, then flush any result that
// settled unusually early. Re-check admission at that exact boundary: a
// cancellation while readiness was pending is a refusal, not a late
// publication into a terminal workflow. A readiness rejection is a START
// failure, not AGENT_RESULT; the worker never receives a handle, so the
// host disposes the registered attempt. Identity guards preserve the one
// disposal memo against concurrent host teardown.
void run.started.then(
() => {
if (startReplySent) return
const failure = this.childAdmissionFailure()
if (failure !== undefined) {
refusePublication(failure)
return
}
startReplySent = true
this.post(HostToWorkerType.ChildStarted, { callId, childId })
void forwardResult.then((forward) => { forward() })
},
(error: unknown) => {
if (startReplySent) return
startReplySent = true
this.post(HostToWorkerType.ChildStartError, { callId, rendered: renderThrown(error) })
if (this.children.get(callId) === run) void this.disposeChild(callId, run)
},
)
// Close the synchronous hole around provider.start(): cancel()/dispose()
// can run before the returned run is visible to their children loop.
const reentrantFailure = this.childAdmissionFailure()
if (reentrantFailure !== undefined) refusePublication(reentrantFailure)
}
private onChildDispose(callId: number): void {
@@ -427,17 +515,26 @@ export class WorkerRun implements WorkflowRun {
private disposeChild(callId: number, run: SubagentRun): Promise<void> {
let disposal = this.childDisposals.get(callId)
if (disposal === undefined) {
// Claim before run.dispose() invokes provider code. Reentrant holder
// disposal then joins this exact child transaction instead of entering
// the provider wrapper twice before either memo is installed.
const claimed = Promise.withResolvers<undefined>()
disposal = claimed.promise
this.childDisposals.set(callId, disposal)
// The seam promises a Promise, but invoke inside an async boundary so a
// contract-violating synchronous throw is contained exactly like a
// rejected disposal and cannot break host quiescence.
disposal = (async () => { await run.dispose() })().then(
() => { this.finishChild(callId) },
void (async () => { await run.dispose() })().then(
() => {
this.finishChild(callId)
claimed.resolve(undefined)
},
(error: unknown) => {
this.ctx.logger.warn(`workflow-workerthread: child dispose failed: ${renderThrown(error)}`)
this.finishChild(callId)
claimed.resolve(undefined)
},
)
this.childDisposals.set(callId, disposal)
}
return disposal
}
@@ -446,6 +543,7 @@ export class WorkerRun implements WorkflowRun {
private finishChild(callId: number): void {
this.children.delete(callId)
this.childDisposals.delete(callId)
this.childCancellations.delete(callId)
if (this.children.size === 0) {
for (const waiter of this.quiescenceWaiters.splice(0)) waiter()
}
@@ -469,11 +567,16 @@ export class WorkerRun implements WorkflowRun {
/** Drive both cancellation channels for every child already accepted by the host. */
private cancelChildren(reason: string): void {
this.controller.abort(reason)
for (const run of this.children.values()) this.cancelChild(run, reason)
for (const [callId, run] of this.children) this.cancelChild(callId, run, reason)
}
/** Contain one provider-owned cancel callback so every peer still receives cancellation. */
private cancelChild(run: SubagentRun, reason?: string): void {
/** Invoke one provider-owned cancel callback at most once and contain its exception. */
private cancelChild(callId: number, run: SubagentRun, reason?: string): void {
// Host fanout and the worker's FIFO-later ChildCancel relay are two paths
// to the same provider callback. The seam does not require cancel() to be
// idempotent, so claim the callId before invoking arbitrary provider code.
if (this.childCancellations.has(callId)) return
this.childCancellations.add(callId)
try {
run.cancel(reason)
} catch (error: unknown) {
@@ -482,12 +585,30 @@ export class WorkerRun implements WorkflowRun {
}
private onResult(result: WorkflowResult): void {
// The owned worker session sends one Result. Keep a late duplicate or a
// Result queued behind another terminal source completely side-effect-free.
if (this.terminalClaimed) return
// First-wins is decided when the Result message reaches the host. If no
// external cancellation was already in flight, this result won. Reaping a
// stray child below may synchronously reenter cancel() through provider
// callbacks, but that internal post-result cleanup must not retroactively
// rewrite the worker result that arrived first.
const cancellationWasRequested = this.cancelReason !== undefined
// Claim before either settlement-cleanup cancellation channel invokes
// provider code. A provider callback can reenter cancel() synchronously or
// from a queued microtask; once Result won, that losing cancellation must
// have no state, message, child-fanout, or grace-timer side effects.
this.terminalClaimed = true
// The worker cancels handles it already received, but a fire-and-forget
// child may still be waiting on readiness and therefore have no worker
// handle. Drive BOTH provider-permitted channels from the host before the
// workflow becomes externally settled.
if (this.cancelReason === undefined) this.cancelChildren('workflow settled')
if (this.cancelReason !== undefined && result.stopReason !== 'cancelled') {
if (!cancellationWasRequested) {
this.cancelChildren('workflow settled')
this.settleResult(result)
return
}
if (result.stopReason !== 'cancelled') {
// The script settled while our cancel was crossing the thread boundary
// — the seam-visible result had NOT settled when cancellation was
// requested, so report cancelled (the vm drive()'s post-settle check,
@@ -498,21 +619,39 @@ export class WorkerRun implements WorkflowRun {
this.settleResult(result)
}
/** An unexpected worker death (or the expected exit after termination). */
private onWorkerDeath(message: string): void {
// Whatever the worker left behind must not leak — abort + dispose it all.
if (this.children.size > 0) this.reapChildren('workflow worker gone')
// The thread is gone: no more worker-authored agent-ends can arrive —
// pair every stranded start (a start that crossed between the grace
// force-settle and this exit included) before the run settles.
this.endStrandedAgents()
// settleResult no-ops on an already-settled run (the expected exit after
// a dispose's terminate lands here too).
if (this.cancelReason !== undefined) {
this.settleResult(this.cancelledResult(this.hostStarted))
return
/** Process an error/messageerror/exit signal; `exit` also performs the final disposal sweep. */
private onWorkerDeath(message: string, isExit: boolean): void {
if (!this.workerDeathObserved) {
// Close message admission BEFORE cleanup callbacks: Node can deliver a
// message queued before the crash after its `error` event. Treating the
// first death signal as a logical barrier prevents that late message
// from creating work or narrating after workflow/end.
this.workerDeathObserved = true
const outcomeWasClaimed = this.terminalClaimed
const cancellationWasRequested = this.cancelReason !== undefined
// When death is itself the terminal source, claim BEFORE child reap or
// synthesized observer callbacks. Either can reenter cancel(); a death
// that arrived first remains an error, while a cancellation already
// accepted before death remains cancelled. If Result/grace already won,
// preserve it while still performing prompt failure-time cleanup.
if (!outcomeWasClaimed) this.terminalClaimed = true
if (this.children.size > 0) this.reapChildren('workflow worker gone')
this.endStrandedAgents()
if (!outcomeWasClaimed) {
if (cancellationWasRequested) {
this.settleResult(this.cancelledResult(this.hostStarted))
} else {
this.settleResult({ value: null, stopReason: 'error', error: message, agentsStarted: this.hostStarted })
}
}
}
this.settleResult({ value: null, stopReason: 'error', error: message, agentsStarted: this.hostStarted })
if (!isExit) return
// `error` is not Node's physical delivery barrier: a queued message may
// precede `exit`. Admission is already closed, so this final sweep only
// joins/starts disposal for registry survivors; it deliberately does not
// repeat explicit provider cancellation.
for (const [callId, run] of [...this.children]) void this.disposeChild(callId, run)
this.endStrandedAgents()
}
/**
@@ -531,11 +670,14 @@ export class WorkerRun implements WorkflowRun {
/**
* Synthesize the missing `agent-end` for every started-but-unpaired agent,
* outcome `'cancelled'`: the reap cancels every child, and a real
* settlement racing the force-settle loses to the cancellation — the same
* first-wins override {@link onResult} applies to the run's own result.
* settlement racing the force-settle loses to that already-started external
* cancellation. The atomic terminal boundaries in {@link onResult} and
* {@link onWorkerDeath} deliberately exclude teardown callbacks as contenders.
* Called where the worker can no longer speak (the grace force-settle,
* worker death), BEFORE settleResult, so the paired ends reach observers
* before `workflow/end`.
* worker death, physical exit). When grace/death is the terminal source it
* runs before settleResult, so already-known pairs precede `workflow/end`;
* after an earlier Result, exit cleanup may close a survivor afterward.
* The ledger preserves exactly-once pairing in both orders.
*/
private endStrandedAgents(): void {
for (const info of [...this.liveAgents.values()]) {
@@ -563,7 +705,11 @@ export class WorkerRun implements WorkflowRun {
/** First settle wins; disarms the grace timer and releases the caller signal. */
private settleResult(result: WorkflowResult): void {
// Every current terminal source claims ownership before calling here; keep
// the fallback local so a future caller cannot resolve twice.
/* v8 ignore next -- defensive fallback outside the claimed state machine */
if (this.settled) return
this.terminalClaimed = true
this.settled = true
this.detachInputSignal()
clearTimeout(this.graceTimer)

28
packages/workflow/workflow-workerthread/src/runtime.ts
View File

@@ -83,7 +83,9 @@ function defaultLabel(prompt: string): string {
/**
* One live script execution inside the worker. Constructed per run by the
* session; `drive()` is called exactly once and NEVER rejects — every failure
* becomes a {@link WorkflowResult} with a non-`completed` stop reason.
* becomes a {@link WorkflowResult} with a non-`completed` stop reason. After
* the session publishes that result it calls {@link reapAfterResult} exactly
* once to cancel any dropped child work without racing terminal publication.
*/
export class WorkflowExecution {
/** 1-based count of `agent()` calls started (the `agentsStarted` result field). */
@@ -171,7 +173,8 @@ export class WorkflowExecution {
* worker. Idempotent; the first reason wins.
* @param reason - human-readable cause, carried on the CANCELLED error and
* into child cancel RPCs. Required: every caller (the session's cancel
* message, drive()'s settle-reap) has a concrete reason.
* message and its post-result {@link reapAfterResult} call) has a concrete
* reason.
*/
cancel(reason: string): void {
if (this.cancelReason !== undefined) return
@@ -185,8 +188,9 @@ export class WorkflowExecution {
* Run the script to settlement. Resolves — never rejects — with the run's
* {@link WorkflowResult}: the materialized return value on `completed`, the
* failure message on `error`, and `cancelled` when the script died of
* cancellation. After settlement, any stray children a script fired without
* awaiting are cancelled (their `agent()` wrappers dispose them via RPC).
* cancellation. This method only chooses the result; the session must publish
* it and then call {@link reapAfterResult}, so the terminal message precedes
* settlement-only child cancellation on the worker-to-host FIFO channel.
* @returns the settled outcome — this promise NEVER rejects (the seam's
* `result`-never-rejects contract); every failure maps to a variant.
*/
@@ -214,15 +218,19 @@ export class WorkflowExecution {
// cannot throw — drive() resolving is the `result` never-rejects seam
// contract.
return { value: null, stopReason: 'error', error: renderThrown(error), agentsStarted: this.started }
} finally {
// Reap strays: a script that fired agent() calls without awaiting them
// leaves live children behind after settlement — cancel them all. (The
// per-call wrappers dispose each child; the contain() consumer keeps
// their rejections from going unhandled.)
if (this.cancelReason === undefined) this.cancel('workflow settled')
}
}
/**
* Reap strays only after the caller publishes the chosen terminal result.
* Aborting the controller synchronously sends child-cancel RPCs, so calling
* this before publication would let a provider callback reenter host
* cancellation and misclassify a result the script had already chosen.
*/
reapAfterResult(): void {
if (this.cancelReason === undefined) this.cancel('workflow settled')
}
/**
* Attach a no-op rejection consumer WITHOUT changing what the caller
* receives: if the script drops the promise (no await), cancellation cannot

14
packages/workflow/workflow-workerthread/src/session.ts
View File

@@ -14,6 +14,11 @@
* A `cancel` arriving instead of `go` still releases the gate: `drive()`
* sees the cancelled state and settles without running the body.
*
* Terminal ordering is Result first, settlement cleanup second. The session
* queues the Result message before asking the execution to reap stray children;
* MessagePort FIFO therefore lets the host atomically claim the result before a
* cleanup ChildCancel can invoke arbitrary provider code.
*
* @module @deepseek-ai/dsh-workflow-workerthread/session
*/
@@ -208,5 +213,12 @@ export async function runWorkerSession(port: MessagePort, init: WorkerInit): Pro
post(WorkerToHostType.Ready, {})
await gate.promise
const result = await execution.drive()
post(WorkerToHostType.Result, { result })
try {
// This post is the worker's terminal claim. Queue it BEFORE aborting stray
// children: MessagePort FIFO then guarantees the host claims Result before
// any settlement-only ChildCancel can invoke arbitrary provider callbacks.
post(WorkerToHostType.Result, { result })
} finally {
execution.reapAfterResult()
}
}

30
packages/workflow/workflow-workerthread/tests/session.spec.ts
View File

@@ -281,6 +281,36 @@ describe('runWorkerSession over an in-process MessageChannel', () => {
}
})
it('queues Result before settlement-only cancellation of a ready stray', async () => {
const host = fakeHost({ manual: true })
const session = runWorkerSession(host.port, init(`
agent('ready stray')
return await agent('gate')
`))
await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart)).toHaveLength(2) })
const starts = host.ofType(WorkerToHostType.ChildStart)
const stray = starts.find(message => message.request.prompt === 'ready stray')!
const gate = starts.find(message => message.request.prompt === 'gate')!
host.send({ type: HostToWorkerType.ChildStarted, callId: stray.callId, childId: 'stray-child' })
host.send({ type: HostToWorkerType.ChildStarted, callId: gate.callId, childId: 'gate-child' })
host.send({ type: HostToWorkerType.ChildSettled, callId: gate.callId, result: text('gate completed') })
const result = await host.result()
await session
await vi.waitFor(() => {
expect(host.ofType(WorkerToHostType.ChildCancel).map(message => message.callId)).toContain(stray.callId)
})
expect(result).toMatchObject({ value: 'gate completed', stopReason: 'completed', agentsStarted: 2 })
const resultIndex = host.messages.findIndex(message => message.type === WorkerToHostType.Result)
const strayCancelIndex = host.messages.findIndex(message =>
message.type === WorkerToHostType.ChildCancel && message.callId === stray.callId)
expect(resultIndex).toBeGreaterThanOrEqual(0)
expect(strayCancelIndex).toBeGreaterThan(resultIndex)
host.send({ type: HostToWorkerType.ChildSettled, callId: stray.callId, result: { output: [], stopReason: 'aborted' } })
host.close()
})
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 ((('))

423
packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts
View File

@@ -1,5 +1,6 @@
import { describe, expect, it, vi } from 'vitest'
import { fileURLToPath } from 'node:url'
import type { Worker } from 'node:worker_threads'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { AgentId } from '@deepseek-ai/dsh-agent'
@@ -8,7 +9,7 @@ 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, { HostToWorkerType, type Config } from '../src/index.ts'
import WorkerWorkflowEngine, { HostToWorkerType, WorkerToHostType, type Config } from '../src/index.ts'
/** A minimal parent stand-in: the engine only threads it through to the provider. */
function fakeParent(): Agent {
@@ -74,6 +75,8 @@ class StubProvider implements SubagentProvider {
private readonly reply?: (request: SubagentStartRequest, index: number) => SubagentResult,
private readonly disposeDelayMs = 0,
private readonly deferStart = false,
private readonly onCancel?: (reason: string | undefined, index: number) => void,
private readonly onSignalAbort?: (reason: unknown, index: number) => void,
) {}
start(request: SubagentStartRequest): SubagentRun {
@@ -91,7 +94,10 @@ class StubProvider implements SubagentProvider {
}
this.runs.push(controlled)
const index = this.runs.length - 1
request.signal?.addEventListener('abort', () => { terminal.resolve({ output: [], stopReason: 'aborted' }) }, { once: true })
request.signal?.addEventListener('abort', () => {
this.onSignalAbort?.(request.signal?.reason, index)
terminal.resolve({ output: [], stopReason: 'aborted' })
}, { once: true })
if (!this.deferStart) readiness.resolve(undefined)
if (this.reply) {
const reply = this.reply
@@ -103,6 +109,7 @@ class StubProvider implements SubagentProvider {
result: terminal.promise,
cancel: (reason?: string) => {
controlled.cancelled = reason ?? 'cancelled'
this.onCancel?.(reason, index)
terminal.resolve({ output: [], stopReason: 'aborted' })
},
dispose: () => {
@@ -133,6 +140,8 @@ interface SetupOptions {
manual?: boolean
disposeDelayMs?: number
deferStart?: boolean
onChildCancel?: (reason: string | undefined, index: number) => void
onChildSignalAbort?: (reason: unknown, index: number) => void
}
async function setup(options?: SetupOptions) {
@@ -143,6 +152,8 @@ async function setup(options?: SetupOptions) {
options?.manual ? undefined : options?.reply ?? (() => text('stub reply')),
options?.disposeDelayMs ?? 0,
options?.deferStart ?? false,
options?.onChildCancel,
options?.onChildSignalAbort,
)
ctx.subagents.registerProvider(provider)
// A fixed concurrency ceiling: the auto-resolved default is machine-derived
@@ -821,6 +832,153 @@ describe('dsh-workflow-workerthread', () => {
expect(provider.runs[0]!.disposeCalls).toBe(1)
})
it('post-result child cleanup cannot reentrantly rewrite a completed workflow as cancelled', async () => {
let cancelCallbacks = 0
let signalCallbacks = 0
const { ctx, parent, provider } = await setup({
manual: true,
deferStart: true,
onChildCancel: () => {
cancelCallbacks += 1
// The first callback is host cleanup for the already-arrived Result.
// Reentering cancel() here is later than that message and must not
// retroactively win the result race. Its nested child cancel is
// intentionally ignored to keep the adversarial callback finite.
if (cancelCallbacks === 1) handle.cancel('reentrant child cleanup')
},
onChildSignalAbort: () => {
signalCallbacks += 1
handle.cancel('reentrant signal cleanup')
},
})
const handle = ctx.workflows.start({
...scripted(`
agent('readiness-pending stray')
return 'completed first'
`),
parent,
})
const result = await handle.result
expect(result).toMatchObject({ value: 'completed first', stopReason: 'completed', agentsStarted: 1 })
expect(signalCallbacks).toBe(1)
expect(cancelCallbacks).toBe(1)
// Readiness crossing after Result is a terminal-admission refusal: no
// ChildStarted/lifecycle publication, and host-owned disposal begins.
provider.runs[0]!.publish()
await waitFor(() => { expect(provider.runs[0]!.disposed).toBe(true) }, 1000)
expect(cancelCallbacks).toBe(1)
await handle.dispose()
await ctx.fiber.dispose()
})
it('late readiness after completed disposal cannot cancel or dispose the retired child twice', async () => {
let explicitCancels = 0
const lifecycle: string[] = []
const { ctx, parent, provider } = await setup({
manual: true,
deferStart: true,
onChildCancel: () => { explicitCancels += 1 },
})
ctx.on('workflow/agent-start', () => { lifecycle.push('start') })
ctx.on('workflow/agent-end', () => { lifecycle.push('end') })
const handle = ctx.workflows.start({
...scripted("agent('retired readiness')\nreturn 'done'"),
parent,
})
await expect(handle.result).resolves.toMatchObject({ stopReason: 'completed' })
expect(explicitCancels).toBe(1)
await handle.dispose()
expect(provider.runs[0]!.disposed).toBe(true)
expect(provider.runs[0]!.disposeCalls).toBe(1)
// The Promise may still fulfill after its run left every host ledger.
// Refusal replies once but must not recreate the deleted cancel gate.
provider.runs[0]!.publish()
await Promise.resolve()
await Promise.resolve()
expect(explicitCancels).toBe(1)
expect(provider.runs[0]!.disposeCalls).toBe(1)
expect(lifecycle).toEqual([])
await ctx.fiber.dispose()
})
it.each([
['synchronous', (cancel: () => void) => { cancel() }],
['microtask', (cancel: () => void) => { queueMicrotask(cancel) }],
])('a ready stray %s cleanup callback cannot beat the earlier worker result claim', async (_mode, reenter) => {
let reentered = false
const explicitCancels = new Map<number, number>()
const { ctx, parent, provider } = await setup({
manual: true,
onChildCancel: (_reason, index) => {
explicitCancels.set(index, (explicitCancels.get(index) ?? 0) + 1)
if (index !== 0 || reentered) return
reentered = true
reenter(() => { handle.cancel('reentered from child cleanup') })
},
})
const handle = ctx.workflows.start({
...scripted(`
agent('ready stray')
return await agent('gate')
`),
parent,
})
const cancelChildSpy = vi.spyOn(handle as unknown as {
cancelChild(callId: number, run: SubagentRun, reason?: string): void
}, 'cancelChild')
await waitFor(() => { expect(provider.runs).toHaveLength(2) })
provider.runs[1]!.settle(text('gate completed'))
const result = await handle.result
await Promise.resolve()
expect(result).toMatchObject({ value: 'gate completed', stopReason: 'completed', agentsStarted: 2 })
expect(reentered).toBe(true)
// The host claim and worker's FIFO-later ChildCancel both reach the
// routing gate, but the provider callback is not an idempotent seam:
// invoke it exactly once for this callId.
await waitFor(() => {
expect(cancelChildSpy.mock.calls.filter(([callId]) => callId === 1)).toHaveLength(2)
}, 1000)
expect(explicitCancels.get(0)).toBe(1)
cancelChildSpy.mockRestore()
await handle.dispose()
await ctx.fiber.dispose()
})
it('a duplicate Result after the terminal claim cannot repeat cleanup or rewrite the outcome', async () => {
let explicitCancels = 0
const { ctx, parent, provider } = await setup({
manual: true,
onChildCancel: (_reason, index) => { if (index === 0) explicitCancels += 1 },
})
const handle = ctx.workflows.start({
...scripted("agent('stray')\nawait new Promise(() => {})"),
parent,
})
await waitFor(() => { expect(provider.runs).toHaveLength(1) })
const worker = (handle as unknown as { worker: Worker }).worker
worker.emit('message', {
type: WorkerToHostType.Result,
result: { value: 'first', stopReason: 'completed', agentsStarted: 1 },
})
worker.emit('message', {
type: WorkerToHostType.Result,
result: { value: 'late', stopReason: 'completed', agentsStarted: 1 },
})
await expect(handle.result).resolves.toMatchObject({ value: 'first', stopReason: 'completed' })
expect(explicitCancels).toBe(1)
await handle.dispose()
expect(explicitCancels).toBe(1)
await ctx.fiber.dispose()
})
it('contains a throwing child cancel and still settles after cancelling peer strays', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
@@ -918,6 +1076,198 @@ describe('dsh-workflow-workerthread', () => {
await handle.dispose()
}, 15_000)
it.each(['fulfills', 'rejects'] as const)('provider.start() reentrant cancellation refuses the run when readiness later %s', async (readinessOutcome) => {
const ctx = new Context()
await ctx.plugin(SubagentService)
const readiness = Promise.withResolvers<undefined>()
let starts = 0
let explicitCancels = 0
let disposals = 0
let sawAbortedSignal = false
const lifecycle: string[] = []
const provider: SubagentProvider = {
name: 'start-reentry',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: (request) => {
starts += 1
// This arbitrary provider callback runs before onChildStart can put
// the returned run in its registry. Cancellation must be rechecked
// after return instead of trusting the pre-start admission check.
handle.cancel('provider start reentered cancellation')
sawAbortedSignal = request.signal?.aborted === true
return {
id: AgentId('start-reentry-child'),
started: readiness.promise,
result: new Promise(() => { /* refusal owns teardown */ }),
// Deliberately honors only the explicit channel. It must still be
// reached promptly even though the first host fanout saw no run.
cancel: () => { explicitCancels += 1 },
dispose: () => {
disposals += 1
return Promise.resolve()
},
}
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, {
provider: 'start-reentry',
maxConcurrentAgents: 2,
disposeGraceMs: 30_000,
})
ctx.on('workflow/agent-start', () => { lifecycle.push('start') })
ctx.on('workflow/agent-end', () => { lifecycle.push('end') })
const handle = ctx.workflows.start({
...scripted("await agent('reentrant provider')\nreturn 'unreachable'"),
parent: fakeParent(),
})
await waitFor(() => { expect(starts).toBe(1) })
// Either later readiness settlement must not answer the already-refused
// start again or emit a workflow lifecycle pair.
if (readinessOutcome === 'fulfills') readiness.resolve(undefined)
else readiness.reject(new Error('late readiness rejection after refusal'))
let result: WorkflowResult | undefined
void handle.result.then((value) => { result = value })
await waitFor(() => {
expect(explicitCancels).toBe(1)
expect(disposals).toBe(1)
expect(result?.stopReason).toBe('cancelled')
}, 1000)
expect(sawAbortedSignal).toBe(true)
expect(lifecycle).toEqual([])
await handle.dispose()
expect(explicitCancels).toBe(1)
expect(disposals).toBe(1)
await ctx.fiber.dispose()
})
it('claims workflow and child disposal before a raw provider disposer reenters handle.dispose()', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
const terminal = Promise.withResolvers<SubagentResult>()
const observed: { reentrant?: Promise<void> } = {}
let starts = 0
let rawDisposeCalls = 0
const provider: SubagentProvider = {
name: 'dispose-reentry',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: () => {
starts += 1
return {
id: AgentId('dispose-reentry-child'),
started: Promise.resolve(),
result: terminal.promise,
cancel: () => { terminal.resolve({ output: [], stopReason: 'aborted' }) },
dispose: () => {
rawDisposeCalls += 1
observed.reentrant = handle.dispose()
return Promise.resolve()
},
}
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, { provider: 'dispose-reentry', maxConcurrentAgents: 2 })
const handle = ctx.workflows.start({
...scripted("await agent('live child')\nreturn 'unreachable'"),
parent: fakeParent(),
})
await waitFor(() => { expect(starts).toBe(1) })
const disposal = handle.dispose()
expect(observed.reentrant).toBe(disposal)
await disposal
expect(rawDisposeCalls).toBe(1)
await expect(handle.result).resolves.toMatchObject({ stopReason: 'cancelled' })
await ctx.fiber.dispose()
})
it('claims worker-originated child disposal before its raw disposer reenters holder disposal', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
const terminal = Promise.withResolvers<SubagentResult>()
const observed: { reentrant?: Promise<void> } = {}
let starts = 0
let rawDisposeCalls = 0
const provider: SubagentProvider = {
name: 'child-dispose-reentry',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: () => {
starts += 1
return {
id: AgentId('child-dispose-reentry-child'),
started: Promise.resolve(),
result: terminal.promise,
cancel: () => { terminal.resolve({ output: [], stopReason: 'aborted' }) },
dispose: () => {
rawDisposeCalls += 1
// This begins holder disposal from the worker's ChildDispose
// callback, before any public handle.dispose() call exists.
observed.reentrant = handle.dispose()
return Promise.resolve()
},
}
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, { provider: 'child-dispose-reentry', maxConcurrentAgents: 2 })
const handle = ctx.workflows.start({
...scripted("return await agent('settling child')"),
parent: fakeParent(),
})
const finishChildSpy = vi.spyOn(handle as unknown as {
finishChild(callId: number): void
}, 'finishChild')
await waitFor(() => { expect(starts).toBe(1) })
terminal.resolve({ output: [{ type: 'text', text: 'done' }], stopReason: 'completed' })
await waitFor(() => { expect(observed.reentrant).toBeDefined() }, 1000)
await observed.reentrant
expect(rawDisposeCalls).toBe(1)
expect(finishChildSpy.mock.calls.filter(([callId]) => callId === 1)).toHaveLength(1)
finishChildSpy.mockRestore()
await expect(handle.result).resolves.toMatchObject({ stopReason: 'cancelled' })
await ctx.fiber.dispose()
})
it('a grace-terminated worker reaps its child on exit without waiting for consumer dispose()', async () => {
const { ctx, parent, provider } = await setup({
manual: true,
config: { provider: 'stub', maxConcurrentAgents: 2, disposeGraceMs: 100 },
})
const handle = ctx.workflows.start({
// Let child-start cross, then make the worker unable to process its
// Cancel message. Grace settles the result and terminates the thread;
// that exit must independently own the host registry's disposal pass.
...scripted(`
agent('survives until exit reap')
for (let i = 0; i < 20; i++) await null
const end = Date.now() + 1500
while (Date.now() < end) {}
return 'unreachable'
`),
parent,
})
await waitFor(() => { expect(provider.runs).toHaveLength(1) })
handle.cancel('force termination')
const result = await handle.result
expect(result.stopReason).toBe('cancelled')
// Deliberately assert before handle.dispose(): host-owned worker exit,
// not consumer courtesy, is responsible for this resource guarantee.
await waitFor(() => { expect(provider.runs[0]!.disposed).toBe(true) }, 1000)
expect(provider.runs[0]!.disposeCalls).toBe(1)
await handle.dispose()
await ctx.fiber.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,
@@ -1045,23 +1395,71 @@ describe('dsh-workflow-workerthread', () => {
})
describe('worker death', () => {
it('the first death signal closes admission to messages Node delivers before exit', async () => {
const { ctx, parent, provider } = await setup({ manual: true })
const phases: string[] = []
ctx.on('workflow/phase', (_info, title) => { phases.push(title) })
const handle = ctx.workflows.start({
...scripted('await new Promise(() => {})'),
parent,
})
const worker = (handle as unknown as { worker: Worker }).worker
// Node may physically emit error -> queued message -> exit. Reproduce
// that ordering deterministically at the Worker event boundary: the
// late protocol data must not create work, narrate, or rewrite error.
worker.emit('error', new Error('synthetic error-before-message'))
worker.emit('message', { type: WorkerToHostType.Phase, title: 'late phase' })
worker.emit('message', {
type: WorkerToHostType.ChildStart,
callId: 999,
request: { prompt: 'late child' },
})
worker.emit('message', {
type: WorkerToHostType.Result,
result: { value: 'late', stopReason: 'completed', agentsStarted: 1 },
})
const result = await handle.result
expect(result.stopReason).toBe('error')
expect(result.error).toContain('synthetic error-before-message')
expect(provider.runs).toHaveLength(0)
expect(phases).toEqual([])
await handle.dispose()
await ctx.fiber.dispose()
})
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 signalAborts: unknown[] = []
const provider: SubagentProvider = {
name: 'doomed',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: () => ({
id: AgentId('doomed-child'),
started: Promise.resolve(),
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')),
}),
start: (request) => {
request.signal?.addEventListener('abort', () => {
signalAborts.push(request.signal?.reason)
// The death claim precedes the shared-signal fanout. This
// synchronous callback cannot turn death into cancellation.
handle.cancel('reentered from worker-death signal cleanup')
}, { once: true })
return {
id: AgentId('doomed-child'),
started: Promise.resolve(),
result: new Promise(() => { /* never settles; the reap is the teardown */ }),
cancel: (reason?: string) => {
cancelled.push(reason ?? 'cancelled')
// Exercise the later microtask case too: terminal ownership
// remains closed after the death callback returns.
queueMicrotask(() => { handle.cancel('reentered from worker-death child cleanup') })
},
dispose: () => Promise.reject(new Error('dispose exploded during reap')),
}
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, { provider: 'doomed', maxConcurrentAgents: 2 })
@@ -1089,7 +1487,12 @@ describe('dsh-workflow-workerthread', () => {
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 waitFor(() => {
expect(signalAborts).toEqual(['workflow worker gone'])
expect(cancelled).toEqual(['workflow worker gone'])
}, 1000)
await Promise.resolve()
expect(result.stopReason).toBe('error')
await handle.dispose()
}, 15_000)