workflow, subagent: fix Codex code-review round-1 blockers
Six verified A-findings from the code-stage review, each with a regression test: - parallel()/pipeline() resolved to HOST arrays inside the vm realm, exposing host Array.prototype to scripts; combinator results are now realm-built (in-realm Array.from bound at context setup). - materializeFromRealm ran proxy traps (ownKeys/getOwnPropertyDescriptor/ getPrototypeOf) during the descriptor walk — realm code on the host stack, outside the vm timeout, escaping as raw errors; proxies (root, nested, and in the prototype position) are now rejected trap-free via util.types.isProxy before any inspection. - an already-aborted signal or an immediate cancel() no longer reports 'completed' for a hook-free script: drive() checks cancellation before running the body and again when the script settles. - dispose() now waits (bounded by disposeGraceMs) for stray agent() children to FINISH disposing, not just for the script to settle: every agent() call is tracked and quiesce() drains the in-flight set. - workflow/* event payloads were live mutable aliases shared across emissions; emitWorkflowEvent now hands each listener its own structural clone. - the structured-output turn-continuation veto is now prepend: true, so an earlier-registered force-continue listener cannot short-circuit it. Docs updated in the same change (READMEs, core-data-structures/workflow.md, the dynamic-workflows RFC, regenerated cordis catalogs).
This commit is contained in:
Tianyi Cui
@@ -323,7 +323,7 @@ One `agent()` call settled (clean result, child failure, or run cancellation). P
|
||||
'workflow/agent-end'(info: WorkflowRunInfo, agent: WorkflowAgentEndInfo): void
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:91`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:93`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
### `workflow/agent-start` — emit
|
||||
|
||||
@@ -333,7 +333,7 @@ One `agent()` call started a child run. Paired with Events['workflow/agent-end']
|
||||
'workflow/agent-start'(info: WorkflowRunInfo, agent: WorkflowAgentInfo): void
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:83`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:85`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
### `workflow/end` — emit
|
||||
|
||||
@@ -343,7 +343,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:101`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:103`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
### `workflow/log` — emit
|
||||
|
||||
@@ -353,7 +353,7 @@ The script emitted a narration line (a `log(message)` call).
|
||||
'workflow/log'(info: WorkflowRunInfo, message: string): void
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:75`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:77`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
### `workflow/phase` — emit
|
||||
|
||||
@@ -363,7 +363,7 @@ The script entered a phase (a `phase(title)` call) — progress grouping for obs
|
||||
'workflow/phase'(info: WorkflowRunInfo, title: string): void
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:68`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:70`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
### `workflow/start` — emit
|
||||
|
||||
@@ -373,7 +373,7 @@ A workflow run started — the script's meta block validated, the body about to
|
||||
'workflow/start'(info: WorkflowRunInfo): void
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:60`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:62`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
## Inherited events (cordis core + loader/hmr/timer)
|
||||
|
||||
|
||||
@@ -235,13 +235,13 @@ Semantics every implementation must honor:
|
||||
|
||||
- start throws synchronously for a request that cannot begin (an unparseable script, an invalid meta block). Once it returns a WorkflowRun, `result` NEVER rejects — every failure resolves with `stopReason: 'error'` (or `'cancelled'`).
|
||||
- The `workflow/*` events fire through emitWorkflowEvent (data snapshots, per-listener containment); `workflow/end` fires exactly once per started run, after `result` is settled or as it settles.
|
||||
- `dispose()` reaches quiescence within a bounded grace: it cancels, waits for the script to settle, and abandons a stuck script rather than hanging its caller (the engine documents what abandonment leaves behind).
|
||||
- `dispose()` reaches quiescence within a bounded grace: it cancels, waits for the script to settle AND its started children to finish disposing, and abandons whatever is left rather than hanging its caller (the engine documents what abandonment leaves behind).
|
||||
|
||||
```ts cordis-catalog
|
||||
abstract start(request: WorkflowStartRequest): WorkflowRun
|
||||
```
|
||||
|
||||
Source: [`packages/workflow/workflow/src/index.ts:188`](../../packages/workflow/workflow/src/index.ts)
|
||||
Source: [`packages/workflow/workflow/src/index.ts:191`](../../packages/workflow/workflow/src/index.ts)
|
||||
|
||||
## Inherited `ctx` members (cordis core + loader/hmr/timer)
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ interface WorkflowResult {
|
||||
|
||||
## A live run: `WorkflowRun`
|
||||
|
||||
The handle the consumer holds while a script executes. The consumer awaits `result`, may `cancel` mid-flight, and MUST `dispose` on every path. `result` does NOT reject — a script failure resolves with `stopReason: 'error'` — so the consumer maps a non-`completed` reason to an `isError` result. `dispose()` cancels, waits a bounded grace for the script to settle, then abandons it (the engine documents the abandonment semantics); it never hangs on a stuck script.
|
||||
The handle the consumer holds while a script executes. The consumer awaits `result`, may `cancel` mid-flight, and MUST `dispose` on every path. `result` does NOT reject — a script failure resolves with `stopReason: 'error'` — so the consumer maps a non-`completed` reason to an `isError` result. `dispose()` cancels, waits a bounded grace for the script to settle AND its children to finish disposing, then abandons whatever is left (the engine documents the abandonment semantics); it never hangs on a stuck script.
|
||||
|
||||
```ts type-equiv
|
||||
interface WorkflowRun {
|
||||
@@ -65,4 +65,4 @@ Hook misuse inside a script — bad arguments, unknown/deferred `agent()` option
|
||||
|
||||
## Events
|
||||
|
||||
The `workflow/*` events (`workflow/start`, `workflow/phase`, `workflow/log`, `workflow/agent-start`, `workflow/agent-end`, `workflow/end` — see the [events catalog](../cordis-catalog/events.md)) are **observe-only** emits carrying DATA SNAPSHOTS: every payload starts with `WorkflowRunInfo` (id + meta), never the live `WorkflowRun`, so a subscriber cannot gain `cancel`/`dispose`, and `workflow/end` deliberately omits the result value (a listener observing outcomes must not receive a mutable alias of the caller's result). Every emit is per-listener contained — a throwing subscriber is logged, never propagated, and cannot starve the listeners registered after it — mirroring `subagent/start`/`subagent/end`.
|
||||
The `workflow/*` events (`workflow/start`, `workflow/phase`, `workflow/log`, `workflow/agent-start`, `workflow/agent-end`, `workflow/end` — see the [events catalog](../cordis-catalog/events.md)) are **observe-only** emits carrying DATA SNAPSHOTS: every payload starts with `WorkflowRunInfo` (id + meta), never the live `WorkflowRun`, so a subscriber cannot gain `cancel`/`dispose`, and `workflow/end` deliberately omits the result value (a listener observing outcomes must not receive a mutable alias of the caller's result). Every emit is per-listener contained — a throwing subscriber is logged, never propagated, and cannot starve the listeners registered after it — and every listener receives its own payload clone, so mutating it corrupts neither the engine nor other listeners; the containment mirrors `subagent/start`/`subagent/end`.
|
||||
@@ -24,11 +24,11 @@ One deliberate strictness DIVERGENCE from CC: hook misuse — unknown or deferre
|
||||
|
||||
### The engine (dsh-workflow-vm): in-process node:vm
|
||||
|
||||
**Why node:vm and not isolated-vm/worker threads**: isolated-vm is in maintenance mode, needs `--no-node-snapshot` on EVERY consumer process (including the published bins) on Node ≥ 20, and falls back to node-gyp source builds; a worker-thread engine turns every hook into RPC and complicates the per-file coverage gate. Scripts are model-written — the same trust level as the model's existing bash access — so genuine sandboxing is not the current requirement. The interface/implementation split exists precisely so a hardened engine can swap in later. Accepted, documented limitations: vm is not a security boundary, and the vm timeout covers only the initial synchronous slice — a pathological synchronous spin after the first await cannot be killed in-process; `dispose()` cancels, waits a bounded grace, then abandons.
|
||||
**Why node:vm and not isolated-vm/worker threads**: isolated-vm is in maintenance mode, needs `--no-node-snapshot` on EVERY consumer process (including the published bins) on Node ≥ 20, and falls back to node-gyp source builds; a worker-thread engine turns every hook into RPC and complicates the per-file coverage gate. Scripts are model-written — the same trust level as the model's existing bash access — so genuine sandboxing is not the current requirement. The interface/implementation split exists precisely so a hardened engine can swap in later. Accepted, documented limitations: vm is not a security boundary, and the vm timeout covers only the initial synchronous slice — a pathological synchronous spin after the first await cannot be killed in-process; `dispose()` cancels, waits a bounded grace for the script to settle and its children to finish disposing, then abandons.
|
||||
|
||||
**Meta extraction**: a string/comment-aware brace scanner (template interpolation rejected) finds the literal; it is evaluated ALONE in an empty, timed vm context; the result must materialize to plain JSON data and pass shape validation (unknown fields rejected loud); the statement is blanked line-preservingly so stacks keep script line numbers.
|
||||
|
||||
**Realm boundary**: values entering the host (meta, hook options, schemas, the return value) go through `materializeFromRealm` — a descriptor walk that NEVER invokes accessors (the repo's `isJsonValue` is prototype-strict and getter-invoking, so it cannot run first; it would reject every cross-realm object and let realm code run outside the timed window) and rejects loud everything JSON cannot carry, copying via `Object.defineProperty` so a `"__proto__"` key becomes a data property, never a prototype mutation. Values entering the realm (`args`, `agent()` results) are rebuilt INSIDE the realm via the context's own `JSON.parse`, so the script never holds a live host-prototype object. Realm functions (stages, thunks) are called, never materialized.
|
||||
**Realm boundary**: values entering the host (meta, hook options, schemas, the return value) go through `materializeFromRealm` — a descriptor walk that NEVER invokes accessors (the repo's `isJsonValue` is prototype-strict and getter-invoking, so it cannot run first; it would reject every cross-realm object and let realm code run outside the timed window) and rejects loud everything JSON cannot carry, proxies included (the trap-free `util.types.isProxy`, checked before any inspection, so realm-side traps never run on the host stack), copying via `Object.defineProperty` so a `"__proto__"` key becomes a data property, never a prototype mutation. Values entering the realm (`args`, `agent()` results) are rebuilt INSIDE the realm via the context's own `JSON.parse`, and `parallel`/`pipeline` resolve to realm-built arrays, so the script never holds a live host-prototype object. Realm functions (stages, thunks) are called, never materialized.
|
||||
|
||||
**Containment**: every hook-returned promise carries a no-op rejection consumer, so a script that drops a promise cannot surface an unhandled rejection when cancellation rejects it — `dsh-app-boot` exits the process on unhandled rejections. Caps (`maxConcurrentAgents` auto = `min(16, cores - 2)`, `maxTotalAgents` 1000, `maxItemsPerCall` 4096) and timeouts are validated Config, not literals.
|
||||
|
||||
@@ -38,7 +38,7 @@ A `workflow` tool mirroring `dsh-tool-subagent`'s synchronous shape: start, awai
|
||||
|
||||
### The foundation: structured output on the subagent seam
|
||||
|
||||
`agent({schema})` needs `SubagentStartRequest.outputSchema` to actually work; it was vocabulary without an implementation (`outputSchema: false` everywhere). Implemented in `dsh-subagent-inprocess` for both in-process backends: a globally registered `structured_output` capture tool whose per-child schema is enforced by a `prepend: true` `agent/request` listener doing FINAL-REQUEST enforcement (post-processing `await next()` — cooperative mutation would not survive a downstream listener returning a replacement request), an `agent/turn-continuation` veto after capture (no wasted extra model step), validation-retry in-turn via `ToolArgsError`, and a clean-finish nudge loop (`structuredNudgeRetries`). Lifetime is refcounted by backends (plugin lifetime) AND live runs (start → settle). The seam's `outputSchema` type became the raw JSON-Schema SUBSET (`StructuredOutputSchema` in dsh-tools: single-string `type`, `properties`/`required`/`additionalProperties`, `items`, scalar `enum`/`const`; anything unenforced is rejected loud) — the schema travels verbatim to the model as the forced tool's parameters, so the wire format, not the author DSL, is the right vocabulary.
|
||||
`agent({schema})` needs `SubagentStartRequest.outputSchema` to actually work; it was vocabulary without an implementation (`outputSchema: false` everywhere). Implemented in `dsh-subagent-inprocess` for both in-process backends: a globally registered `structured_output` capture tool whose per-child schema is enforced by a `prepend: true` `agent/request` listener doing FINAL-REQUEST enforcement (post-processing `await next()` — cooperative mutation would not survive a downstream listener returning a replacement request), a `prepend: true` `agent/turn-continuation` veto after capture (no wasted extra model step, and an earlier-registered force-continue listener cannot short-circuit it), validation-retry in-turn via `ToolArgsError`, and a clean-finish nudge loop (`structuredNudgeRetries`). Lifetime is refcounted by backends (plugin lifetime) AND live runs (start → settle). The seam's `outputSchema` type became the raw JSON-Schema SUBSET (`StructuredOutputSchema` in dsh-tools: single-string `type`, `properties`/`required`/`additionalProperties`, `items`, scalar `enum`/`const`; anything unenforced is rejected loud) — the schema travels verbatim to the model as the forced tool's parameters, so the wire format, not the author DSL, is the right vocabulary.
|
||||
|
||||
## What was rejected
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ Runs a child as a child [`Agent`](../../core/agent) on the same cordis context (
|
||||
The mechanism behind `outputSchema` for in-process children. One globally registered `structured_output` capture tool (its registered parameters are a placeholder) plus two listeners, registered once per root context and shared by every holder:
|
||||
|
||||
- an `agent/request` waterfall listener registered `prepend: true` that post-processes `await next()` — **final-request enforcement**: the request that hits the wire never carries `structured_output` for an agent without a structured run, and always carries the run's OWN schema (as the tool's `parameters`) for one that has it. Per-agent shaping lives here because the tool registry and prompt assembly are context-global while schemas differ per concurrent child; cooperative mutate-then-`next()` would not survive a downstream listener returning a replacement request.
|
||||
- an `agent/turn-continuation` listener that stops a child's turn once its output is captured, so a successful capture doesn't buy a wasted extra model step.
|
||||
- an `agent/turn-continuation` listener (also `prepend: true` — an earlier-registered force-continue listener returning without `next()` must not decide the turn before the veto runs) that stops a child's turn once its output is captured, so a successful capture doesn't buy a wasted extra model step.
|
||||
|
||||
The capture tool validates each call against the run's schema (`validateStructuredValue`) — violations become an `INVALID_ARGS` isError result the model retries in-turn; a valid call records the value.
|
||||
|
||||
|
||||
@@ -18,7 +18,9 @@
|
||||
*
|
||||
* A companion `agent/turn-continuation` listener stops a child's turn once its
|
||||
* output is captured — without it, the loop's default "had tool calls ⇒
|
||||
* continue" buys a wasted extra model step per structured child.
|
||||
* continue" buys a wasted extra model step per structured child. It is also
|
||||
* `prepend: true`: the veto must run before any earlier-registered listener
|
||||
* that could short-circuit the chain into a forced continue.
|
||||
*
|
||||
* Lifetime is refcounted with two kinds of holder: each backend acquires for
|
||||
* its plugin lifetime (so the tool exists before any run), and each structured
|
||||
@@ -183,11 +185,15 @@ function registerRuntime(root: Context, runtime: StructuredRuntime): void {
|
||||
|
||||
// Stop a structured child's turn once its output is captured: the default
|
||||
// "had tool calls ⇒ continue" would otherwise buy a wasted extra model step
|
||||
// after every successful capture.
|
||||
// after every successful capture. `prepend: true` puts the veto OUTERMOST —
|
||||
// an earlier-registered listener that short-circuits the chain (a goal-style
|
||||
// force-continue returning without `next()`) would otherwise decide the turn
|
||||
// before this listener ever ran, and no downstream decision may resurrect a
|
||||
// structured turn that is already finished.
|
||||
runtime.disposers.push(root.on('agent/turn-continuation', function (
|
||||
this: unknown, agent: Agent, _turn: number, _decision: ContinuationDecision, next: () => Promise<ContinuationDecision>,
|
||||
): Promise<ContinuationDecision> {
|
||||
if (runtime.states.get(agent)?.captured) return Promise.resolve({ action: 'stop' })
|
||||
return next()
|
||||
}))
|
||||
}, { prepend: true }))
|
||||
}
|
||||
@@ -5,6 +5,7 @@ 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 type { Agent, ContinuationDecision } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import * as Invariants from '@deepseek-ai/dsh-invariants'
|
||||
import SubagentService, { type SubagentStartRequest } from '@deepseek-ai/dsh-subagent'
|
||||
@@ -85,6 +86,34 @@ describe('in-process structured output', () => {
|
||||
await run.dispose()
|
||||
})
|
||||
|
||||
it('the captured-turn veto is prepend: an EARLIER force-continue listener cannot short-circuit it', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
// Registered BEFORE the structured runtime exists — without prepend, this
|
||||
// goal-style listener would decide the turn first (returning WITHOUT
|
||||
// calling next()) and the veto would never run.
|
||||
ctx.on('agent/turn-continuation', () => Promise.resolve<ContinuationDecision>({ action: 'continue' }))
|
||||
const acquisition = acquireStructuredRuntime(ctx)
|
||||
const agent = { id: AgentId('structured-child') } as unknown as Agent
|
||||
acquisition.attach(agent, SCHEMA)
|
||||
const captured = await ctx.tools.execute({
|
||||
callId: 'call-1' as never,
|
||||
name: STRUCTURED_OUTPUT_TOOL,
|
||||
arguments: { answer: 1 },
|
||||
agent,
|
||||
})
|
||||
expect(captured.isError).toBeFalsy()
|
||||
const decision = await ctx.waterfall(
|
||||
'agent/turn-continuation', agent, 1,
|
||||
{ action: 'continue' },
|
||||
() => Promise.resolve<ContinuationDecision>({ action: 'continue' }),
|
||||
)
|
||||
expect(decision).toEqual({ action: 'stop' })
|
||||
acquisition.detach(agent)
|
||||
acquisition.release()
|
||||
})
|
||||
|
||||
it('an invalid call gets an INVALID_ARGS isError result and the model retries in-turn', async () => {
|
||||
const { ctx, parent } = await setup([
|
||||
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 'not-a-number' }),
|
||||
|
||||
@@ -10,11 +10,11 @@ The first [`WorkflowService`](../workflow/README.md) implementation: an in-proce
|
||||
|
||||
## Realm discipline
|
||||
|
||||
Values ENTERING the host (the meta literal, hook options/schemas, the script's return) are materialized by `materializeFromRealm`: a descriptor walk that never invokes accessors and rejects loud everything JSON cannot carry (accessors, exotic prototypes, functions, symbols, cycles, sparse arrays, non-finite numbers, nested `undefined`), copying into host containers via `defineProperty` so a `"__proto__"` key becomes a data property, never a prototype mutation. Values ENTERING the realm (`args`, `agent()` results) are rebuilt INSIDE the realm through the context's own `JSON.parse`, so the script never holds an object whose prototype chain reaches host intrinsics.
|
||||
Values ENTERING the host (the meta literal, hook options/schemas, the script's return) are materialized by `materializeFromRealm`: a descriptor walk that never invokes accessors and rejects loud everything JSON cannot carry (accessors, exotic prototypes, functions, symbols, cycles, sparse arrays, non-finite numbers, nested `undefined`, and proxies — rejected via the trap-free `util.types.isProxy` BEFORE any inspection could run a realm-side trap on the host stack), copying into host containers via `defineProperty` so a `"__proto__"` key becomes a data property, never a prototype mutation. Values ENTERING the realm (`args`, `agent()` results) are rebuilt INSIDE the realm through the context's own `JSON.parse`, and the arrays `parallel`/`pipeline` resolve to are realm-built, so the script never holds an object whose prototype chain reaches host intrinsics.
|
||||
|
||||
## Limits, cancellation, disposal
|
||||
|
||||
Per-run: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` aborts every child (a shared `AbortSignal`), rejects waiting `agent()` slots, and makes every future hook call throw `CANCELLED` — the script dies at its next await and the run settles `cancelled`. Once a run settles, stray children a script fired without awaiting are aborted too. Every hook-returned promise carries a no-op rejection consumer, so a dropped promise cannot surface an unhandled rejection (the app boot layer exits the process on those).
|
||||
Per-run: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` aborts every child (a shared `AbortSignal`), rejects waiting `agent()` slots, and makes every future hook call throw `CANCELLED` — the script dies at its next await and the run settles `cancelled`; a cancellation that lands before the body runs (or before it settles) reports `cancelled` even if the script itself needed no hooks. Once a run settles, stray children a script fired without awaiting are aborted too, and `dispose()` waits for those children to finish disposing (bounded by the grace) before returning. Every hook-returned promise carries a no-op rejection consumer, so a dropped promise cannot surface an unhandled rejection (the app boot layer exits the process on those).
|
||||
|
||||
**Documented limitations** (the accepted cost of the in-process mechanism; the seam exists so a worker-thread/isolated-vm engine can swap in): vm is NOT a security boundary — scripts are model-written, the same trust level as the model's bash access — and the vm `timeout` covers only the initial synchronous slice, so a pathological synchronous spin after the first await cannot be killed; `dispose()` waits `disposeGraceMs` then ABANDONS such a script (its settlement stays contained, but an abandoned spin would still occupy the event loop).
|
||||
|
||||
@@ -27,4 +27,4 @@ Per-run: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap
|
||||
| `maxTotalAgents` | `1000` | Total `agent()` calls one run may start (runaway-loop backstop). |
|
||||
| `maxItemsPerCall` | `4096` | Items accepted by one `parallel()`/`pipeline()` call. |
|
||||
| `syncTimeoutMs` | `5000` | vm timeout for the initial synchronous slice and the meta evaluation. |
|
||||
| `disposeGraceMs` | `5000` | How long `dispose()` waits for a cancelled script before abandoning it. |
|
||||
| `disposeGraceMs` | `5000` | How long `dispose()` waits for a cancelled script and its children before abandoning them. |
|
||||
@@ -13,10 +13,12 @@
|
||||
* is correctness containment, not a sandbox.
|
||||
* - The vm `timeout` covers only the initial SYNCHRONOUS slice of the script;
|
||||
* a pathological synchronous spin after the first await cannot be killed
|
||||
* in-process. `dispose()` therefore waits a bounded grace and then ABANDONS
|
||||
* a stuck script: its pending hook promises are already rejected and its
|
||||
* settlement is contained (no unhandled rejection), but an abandoned
|
||||
* synchronous spin would still occupy the event loop.
|
||||
* in-process. `dispose()` waits a bounded grace for the script to settle
|
||||
* AND its children (stray `agent()` calls included) to finish disposing,
|
||||
* then ABANDONS whatever is left: pending hook promises are already
|
||||
* rejected and the script's settlement is contained (no unhandled
|
||||
* rejection), but an abandoned synchronous spin would still occupy the
|
||||
* event loop.
|
||||
*
|
||||
* Plugin export shape: a default-exported {@link WorkflowService} subclass
|
||||
* (the class-based service form, like `dsh-bash-local`).
|
||||
@@ -142,12 +144,22 @@ export class VmWorkflowEngine extends WorkflowService {
|
||||
execution.cancel(reason)
|
||||
},
|
||||
dispose: (): Promise<void> => {
|
||||
// Idempotent: cancel, then wait min(settle, grace). `result` never
|
||||
// rejects, so the race needs no rejection handling; an unsettled
|
||||
// script past the grace is abandoned per the module contract.
|
||||
// Idempotent: cancel, then wait min(settle + child quiescence, grace).
|
||||
// `result` and `quiesce()` never reject, so the race needs no
|
||||
// rejection handling; a script or child still unsettled past the grace
|
||||
// is abandoned per the module contract.
|
||||
disposed ??= (async () => {
|
||||
execution.cancel('workflow disposed')
|
||||
await Promise.race([result, sleep(this.config.disposeGraceMs)])
|
||||
await Promise.race([
|
||||
(async () => {
|
||||
await result
|
||||
// The result settles with the SCRIPT; stray children a script
|
||||
// fired without awaiting are still winding down — dispose must
|
||||
// not return while they hold live resources.
|
||||
await execution.quiesce()
|
||||
})(),
|
||||
sleep(this.config.disposeGraceMs),
|
||||
])
|
||||
})()
|
||||
return disposed
|
||||
},
|
||||
|
||||
@@ -10,7 +10,13 @@
|
||||
* host containers, rejecting loud everything JSON cannot carry:
|
||||
* accessor properties, non-plain prototypes, functions, symbols (keys or
|
||||
* values), bigints, non-finite numbers, `undefined` values, cycles, sparse
|
||||
* arrays, and arrays with non-index own properties.
|
||||
* arrays, arrays with non-index own properties, and proxies. Proxies are
|
||||
* rejected via the trap-free native `util.types.isProxy` check BEFORE any
|
||||
* other inspection — a descriptor walk over a proxy would otherwise run its
|
||||
* realm-side traps (`ownKeys`, `getOwnPropertyDescriptor`, `getPrototypeOf`)
|
||||
* on the host stack, outside the vm's timed window, and a throwing trap would
|
||||
* escape as a raw realm error instead of a {@link MaterializeError}. The same
|
||||
* check guards the PROTOTYPE position (an object whose prototype is a proxy).
|
||||
*
|
||||
* Host objects are built with `Object.defineProperty` into a fresh `{}` —
|
||||
* never plain `target[key] =` assignment, which a `"__proto__"` key would turn
|
||||
@@ -24,6 +30,8 @@
|
||||
* @module @deepseek-ai/dsh-workflow-vm/realm
|
||||
*/
|
||||
|
||||
import { types } from 'node:util'
|
||||
|
||||
/** Thrown by {@link materializeFromRealm}; the caller wraps it into the right `WorkflowError` code. */
|
||||
export class MaterializeError extends Error {
|
||||
constructor(public readonly path: string, public readonly reason: string) {
|
||||
@@ -36,11 +44,13 @@ export class MaterializeError extends Error {
|
||||
* Whether an object's prototype chain is data-shaped: `null`, or a prototype
|
||||
* whose own prototype is `null` (the realm's `Object.prototype` — which we
|
||||
* cannot compare by identity across realms). A `Date`/`Map`/class instance
|
||||
* has a longer chain and is rejected.
|
||||
* has a longer chain and is rejected, as is a proxy sitting in the prototype
|
||||
* position (checked trap-free BEFORE its own prototype is dereferenced).
|
||||
*/
|
||||
function hasPlainPrototype(value: object): boolean {
|
||||
const proto: unknown = Object.getPrototypeOf(value)
|
||||
if (proto === null) return true
|
||||
if (types.isProxy(proto)) return false
|
||||
return Object.getPrototypeOf(proto) === null
|
||||
}
|
||||
|
||||
@@ -81,6 +91,11 @@ function materialize(value: unknown, path: string, seen: Set<object>): unknown {
|
||||
break
|
||||
}
|
||||
if (value === null) return null
|
||||
// BEFORE anything else touches the object: every inspection below —
|
||||
// Array.isArray aside — can trigger a proxy trap, running realm code on the
|
||||
// host stack (module doc). isProxy is a native internal-slot check (no
|
||||
// traps, catches revoked proxies, realm-agnostic).
|
||||
if (types.isProxy(value)) throw new MaterializeError(path, 'proxies cannot cross the workflow realm boundary')
|
||||
const objectValue: object = value
|
||||
if (seen.has(objectValue)) throw new MaterializeError(path, 'circular references are not JSON data')
|
||||
seen.add(objectValue)
|
||||
|
||||
@@ -9,8 +9,11 @@
|
||||
* descriptor walks; values ENTERING the realm from the host (`args`, agent()
|
||||
* results) are rebuilt INSIDE the realm through the context's own
|
||||
* `JSON.parse`, so the script never holds an object whose prototype chain
|
||||
* reaches host intrinsics. Realm functions (pipeline stages, parallel thunks)
|
||||
* are called, not materialized — their values stay realm-side.
|
||||
* reaches host intrinsics. The arrays `parallel`/`pipeline` resolve to are
|
||||
* realm-built for the same reason (their ELEMENTS are realm values already —
|
||||
* only the container needs rebuilding). Realm functions (pipeline stages,
|
||||
* parallel thunks) are called, not materialized — their values stay
|
||||
* realm-side.
|
||||
*
|
||||
* Failure discipline: fatal {@link WorkflowError}s (bad hook arguments,
|
||||
* unsupported options/schemas, tripped caps, seam start failures,
|
||||
@@ -132,7 +135,10 @@ export class WorkflowExecution {
|
||||
private currentPhase: string | undefined
|
||||
private readonly context: vm.Context
|
||||
private readonly realmJsonParse: (text: string) => unknown
|
||||
private readonly realmArrayFrom: (items: unknown[]) => unknown[]
|
||||
private readonly compiled: vm.Script
|
||||
/** Every live `agent()` call promise — awaited or stray — for {@link quiesce}. */
|
||||
private readonly inFlightAgents = new Set<Promise<unknown>>()
|
||||
|
||||
constructor(
|
||||
private readonly ctx: Context,
|
||||
@@ -162,9 +168,12 @@ export class WorkflowExecution {
|
||||
// The realm's own JSON.parse — the host→realm rebuild channel.
|
||||
const realmJson = vm.runInContext('JSON', this.context) as { parse(text: string): unknown }
|
||||
this.realmJsonParse = (text: string) => realmJson.parse(text)
|
||||
// The realm's own Array.from, bound NOW so a script reassigning its
|
||||
// globals later cannot swap it: combinator results must be realm arrays.
|
||||
this.realmArrayFrom = vm.runInContext('Array.from.bind(Array)', this.context) as (items: unknown[]) => unknown[]
|
||||
|
||||
const globals: Record<string, unknown> = {
|
||||
agent: (prompt: unknown, opts?: unknown) => this.contain(this.agent(prompt, opts)),
|
||||
agent: (prompt: unknown, opts?: unknown) => this.contain(this.track(this.agent(prompt, opts))),
|
||||
parallel: (thunks: unknown) => this.contain(this.parallel(thunks)),
|
||||
pipeline: (items: unknown, ...stages: unknown[]) => this.contain(this.pipeline(items, stages)),
|
||||
phase: (title: unknown) => { this.phase(title) },
|
||||
@@ -216,8 +225,15 @@ export class WorkflowExecution {
|
||||
*/
|
||||
async drive(): Promise<WorkflowResult> {
|
||||
try {
|
||||
// Cancelled before the body ever ran (an already-aborted start signal):
|
||||
// the script must not execute at all, let alone report `completed`.
|
||||
if (this.isCancelled()) throw this.cancelledError()
|
||||
const scriptPromise = this.compiled.runInContext(this.context, { timeout: this.limits.syncTimeoutMs }) as Promise<unknown>
|
||||
const raw: unknown = await this.contain(Promise.resolve(scriptPromise))
|
||||
// Cancelled while the body ran: a script that settled without touching
|
||||
// another hook (or without any) must still report `cancelled` — the
|
||||
// holder asked for cancellation and `completed` would be a lie.
|
||||
if (this.isCancelled()) throw this.cancelledError()
|
||||
const value = raw === undefined ? null : this.materializeResult(raw)
|
||||
return { value, stopReason: 'completed', agentsStarted: this.started }
|
||||
} catch (error: unknown) {
|
||||
@@ -245,6 +261,31 @@ export class WorkflowExecution {
|
||||
return promise
|
||||
}
|
||||
|
||||
/**
|
||||
* Register one `agent()` call promise for {@link quiesce} tracking; the
|
||||
* entry drops when the call fully settles (which is AFTER its child's
|
||||
* `dispose()` — the call wrapper disposes in its `finally`).
|
||||
*/
|
||||
private track<T>(promise: Promise<T>): Promise<T> {
|
||||
this.inFlightAgents.add(promise)
|
||||
const drop = (): void => { this.inFlightAgents.delete(promise) }
|
||||
promise.then(drop, drop)
|
||||
return promise
|
||||
}
|
||||
|
||||
/**
|
||||
* Settles once every `agent()` call — awaited or stray — has fully settled,
|
||||
* INCLUDING each child's `dispose()`. The reap in {@link drive}'s finally
|
||||
* aborts strays; this is the wait for those aborts to reach quiescence, so
|
||||
* the engine's `dispose()` cannot return while a child is still winding
|
||||
* down. Never rejects (the tracked promises' rejections are contained).
|
||||
*/
|
||||
async quiesce(): Promise<void> {
|
||||
while (this.inFlightAgents.size > 0) {
|
||||
await Promise.allSettled([...this.inFlightAgents])
|
||||
}
|
||||
}
|
||||
|
||||
private cancelledError(): WorkflowError {
|
||||
// cancel() arms cancelError before any caller can observe isCancelled()
|
||||
// === true; the fallback guards the type, not a reachable path.
|
||||
@@ -430,7 +471,7 @@ export class WorkflowExecution {
|
||||
}
|
||||
return thunk as () => unknown
|
||||
})
|
||||
return Promise.all(thunks.map(async (thunk) => {
|
||||
const settled = await Promise.all(thunks.map(async (thunk) => {
|
||||
try {
|
||||
return await thunk()
|
||||
} catch (error: unknown) {
|
||||
@@ -438,6 +479,9 @@ export class WorkflowExecution {
|
||||
return null
|
||||
}
|
||||
}))
|
||||
// The container must be a REALM array (module doc); the elements are
|
||||
// realm values already.
|
||||
return this.realmArrayFrom(settled)
|
||||
}
|
||||
|
||||
/** The `pipeline(items, ...stages)` hook: per-item stage chains, NO cross-stage barrier. */
|
||||
@@ -455,7 +499,7 @@ export class WorkflowExecution {
|
||||
}
|
||||
return stage as (previous: unknown, item: unknown, index: number) => unknown
|
||||
})
|
||||
return Promise.all(rawItems.map(async (item: unknown, index) => {
|
||||
const settled = await Promise.all(rawItems.map(async (item: unknown, index) => {
|
||||
let value: unknown = item
|
||||
try {
|
||||
for (const stage of stages) {
|
||||
@@ -469,6 +513,9 @@ export class WorkflowExecution {
|
||||
return null
|
||||
}
|
||||
}))
|
||||
// The container must be a REALM array (module doc); the elements are
|
||||
// realm values already.
|
||||
return this.realmArrayFrom(settled)
|
||||
}
|
||||
|
||||
private assertItemCap(length: number, hook: string): void {
|
||||
|
||||
@@ -108,6 +108,14 @@ return 2`
|
||||
expect(error.message).toContain('JSON data')
|
||||
})
|
||||
|
||||
it('rejects a meta literal containing a proxy as META_INVALID — its traps never run', () => {
|
||||
// bad() rethrows anything that is not a WorkflowError, so a trap firing
|
||||
// ('trap ran') would fail this test instead of mapping to META_INVALID.
|
||||
const error = bad('export const meta = { name: "x", description: "d", phases: new Proxy([], { getPrototypeOf() { throw new Error("trap ran") } }) }')
|
||||
expect(error.code).toBe('META_INVALID')
|
||||
expect(error.message).toContain('proxies cannot cross')
|
||||
})
|
||||
|
||||
it('rejects shape violations with EVERY violation listed (META_INVALID)', () => {
|
||||
const error = bad('export const meta = { description: 7, bogus: 1 }\nreturn 1')
|
||||
expect(error.code).toBe('META_INVALID')
|
||||
|
||||
@@ -81,6 +81,27 @@ describe('materializeFromRealm', () => {
|
||||
expect(materializeFromRealm(inRealm('Object.assign(Object.create(null), { a: 1 })'))).toEqual({ a: 1 })
|
||||
})
|
||||
|
||||
it('rejects proxies (root, nested, revoked, host-realm) WITHOUT running any trap', () => {
|
||||
const trapped = inRealm(`new Proxy({ a: 1 }, {
|
||||
ownKeys() { throw new Error('trap ran') },
|
||||
getOwnPropertyDescriptor() { throw new Error('trap ran') },
|
||||
getPrototypeOf() { throw new Error('trap ran') },
|
||||
})`)
|
||||
// A trap firing would surface 'trap ran' (a non-MaterializeError) instead.
|
||||
expect(rejection(trapped)).toContain('proxies cannot cross')
|
||||
expect(rejection(inRealm('{ nested: new Proxy([], {}) }'))).toContain('value.nested')
|
||||
const revoked = inRealm('(() => { const r = Proxy.revocable({}, {}); r.revoke(); return r.proxy })()')
|
||||
expect(rejection(revoked)).toContain('proxies cannot cross')
|
||||
expect(rejection(new Proxy({}, {}))).toContain('proxies cannot cross')
|
||||
})
|
||||
|
||||
it('rejects an object whose PROTOTYPE is a proxy without dereferencing through it', () => {
|
||||
const value = inRealm(`Object.create(new Proxy({}, {
|
||||
getPrototypeOf() { throw new Error('trap ran') },
|
||||
}))`)
|
||||
expect(rejection(value)).toContain('exotic prototype')
|
||||
})
|
||||
|
||||
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 } })()')
|
||||
|
||||
@@ -36,6 +36,7 @@ class StubProvider implements SubagentProvider {
|
||||
constructor(
|
||||
readonly name: string,
|
||||
private readonly reply?: (request: SubagentStartRequest, index: number) => SubagentResult,
|
||||
private readonly disposeDelayMs = 0,
|
||||
) {}
|
||||
|
||||
start(request: SubagentStartRequest): SubagentRun {
|
||||
@@ -57,8 +58,17 @@ class StubProvider implements SubagentProvider {
|
||||
settle({ output: [], stopReason: 'aborted' })
|
||||
},
|
||||
dispose: () => {
|
||||
controlled.disposed = true
|
||||
return Promise.resolve()
|
||||
if (this.disposeDelayMs === 0) {
|
||||
controlled.disposed = true
|
||||
return Promise.resolve()
|
||||
}
|
||||
// A slow-winding child (quiescence tests): disposal completes late.
|
||||
return new Promise<void>((resolve) => {
|
||||
setTimeout(() => {
|
||||
controlled.disposed = true
|
||||
resolve()
|
||||
}, this.disposeDelayMs)
|
||||
})
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -73,12 +83,17 @@ 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')))
|
||||
const provider = new StubProvider(
|
||||
'stub',
|
||||
options?.manual ? undefined : options?.reply ?? (() => text('stub reply')),
|
||||
options?.disposeDelayMs ?? 0,
|
||||
)
|
||||
ctx.subagents.registerProvider(provider)
|
||||
await ctx.plugin(VmWorkflowEngine, { provider: 'stub', ...options?.config })
|
||||
return { ctx, provider, parent: fakeParent() }
|
||||
@@ -405,6 +420,50 @@ describe('dsh-workflow-vm', () => {
|
||||
expect((await run(ctx, parent, script('return typeof args'))).value).toBe('undefined')
|
||||
})
|
||||
|
||||
it('parallel/pipeline resolve to REALM arrays: instanceof holds in-script, host intrinsics stay unreachable', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const result = await run(ctx, parent, script(`
|
||||
const fromParallel = await parallel([() => agent('a'), () => 'plain'])
|
||||
const fromPipeline = await pipeline([1], (prev) => prev + 1)
|
||||
Object.getPrototypeOf(fromParallel).polluted = 'realm-only'
|
||||
return {
|
||||
parallelIsRealmArray: fromParallel instanceof Array,
|
||||
pipelineIsRealmArray: fromPipeline instanceof Array,
|
||||
values: [fromParallel[1], fromPipeline[0]],
|
||||
}
|
||||
`))
|
||||
expect(result.stopReason).toBe('completed')
|
||||
expect(result.value).toEqual({
|
||||
parallelIsRealmArray: true,
|
||||
pipelineIsRealmArray: true,
|
||||
values: ['plain', 2],
|
||||
})
|
||||
// The script's prototype mutation stayed realm-side: the HOST
|
||||
// Array.prototype was never reachable through a combinator result.
|
||||
expect(([] as unknown as Record<string, unknown>).polluted).toBeUndefined()
|
||||
})
|
||||
|
||||
it('a returned proxy is rejected as RESULT_UNSERIALIZABLE without running its traps', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const result = await run(ctx, parent, script(`
|
||||
return new Proxy({ a: 1 }, { ownKeys() { throw new Error('trap ran') } })
|
||||
`))
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('not plain JSON data')
|
||||
expect(result.error).toContain('proxies cannot cross')
|
||||
expect(result.error).not.toContain('trap ran')
|
||||
})
|
||||
|
||||
it('agent() options passed as a proxy are rejected loudly, traps never invoked', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const result = await run(ctx, parent, script(`
|
||||
return await agent('p', new Proxy({}, { ownKeys() { throw new Error('trap ran') } }))
|
||||
`))
|
||||
expect(result.stopReason).toBe('error')
|
||||
expect(result.error).toContain('options must be plain JSON data')
|
||||
expect(result.error).not.toContain('trap ran')
|
||||
})
|
||||
|
||||
it('a non-JSON return value fails loud as RESULT_UNSERIALIZABLE', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const withDate = await run(ctx, parent, script('return { when: new Date(0) }'))
|
||||
@@ -452,6 +511,51 @@ describe('dsh-workflow-vm', () => {
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('an already-aborted signal cancels a HOOK-FREE script: the body never runs at all', async () => {
|
||||
const { ctx, parent } = 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({ script: script("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([])
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('cancel() right after start() reports cancelled even when the script needed no hooks', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const handle = ctx.workflows.start({ script: script('return 123'), parent })
|
||||
handle.cancel('changed my mind')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(result.value).toBeNull()
|
||||
expect(result.error).toContain('changed my mind')
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('an agent() call AFTER a mid-run cancel rejects at entry — no child ever starts', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const handle = ctx.workflows.start({
|
||||
script: script(`
|
||||
await agent('first')
|
||||
return await agent('second')
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
await vi.waitFor(() => { expect(provider.runs.length).toBe(1) })
|
||||
// Same synchronous block: the first child settles completed, then the
|
||||
// cancel lands BEFORE the script's continuation can call agent() again.
|
||||
provider.runs[0]!.settle(text('first done'))
|
||||
handle.cancel('mid-run')
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('cancelled')
|
||||
expect(provider.runs.length).toBe(1)
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('the signal aborting mid-run cancels like cancel()', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true })
|
||||
const controller = new AbortController()
|
||||
@@ -577,6 +681,24 @@ describe('dsh-workflow-vm', () => {
|
||||
})
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('dispose() waits for a stray child to FINISH disposing (quiescence), not just the script settle', async () => {
|
||||
const { ctx, parent, provider } = await setup({ manual: true, disposeDelayMs: 40 })
|
||||
const handle = ctx.workflows.start({
|
||||
script: script(`
|
||||
agent('stray')
|
||||
return 'done without awaiting'
|
||||
`),
|
||||
parent,
|
||||
})
|
||||
const result = await handle.result
|
||||
expect(result.stopReason).toBe('completed')
|
||||
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.
|
||||
expect(provider.runs[0]!.disposed).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('service surface', () => {
|
||||
@@ -595,6 +717,24 @@ describe('dsh-workflow-vm', () => {
|
||||
await second.dispose()
|
||||
})
|
||||
|
||||
it('a listener mutating one event payload cannot corrupt later events (per-emission snapshots)', async () => {
|
||||
const { ctx, parent } = await setup()
|
||||
const ends: unknown[] = []
|
||||
let endInfo: WorkflowRunInfo | undefined
|
||||
ctx.on('workflow/agent-start', (info, agent) => {
|
||||
agent.seq = 999
|
||||
agent.label = 'HACKED'
|
||||
info.meta.name = 'HACKED'
|
||||
})
|
||||
ctx.on('workflow/agent-end', (info, agent) => {
|
||||
ends.push(agent)
|
||||
endInfo = info
|
||||
})
|
||||
await run(ctx, parent, script("return await agent('job', { label: 'honest' })"))
|
||||
expect(ends[0]).toMatchObject({ seq: 1, label: 'honest', outcome: 'completed' })
|
||||
expect(endInfo!.meta.name).toBe('test-flow')
|
||||
})
|
||||
|
||||
it('unregisters ctx.workflows when the engine fiber is disposed (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
|
||||
@@ -4,9 +4,9 @@ The **workflow seam** (`ctx.workflows`): an abstract service defining WHAT a wor
|
||||
|
||||
## Service: `WorkflowService` (abstract)
|
||||
|
||||
`start(request: WorkflowStartRequest): WorkflowRun` — parse and execute a script. Throws synchronously (`SCRIPT_PARSE`/`META_INVALID`) for a script that cannot begin; once a run is returned, its `result` NEVER rejects — every failure resolves with `stopReason: 'error'` (or `'cancelled'`). `dispose()` must reach quiescence within a bounded grace (cancel → wait → abandon), never hanging its caller.
|
||||
`start(request: WorkflowStartRequest): WorkflowRun` — parse and execute a script. Throws synchronously (`SCRIPT_PARSE`/`META_INVALID`) for a script that cannot begin; once a run is returned, its `result` NEVER rejects — every failure resolves with `stopReason: 'error'` (or `'cancelled'`). `dispose()` must reach quiescence within a bounded grace (cancel → wait for the script to settle and its children to finish disposing → abandon), never hanging its caller.
|
||||
|
||||
The protected `emitWorkflowEvent` helper dispatches the `workflow/*` events with PER-LISTENER containment (a throwing subscriber is logged, never propagated, and cannot starve later listeners) — the same guarantee as the subagent seam's lifecycle emits.
|
||||
The protected `emitWorkflowEvent` helper dispatches the `workflow/*` events with PER-LISTENER containment and PER-LISTENER payload snapshots (a throwing subscriber is logged, never propagated, and cannot starve later listeners; each subscriber gets its own clone of the payload, so mutating it corrupts neither the engine nor other listeners) — the same containment guarantee as the subagent seam's lifecycle emits.
|
||||
|
||||
## Vocabulary
|
||||
|
||||
|
||||
@@ -13,8 +13,10 @@
|
||||
* carry {@link WorkflowRunInfo} (id + meta), never the live {@link WorkflowRun}
|
||||
* — a listener must not gain `cancel`/`dispose`; control stays with the
|
||||
* `start()` caller holding the run. Every emit is per-listener contained (a
|
||||
* throwing subscriber is logged, never propagated), so one bad observer can
|
||||
* neither strand a live run nor starve later listeners.
|
||||
* throwing subscriber is logged, never propagated) and every listener gets its
|
||||
* own payload clone (mutating it corrupts nothing), so one bad observer can
|
||||
* neither strand a live run, starve later listeners, nor poison another
|
||||
* listener's view.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-workflow
|
||||
*/
|
||||
@@ -182,8 +184,9 @@ export function isFatalWorkflowError(error: unknown): boolean {
|
||||
* snapshots, per-listener containment); `workflow/end` fires exactly once
|
||||
* per started run, after `result` is settled or as it settles.
|
||||
* - `dispose()` reaches quiescence within a bounded grace: it cancels, waits
|
||||
* for the script to settle, and abandons a stuck script rather than
|
||||
* hanging its caller (the engine documents what abandonment leaves behind).
|
||||
* for the script to settle AND its started children to finish disposing,
|
||||
* and abandons whatever is left rather than hanging its caller (the engine
|
||||
* documents what abandonment leaves behind).
|
||||
*/
|
||||
export abstract class WorkflowService extends Service {
|
||||
constructor(ctx: Context) {
|
||||
@@ -199,12 +202,16 @@ export abstract class WorkflowService extends Service {
|
||||
abstract start(request: WorkflowStartRequest): WorkflowRun
|
||||
|
||||
/**
|
||||
* Emit one `workflow/*` lifecycle event with PER-LISTENER containment:
|
||||
* dispatch each subscriber individually and log (never propagate) a thrown
|
||||
* one, so one bad subscriber can neither fail the engine mid-run, surface as
|
||||
* an unhandled rejection on a detached settle hook, nor starve the listeners
|
||||
* registered after it (cordis `emit` halts on the first throw — same
|
||||
* guarantee as the subagent seam's lifecycle emits).
|
||||
* Emit one `workflow/*` lifecycle event with PER-LISTENER containment and
|
||||
* PER-LISTENER payload snapshots: each subscriber is dispatched individually
|
||||
* with its OWN structural clone of the payload (the payloads are plain JSON
|
||||
* data by the seam contract), so a listener mutating what it received can
|
||||
* corrupt neither the engine's live state nor any other listener's or later
|
||||
* event's view; a thrown listener is logged (never propagated), so one bad
|
||||
* subscriber can neither fail the engine mid-run, surface as an unhandled
|
||||
* rejection on a detached settle hook, nor starve the listeners registered
|
||||
* after it (cordis `emit` halts on the first throw — same guarantee as the
|
||||
* subagent seam's lifecycle emits).
|
||||
* @param name - the `workflow/*` event to dispatch.
|
||||
* @param args - the event's payload, matching its declared signature.
|
||||
*/
|
||||
@@ -213,7 +220,7 @@ export abstract class WorkflowService extends Service {
|
||||
try {
|
||||
// The declared workflow/* signatures are all void-returning emits; the
|
||||
// dispatch callback applies the payload tuple.
|
||||
;(callback as (...payload: unknown[]) => void)(...args)
|
||||
;(callback as (...payload: unknown[]) => void)(...structuredClone(args))
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`workflow: ${name} listener threw: ${String(error)}`)
|
||||
}
|
||||
|
||||
@@ -66,6 +66,28 @@ describe('dsh-workflow (interface)', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('gives each listener its OWN payload snapshot: mutation corrupts neither peers nor the caller', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(StubEngine)
|
||||
const seen: string[] = []
|
||||
ctx.on('workflow/agent-start', (info, agent) => {
|
||||
agent.label = 'HACKED'
|
||||
info.meta.name = 'HACKED'
|
||||
seen.push('mutator')
|
||||
})
|
||||
ctx.on('workflow/agent-start', (info, agent) => {
|
||||
seen.push(`${info.meta.name}/${agent.label}`)
|
||||
})
|
||||
const engine = ctx.workflows as StubEngine
|
||||
const info: WorkflowRunInfo = { id: WorkflowRunId('run-2'), meta: { name: 'w', description: 'd' } }
|
||||
const payload = { seq: 1, label: 'original', childId: 'c' }
|
||||
engine.emit('workflow/agent-start', info, payload)
|
||||
expect(seen).toEqual(['mutator', 'w/original'])
|
||||
// The caller's own objects are pristine too — no listener ever saw them.
|
||||
expect(info.meta.name).toBe('w')
|
||||
expect(payload.label).toBe('original')
|
||||
})
|
||||
|
||||
it('contains a throwing listener PER LISTENER: later listeners still run, nothing propagates', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(StubEngine)
|
||||
|
||||
Reference in New Issue
Block a user