From 64d0703c408016999948fd3bafabf207e652ed01 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:05:43 +0800 Subject: [PATCH] workflow: host-guarantee the agent-start/agent-end pairing on every stop path agent-end was worker-authored only, so a start already forwarded to observers lost its paired end whenever the worker could no longer speak - the grace force-settle terminating a wedged script, or an unexpected worker death - stranding progress consumers with agents that never finish (ds-review-bot finding on #233). The host now keeps a ledger of forwarded starts and funnels every agent-end through one gate: worker-reported ends pair (and clear) their entry, and both termination paths drain the remainder as synthesized 'cancelled' ends BEFORE the run settles, so ends always precede workflow/end. A real settlement racing the force-settle loses to the synthesized cancellation - the same first-wins override onResult applies to the run's own result. --- docs/cordis-catalog/events.md | 6 +- docs/cordis-catalog/services.md | 2 +- .../workflow/workflow-workerthread/README.md | 2 +- .../workflow-workerthread/src/host.ts | 54 ++++++++- .../tests/workflow-workerthread.spec.ts | 106 ++++++++++++++++++ packages/workflow/workflow/src/index.ts | 5 +- 6 files changed, 164 insertions(+), 11 deletions(-) diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 02e12359ac..28dc1f17f5 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -365,13 +365,13 @@ Source: [`packages/core/tools/src/index.ts:77`](../../packages/core/tools/src/in ### `workflow/agent-end` — emit -One `agent()` call settled (clean result, child failure, or run cancellation). Paired with Events['workflow/agent-start']. +One `agent()` call settled (clean result, child failure, or run cancellation). Paired with Events['workflow/agent-start'] by `agent.seq`, exactly once per started call on every stop path — on an engine termination path (a worker killed past its grace) the end is engine-synthesized with outcome `'cancelled'`. ```ts cordis-catalog 'workflow/agent-end'(info: WorkflowRunInfo, agent: WorkflowAgentEndInfo): void ``` -Source: [`packages/workflow/workflow/src/index.ts:93`](../../packages/workflow/workflow/src/index.ts) +Source: [`packages/workflow/workflow/src/index.ts:96`](../../packages/workflow/workflow/src/index.ts) ### `workflow/agent-start` — emit @@ -391,7 +391,7 @@ A workflow run settled (any stop reason). Fired when WorkflowRun.result resolves 'workflow/end'(info: WorkflowRunInfo, result: WorkflowResultInfo): void ``` -Source: [`packages/workflow/workflow/src/index.ts:103`](../../packages/workflow/workflow/src/index.ts) +Source: [`packages/workflow/workflow/src/index.ts:106`](../../packages/workflow/workflow/src/index.ts) ### `workflow/log` — emit diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 493381d1c9..0622e56a60 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -276,7 +276,7 @@ Semantics every implementation must honor: abstract start(request: WorkflowStartRequest): WorkflowRun ``` -Source: [`packages/workflow/workflow/src/index.ts:207`](../../packages/workflow/workflow/src/index.ts) +Source: [`packages/workflow/workflow/src/index.ts:210`](../../packages/workflow/workflow/src/index.ts) ## Inherited `ctx` members (cordis core + loader/hmr/timer) diff --git a/packages/workflow/workflow-workerthread/README.md b/packages/workflow/workflow-workerthread/README.md index dc8859d351..2d98273ae0 100644 --- a/packages/workflow/workflow-workerthread/README.md +++ b/packages/workflow/workflow-workerthread/README.md @@ -32,7 +32,7 @@ Values LEAVING the script (hook options/schemas, the script's return) are materi 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). 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`. -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. Once a run settles, stray children a script fired without awaiting are cancelled too, and `dispose()` waits for their disposal (bounded by the grace) before returning. +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. Once a run settles, stray children a script fired without awaiting are cancelled too, and `dispose()` 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. **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. diff --git a/packages/workflow/workflow-workerthread/src/host.ts b/packages/workflow/workflow-workerthread/src/host.ts index c512c15dee..28db34f097 100644 --- a/packages/workflow/workflow-workerthread/src/host.ts +++ b/packages/workflow/workflow-workerthread/src/host.ts @@ -22,7 +22,11 @@ * survivor when the worker dies or is terminated mid-flight. The three * 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). On a termination path `agentsStarted` reports 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 * 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. @@ -37,7 +41,7 @@ import type { Context } from 'cordis' import type { Agent } from '@deepseek-ai/dsh-agent' import { assertNever } from '@deepseek-ai/dsh-llm' import type { SubagentRun } from '@deepseek-ai/dsh-subagent' -import type { WorkflowMeta, WorkflowResult, WorkflowRun, WorkflowRunId } from '@deepseek-ai/dsh-workflow' +import type { WorkflowAgentEndInfo, WorkflowAgentInfo, WorkflowMeta, WorkflowResult, WorkflowRun, WorkflowRunId } from '@deepseek-ai/dsh-workflow' import { renderThrown } from './realm.ts' import type { ExecutionObserver } from './runtime.ts' import { HostToWorkerType, WorkerToHostType } from './protocol.ts' @@ -92,6 +96,8 @@ export class WorkerRun implements WorkflowRun { private readonly children = new Map() /** 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>() + /** Started-but-not-ended agents by seq — the pairing ledger the HOST guarantees (see {@link endAgent}). */ + private readonly liveAgents = new Map() private readonly quiescenceWaiters: (() => void)[] = [] /** The per-run abort fanout every child start request carries. */ private readonly controller = new AbortController() @@ -155,6 +161,10 @@ export class WorkerRun implements WorkflowRun { // ChildCancel relay (those later RPCs land as idempotent no-ops). for (const run of this.children.values()) run.cancel(this.cancelReason) this.graceTimer = setTimeout(() => { + // 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. + this.endStrandedAgents() this.settleResult(this.cancelledResult(this.hostStarted)) void this.worker.terminate() }, this.disposeGraceMs) @@ -225,13 +235,15 @@ export class WorkerRun implements WorkflowRun { if (this.cancelReason === undefined) this.observer.log(message.message) break case WorkerToHostType.AgentStart: + this.liveAgents.set(message.info.seq, message.info) this.observer.agentStart(message.info) break case WorkerToHostType.AgentEnd: // NOT suppressed on cancel: cancelled children report their paired - // agent-end with outcome 'cancelled' (the one-pair-per-started-child - // contract holds on every stop path). - this.observer.agentEnd(message.info) + // agent-end with outcome 'cancelled'. The gate (with the termination + // paths' synthesis) is what makes the one-pair-per-started-child + // contract hold on every stop path. + this.endAgent(message.info) break case WorkerToHostType.ChildStart: this.onChildStart(message.callId, message.request) @@ -373,6 +385,10 @@ export class WorkerRun implements WorkflowRun { 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) { @@ -382,6 +398,34 @@ export class WorkerRun implements WorkflowRun { this.settleResult({ value: null, stopReason: 'error', error: message, agentsStarted: this.hostStarted }) } + /** + * The single agent-end emission gate: forwards `end` iff its start is still + * unpaired in the ledger, so every forwarded `workflow/agent-start` gets + * EXACTLY one `workflow/agent-end` — the worker's own report where it can + * speak, a host-synthesized one where it cannot ({@link endStrandedAgents}). + * @param end - the settlement to emit (worker-reported or synthesized). + */ + private endAgent(end: WorkflowAgentEndInfo): void { + /* v8 ignore next -- a real end still in flight across the grace force-settle: not orderable in-process */ + if (!this.liveAgents.delete(end.seq)) return + this.observer.agentEnd(end) + } + + /** + * 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. + * 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`. + */ + private endStrandedAgents(): void { + for (const info of [...this.liveAgents.values()]) { + this.endAgent({ ...info, outcome: 'cancelled' }) + } + } + private cancelledResult(agentsStarted: number): WorkflowResult { // cancel() is the only writer of cancelReason and every caller checks it // first; the fallback guards the type, not a reachable path. diff --git a/packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts b/packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts index c1c26dcf0f..eb50ec1121 100644 --- a/packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts +++ b/packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts @@ -597,6 +597,73 @@ describe('dsh-workflow-workerthread', () => { // 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 vi.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 vi.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', () => { @@ -669,6 +736,45 @@ describe('dsh-workflow-workerthread', () => { 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 vi.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 diff --git a/packages/workflow/workflow/src/index.ts b/packages/workflow/workflow/src/index.ts index 6c2b797362..91caa314dc 100644 --- a/packages/workflow/workflow/src/index.ts +++ b/packages/workflow/workflow/src/index.ts @@ -85,7 +85,10 @@ declare module 'cordis' { 'workflow/agent-start'(info: WorkflowRunInfo, agent: WorkflowAgentInfo): void /** * One `agent()` call settled (clean result, child failure, or run - * cancellation). Paired with {@link Events['workflow/agent-start']}. + * cancellation). Paired with {@link Events['workflow/agent-start']} by + * `agent.seq`, exactly once per started call on every stop path — on an + * engine termination path (a worker killed past its grace) the end is + * engine-synthesized with outcome `'cancelled'`. * @param info - the run's identity snapshot. * @param agent - the call identity plus its outcome. * @mode emit