From 7d8224c008ee9148d27cb089d6a92fd6a1975b30 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 17:02:41 -0700 Subject: [PATCH 1/4] feat(workflow): add ParallelWorker and JoinNode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 4/9 of the feature/workflows split, stacked on the built-in nodes. - nodes/parallel_worker: runs a wrapped node once per item of a list input, order-preserving, bounded by maxParallelWorkers, cancelling on first error (a non-list input is treated as a single-element list). Registers a factory with the engine (registerParallelWorkerFactory) so buildNode(..., {parallelWorker: true}) works without a static import. - nodes/join_node: a fan-in barrier that requires all predecessors and emits the aggregated predecessor outputs as its output. Tests (9): ParallelWorker mapping/order, single-item + empty-list handling, concurrency bounding, first-error propagation, the registry factory (buildNode + parallelWorker / maxParallelWorkers guard), and JoinNode passthrough — all driven directly against a NodeContext. The graph-level parallel/fan-in integration tests land in Part 6 with the runner. Full core suite green (2384 tests). --- core/src/workflow/nodes/join_node.ts | 34 +++++++ core/src/workflow/nodes/parallel_worker.ts | 113 +++++++++++++++++++++ core/test/workflow/parallel_worker_test.ts | 95 +++++++++++++++++ 3 files changed, 242 insertions(+) create mode 100644 core/src/workflow/nodes/join_node.ts create mode 100644 core/src/workflow/nodes/parallel_worker.ts create mode 100644 core/test/workflow/parallel_worker_test.ts diff --git a/core/src/workflow/nodes/join_node.ts b/core/src/workflow/nodes/join_node.ts new file mode 100644 index 000000000..23543d338 --- /dev/null +++ b/core/src/workflow/nodes/join_node.ts @@ -0,0 +1,34 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {createEvent, Event} from '../../events/event.js'; +import {BaseNode} from '../base_node.js'; +import {NodeContext} from '../node_context.js'; + +/** + * A fan-in barrier node: it waits for ALL of its predecessors to complete, then + * emits the aggregated inputs (a map of predecessor name → output) as its + * output. + * + * Ported from `google/adk-python` `workflow/_join_node.py`. + */ +export class JoinNode extends BaseNode { + override get requiresAllPredecessors(): boolean { + return true; + } + + protected async *runImpl( + ctx: NodeContext, + input: unknown, + ): AsyncGenerator { + yield createEvent({ + author: this.name, + invocationId: ctx.invocationId, + branch: ctx.branch, + output: input, + }); + } +} diff --git a/core/src/workflow/nodes/parallel_worker.ts b/core/src/workflow/nodes/parallel_worker.ts new file mode 100644 index 000000000..58f362543 --- /dev/null +++ b/core/src/workflow/nodes/parallel_worker.ts @@ -0,0 +1,113 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {BaseNode} from '../base_node.js'; +import {NodeContext} from '../node_context.js'; +import {RetryConfig} from '../retry_config.js'; +import {registerParallelWorkerFactory} from '../utils/workflow_graph_utils.js'; + +/** Options for a {@link ParallelWorker}. */ +export interface ParallelWorkerConfig { + /** Maximum number of items processed concurrently. `undefined` = unlimited. */ + maxParallelWorkers?: number; + retryConfig?: RetryConfig; + timeout?: number; +} + +/** + * A node that runs a wrapped node in parallel for each item of a list input, + * preserving order, bounded by `maxParallelWorkers`, cancelling on first error. + * + * Ported from `google/adk-python` `workflow/_parallel_worker.py`. A non-list + * input is treated as a single-element list. Each item runs via + * `ctx.runNode(inner, item, {useSubBranch: true})`; the node's output is the + * ordered list of the children's outputs. + */ +export class ParallelWorker extends BaseNode { + readonly maxParallelWorkers?: number; + private readonly inner: BaseNode; + + constructor(inner: BaseNode, config: ParallelWorkerConfig = {}) { + super({ + name: inner.name, + rerunOnResume: true, + retryConfig: config.retryConfig, + timeout: config.timeout, + }); + if ( + config.maxParallelWorkers !== undefined && + config.maxParallelWorkers < 1 + ) { + throw new Error('maxParallelWorkers must be greater than or equal to 1.'); + } + this.inner = inner; + this.maxParallelWorkers = config.maxParallelWorkers; + } + + protected async *runImpl( + ctx: NodeContext, + input: unknown, + ): AsyncGenerator { + const items = Array.isArray(input) ? input : [input]; + if (items.length === 0) { + yield []; + return; + } + + const results = new Array(items.length); + const poolSize = Math.min( + this.maxParallelWorkers ?? items.length, + items.length, + ); + + let nextIndex = 0; + let firstError: unknown; + + const worker = async (): Promise => { + for (;;) { + if (firstError !== undefined) { + return; + } + const i = nextIndex++; + if (i >= items.length) { + return; + } + try { + // Key each child run by its item index (not call order) so the + // run id -> item mapping is deterministic. On resume this lets each + // item fast-forward from its own cached run rather than being matched + // to a differently-ordered run id. + const child = await ctx.runNode(this.inner, items[i], { + useSubBranch: true, + runId: String(i), + }); + results[i] = child.output; + } catch (err) { + if (firstError === undefined) { + firstError = err; + } + return; + } + } + }; + + await Promise.all(Array.from({length: poolSize}, () => worker())); + + if (firstError !== undefined) { + throw firstError; + } + yield results; + } +} + +/** + * Registers the factory the engine uses to wrap a built node in a + * {@link ParallelWorker} when `buildNode(..., {parallelWorker: true})` is + * requested — keeping the engine core free of a static import of this module. + */ +registerParallelWorkerFactory( + (inner, options) => new ParallelWorker(inner, options), +); diff --git a/core/test/workflow/parallel_worker_test.ts b/core/test/workflow/parallel_worker_test.ts new file mode 100644 index 000000000..9b963ba25 --- /dev/null +++ b/core/test/workflow/parallel_worker_test.ts @@ -0,0 +1,95 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import {FunctionNode} from '../../src/workflow/nodes/function_node.js'; +import {JoinNode} from '../../src/workflow/nodes/join_node.js'; +import {ParallelWorker} from '../../src/workflow/nodes/parallel_worker.js'; +import {buildNode} from '../../src/workflow/utils/workflow_graph_utils.js'; +import {driveNode} from './test_helpers.js'; + +describe('ParallelWorker', () => { + it('maps a list input through the inner node, preserving order', async () => { + const inner = new FunctionNode('double', (_c, n: number) => n * 2); + const {output} = await driveNode(new ParallelWorker(inner), [1, 2, 3, 4]); + expect(output).toEqual([2, 4, 6, 8]); + }); + + it('treats a non-list input as a single-element list', async () => { + const inner = new FunctionNode('double', (_c, n: number) => n * 2); + const {output} = await driveNode(new ParallelWorker(inner), 5); + expect(output).toEqual([10]); + }); + + it('yields an empty list for an empty input', async () => { + const inner = new FunctionNode('id', (_c, x) => x); + const {output} = await driveNode(new ParallelWorker(inner), []); + expect(output).toEqual([]); + }); + + it('bounds concurrency by maxParallelWorkers', async () => { + let active = 0; + let peak = 0; + const inner = new FunctionNode('track', async (_c, n: number) => { + active++; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 5)); + active--; + return n; + }); + const {output} = await driveNode( + new ParallelWorker(inner, {maxParallelWorkers: 2}), + [1, 2, 3, 4, 5], + ); + expect(output).toEqual([1, 2, 3, 4, 5]); + expect(peak).toBeLessThanOrEqual(2); + }); + + it('rejects maxParallelWorkers < 1', () => { + const inner = new FunctionNode('x', (_c, v) => v); + expect(() => new ParallelWorker(inner, {maxParallelWorkers: 0})).toThrow( + /greater than or equal to 1/, + ); + }); + + it('propagates the first error from a failing item', async () => { + const inner = new FunctionNode('boom', (_c, n: number) => { + if (n === 3) { + throw new Error('boom at 3'); + } + return n; + }); + await expect( + driveNode(new ParallelWorker(inner), [1, 2, 3, 4]), + ).rejects.toThrow('boom at 3'); + }); +}); + +describe('ParallelWorker registry factory', () => { + it('buildNode wraps the built node when parallelWorker is requested', () => { + const node = buildNode((_c: unknown, n: number) => n, { + name: 'w', + parallelWorker: true, + }); + expect(node).toBeInstanceOf(ParallelWorker); + }); + + it('rejects maxParallelWorkers without parallelWorker', () => { + expect(() => + buildNode(() => {}, {name: 'x', maxParallelWorkers: 2}), + ).toThrow(/maxParallelWorkers can only be set/); + }); +}); + +describe('JoinNode', () => { + it('emits its aggregated input as output and requires all predecessors', async () => { + const join = new JoinNode({name: 'join'}); + const aggregated = {a: 1, b: 2}; + const {output} = await driveNode(join, aggregated); + expect(output).toEqual(aggregated); + expect(join.requiresAllPredecessors).toBe(true); + }); +}); From 651b65860559fd66b7116f41b4a9829755e515f2 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 14:06:35 -0700 Subject: [PATCH 2/4] refactor(workflow): wire ParallelWorker factory via the static const Follow the registry removal: PARALLEL_WORKER_FACTORY is set in node_builders.ts instead of parallel_worker.ts calling registerParallelWorkerFactory at import time. Update the engine-util test now that the factory is present (parallelWorker wraps in a ParallelWorker). --- core/src/workflow/node_builders.ts | 12 +++++++----- core/src/workflow/nodes/parallel_worker.ts | 12 +++--------- core/test/workflow/workflow_graph_utils_test.ts | 7 +++++-- 3 files changed, 15 insertions(+), 16 deletions(-) diff --git a/core/src/workflow/node_builders.ts b/core/src/workflow/node_builders.ts index 2680b5ba3..99e09981e 100644 --- a/core/src/workflow/node_builders.ts +++ b/core/src/workflow/node_builders.ts @@ -6,6 +6,7 @@ import {BaseTool, isBaseTool} from '../tools/base_tool.js'; import {FunctionNode, FunctionNodeHandler} from './nodes/function_node.js'; +import {ParallelWorker} from './nodes/parallel_worker.js'; import {ToolNode} from './nodes/tool_node.js'; import type { NodeBuilder, @@ -48,9 +49,10 @@ export const NODE_BUILDERS: readonly NodeBuilder[] = [ ]; /** - * Wraps an already-built node in a parallel worker. Wired in by the - * parallel-worker node part; `undefined` until then, so requesting - * `parallelWorker` before that part is present throws. + * Wraps an already-built node in a parallel worker, used by the engine for + * `buildNode(..., {parallelWorker: true})`. */ -export const PARALLEL_WORKER_FACTORY: ParallelWorkerFactory | undefined = - undefined; +export const PARALLEL_WORKER_FACTORY: ParallelWorkerFactory | undefined = ( + inner, + options, +) => new ParallelWorker(inner, options); diff --git a/core/src/workflow/nodes/parallel_worker.ts b/core/src/workflow/nodes/parallel_worker.ts index 58f362543..08b27ae44 100644 --- a/core/src/workflow/nodes/parallel_worker.ts +++ b/core/src/workflow/nodes/parallel_worker.ts @@ -7,7 +7,6 @@ import {BaseNode} from '../base_node.js'; import {NodeContext} from '../node_context.js'; import {RetryConfig} from '../retry_config.js'; -import {registerParallelWorkerFactory} from '../utils/workflow_graph_utils.js'; /** Options for a {@link ParallelWorker}. */ export interface ParallelWorkerConfig { @@ -103,11 +102,6 @@ export class ParallelWorker extends BaseNode { } } -/** - * Registers the factory the engine uses to wrap a built node in a - * {@link ParallelWorker} when `buildNode(..., {parallelWorker: true})` is - * requested — keeping the engine core free of a static import of this module. - */ -registerParallelWorkerFactory( - (inner, options) => new ParallelWorker(inner, options), -); +// The factory the engine uses to wrap a built node in a ParallelWorker (for +// `buildNode(..., {parallelWorker: true})`) is wired into PARALLEL_WORKER_FACTORY +// in ../node_builders.ts. diff --git a/core/test/workflow/workflow_graph_utils_test.ts b/core/test/workflow/workflow_graph_utils_test.ts index 79f48a287..e75771df2 100644 --- a/core/test/workflow/workflow_graph_utils_test.ts +++ b/core/test/workflow/workflow_graph_utils_test.ts @@ -6,6 +6,7 @@ import {describe, expect, it} from 'vitest'; import {BaseNode, START} from '../../src/workflow/base_node.js'; +import {ParallelWorker} from '../../src/workflow/nodes/parallel_worker.js'; import { buildNode, isNodeLike, @@ -57,8 +58,10 @@ describe('buildNode', () => { expect(() => buildNode(node, {maxParallelWorkers: 2})).toThrow(); }); - it('throws when parallelWorker is requested but unavailable', () => { + it('wraps a node in a ParallelWorker when requested', () => { const node = new FnNode('n', (_c, i) => i); - expect(() => buildNode(node, {parallelWorker: true})).toThrow(); + expect(buildNode(node, {parallelWorker: true})).toBeInstanceOf( + ParallelWorker, + ); }); }); From fb30901f9ba2316172328059073b6ecb769d7b89 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 14:28:41 -0700 Subject: [PATCH 3/4] =?UTF-8?q?fix(workflow):=20address=20Part=204=20revie?= =?UTF-8?q?w=20=E2=80=94=20ParallelWorker=20cancellation,=20bounds,=20retr?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ParallelWorker: - Don't apply retryConfig/timeout to the wrapper — they belong on the inner node (per item), so the two levels no longer compose. Dropped from ParallelWorkerConfig, the ParallelWorkerFactory options, and buildNode's factory call. - Bound default concurrency (DEFAULT_MAX_PARALLEL_WORKERS = 8) instead of unlimited; pass Infinity for unbounded. - Observe cancellation: the worker loop now stops claiming items when ctx.abortSignal or the invocation's abort signal fires (documented as stops-scheduling only — in-flight items still finish), and doesn't emit a partial list on abort. - Track failure with a dedicated `failed` flag so an item that rejects with `undefined` still fails instead of leaving a silent hole. - Give each child a distinct node path (overrideNodePath) so its events are attributable, not just a distinct branch/runId. - Doc: "stopping on first error" (nothing is cancelled), and state the all-or-nothing semantics explicitly. JoinNode: doc now says it emits its input unchanged (the engine supplies the predecessor-name -> output map); the barrier is enforced by the orchestrator via requiresAllPredecessors in a later part. Tests: pin the concurrency peak (toBe), add default-bound / undefined-reject / abort-stops-scheduling cases. --- core/src/workflow/nodes/join_node.ts | 11 ++- core/src/workflow/nodes/parallel_worker.ts | 72 +++++++++++++------ .../workflow/utils/workflow_graph_utils.ts | 13 ++-- core/test/workflow/parallel_worker_test.ts | 61 +++++++++++++++- 4 files changed, 125 insertions(+), 32 deletions(-) diff --git a/core/src/workflow/nodes/join_node.ts b/core/src/workflow/nodes/join_node.ts index 23543d338..ce4fbde1c 100644 --- a/core/src/workflow/nodes/join_node.ts +++ b/core/src/workflow/nodes/join_node.ts @@ -9,9 +9,14 @@ import {BaseNode} from '../base_node.js'; import {NodeContext} from '../node_context.js'; /** - * A fan-in barrier node: it waits for ALL of its predecessors to complete, then - * emits the aggregated inputs (a map of predecessor name → output) as its - * output. + * A fan-in barrier node: via {@link requiresAllPredecessors} the engine holds it + * until ALL of its predecessors complete, then runs it with their aggregated + * outputs as input. + * + * This node emits that input unchanged — the engine supplies the + * predecessor-name → output map; the join just passes it through as its output. + * The barrier itself is enforced by the orchestrator (which reads + * `requiresAllPredecessors`) and lands in a later part. * * Ported from `google/adk-python` `workflow/_join_node.py`. */ diff --git a/core/src/workflow/nodes/parallel_worker.ts b/core/src/workflow/nodes/parallel_worker.ts index 08b27ae44..e2a436615 100644 --- a/core/src/workflow/nodes/parallel_worker.ts +++ b/core/src/workflow/nodes/parallel_worker.ts @@ -6,36 +6,50 @@ import {BaseNode} from '../base_node.js'; import {NodeContext} from '../node_context.js'; -import {RetryConfig} from '../retry_config.js'; + +/** + * Default concurrency when `maxParallelWorkers` is not set. Bounded so a + * data-driven list length can't fan out into an unbounded burst of concurrent + * inner runs (a rate-limit / cost hazard when the inner node is an LLM or a + * remote tool). Pass `Infinity` for explicitly unbounded concurrency. + */ +const DEFAULT_MAX_PARALLEL_WORKERS = 8; /** Options for a {@link ParallelWorker}. */ export interface ParallelWorkerConfig { - /** Maximum number of items processed concurrently. `undefined` = unlimited. */ + /** + * Maximum number of items processed concurrently. Defaults to + * {@link DEFAULT_MAX_PARALLEL_WORKERS}; pass `Infinity` for unbounded. + */ maxParallelWorkers?: number; - retryConfig?: RetryConfig; - timeout?: number; } /** - * A node that runs a wrapped node in parallel for each item of a list input, - * preserving order, bounded by `maxParallelWorkers`, cancelling on first error. + * A node that runs a wrapped node once per item of a list input, preserving + * order, bounded by `maxParallelWorkers`, and stopping on the first error. * * Ported from `google/adk-python` `workflow/_parallel_worker.py`. A non-list * input is treated as a single-element list. Each item runs via * `ctx.runNode(inner, item, {useSubBranch: true})`; the node's output is the * ordered list of the children's outputs. + * + * Notes: + * - **retry/timeout live on the inner node.** `retryConfig`/`timeout` passed to + * `buildNode` apply to the wrapped node (per item); the ParallelWorker itself + * carries neither, so the two levels don't compose. + * - **All-or-nothing.** If any item throws, the first error is rethrown and the + * already-computed sibling outputs are discarded. Make individual items + * failure-tolerant if partial results matter. + * - **Cancellation stops scheduling only.** On abort/timeout the loop stops + * claiming new items, but items already in flight run to completion — + * `ctx.runNode` has no way to forward a signal into a child run. */ export class ParallelWorker extends BaseNode { readonly maxParallelWorkers?: number; private readonly inner: BaseNode; constructor(inner: BaseNode, config: ParallelWorkerConfig = {}) { - super({ - name: inner.name, - rerunOnResume: true, - retryConfig: config.retryConfig, - timeout: config.timeout, - }); + super({name: inner.name, rerunOnResume: true}); if ( config.maxParallelWorkers !== undefined && config.maxParallelWorkers < 1 @@ -58,16 +72,26 @@ export class ParallelWorker extends BaseNode { const results = new Array(items.length); const poolSize = Math.min( - this.maxParallelWorkers ?? items.length, + this.maxParallelWorkers ?? DEFAULT_MAX_PARALLEL_WORKERS, items.length, ); let nextIndex = 0; + // Separate flag from `firstError` so an item that rejects with `undefined` + // (a bare `Promise.reject()`) still counts as a failure instead of leaving a + // silent hole in `results` and resolving successfully. + let failed = false; let firstError: unknown; + // Populated only when the ParallelWorker itself declares a timeout; on a + // plain invocation abort the invocation-level signal is the one that fires. + const isAborted = (): boolean => + ctx.abortSignal?.aborted === true || + ctx.invocationContext.abortSignal?.aborted === true; + const worker = async (): Promise => { for (;;) { - if (firstError !== undefined) { + if (failed || isAborted()) { return; } const i = nextIndex++; @@ -75,17 +99,20 @@ export class ParallelWorker extends BaseNode { return; } try { - // Key each child run by its item index (not call order) so the - // run id -> item mapping is deterministic. On resume this lets each - // item fast-forward from its own cached run rather than being matched - // to a differently-ordered run id. + // Key each child by its item index (not completion order): the runId + // makes the run deterministic, and the distinct node path makes each + // child's events attributable (they'd otherwise all share the inner + // node's path). The scheduler uses the same runId to fast-forward each + // item on resume (lands with the scheduler in a later part). const child = await ctx.runNode(this.inner, items[i], { useSubBranch: true, runId: String(i), + overrideNodePath: `${ctx.nodePath}.${this.inner.name}@${i}`, }); results[i] = child.output; } catch (err) { - if (firstError === undefined) { + if (!failed) { + failed = true; firstError = err; } return; @@ -95,9 +122,14 @@ export class ParallelWorker extends BaseNode { await Promise.all(Array.from({length: poolSize}, () => worker())); - if (firstError !== undefined) { + if (failed) { throw firstError; } + if (isAborted()) { + // Aborted mid-flight: `results` may have holes for unscheduled items, so + // don't emit a wrong partial list — the invocation is being torn down. + return; + } yield results; } } diff --git a/core/src/workflow/utils/workflow_graph_utils.ts b/core/src/workflow/utils/workflow_graph_utils.ts index 50797b69f..f1b80e00c 100644 --- a/core/src/workflow/utils/workflow_graph_utils.ts +++ b/core/src/workflow/utils/workflow_graph_utils.ts @@ -50,14 +50,13 @@ export interface NodeBuilder { /** * Wraps an already-built node in a parallel worker. Provided by the * `parallel_worker` node module via {@link PARALLEL_WORKER_FACTORY}. + * + * `retryConfig`/`timeout` are intentionally not forwarded to the wrapper: they + * apply to the inner node (per item), so the two levels don't compose. */ export type ParallelWorkerFactory = ( inner: BaseNode, - options: { - maxParallelWorkers?: number; - retryConfig?: RetryConfig; - timeout?: number; - }, + options: {maxParallelWorkers?: number}, ) => BaseNode; /** @@ -115,10 +114,10 @@ export function buildNode( 'the parallel worker node module is not part of this build.', ); } + // retryConfig/timeout are applied to the inner (built) node, not the + // wrapper, so they aren't forwarded here (see ParallelWorkerFactory). return PARALLEL_WORKER_FACTORY(built, { maxParallelWorkers: options.maxParallelWorkers, - retryConfig: options.retryConfig, - timeout: options.timeout, }); } return built; diff --git a/core/test/workflow/parallel_worker_test.ts b/core/test/workflow/parallel_worker_test.ts index 9b963ba25..0efcab96f 100644 --- a/core/test/workflow/parallel_worker_test.ts +++ b/core/test/workflow/parallel_worker_test.ts @@ -9,7 +9,7 @@ import {FunctionNode} from '../../src/workflow/nodes/function_node.js'; import {JoinNode} from '../../src/workflow/nodes/join_node.js'; import {ParallelWorker} from '../../src/workflow/nodes/parallel_worker.js'; import {buildNode} from '../../src/workflow/utils/workflow_graph_utils.js'; -import {driveNode} from './test_helpers.js'; +import {createIc, driveNode} from './test_helpers.js'; describe('ParallelWorker', () => { it('maps a list input through the inner node, preserving order', async () => { @@ -45,7 +45,28 @@ describe('ParallelWorker', () => { [1, 2, 3, 4, 5], ); expect(output).toEqual([1, 2, 3, 4, 5]); - expect(peak).toBeLessThanOrEqual(2); + // Pin both halves: never more than 2, and it actually reached 2 (this would + // stay green at peak=1 if the pool regressed to running items serially). + expect(peak).toBe(2); + }); + + it('bounds concurrency by the default when maxParallelWorkers is unset', async () => { + let active = 0; + let peak = 0; + const inner = new FunctionNode('track', async (_c, n: number) => { + active++; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 5)); + active--; + return n; + }); + // 20 items with no explicit limit must not fan out to 20 concurrent runs. + const {output} = await driveNode( + new ParallelWorker(inner), + Array.from({length: 20}, (_v, i) => i), + ); + expect(output).toHaveLength(20); + expect(peak).toBe(8); // DEFAULT_MAX_PARALLEL_WORKERS }); it('rejects maxParallelWorkers < 1', () => { @@ -66,6 +87,42 @@ describe('ParallelWorker', () => { driveNode(new ParallelWorker(inner), [1, 2, 3, 4]), ).rejects.toThrow('boom at 3'); }); + + it('fails (not silently) when an item rejects with undefined', async () => { + const inner = new FunctionNode('bad', (_c, n: number) => { + if (n === 2) { + throw undefined; // bare reject: must still count as a failure + } + return n; + }); + let rejected = false; + try { + await driveNode(new ParallelWorker(inner), [1, 2, 3]); + } catch { + rejected = true; + } + expect(rejected).toBe(true); + }); + + it('stops scheduling items once the invocation is aborted', async () => { + let calls = 0; + const inner = new FunctionNode('count', (_c, n: number) => { + calls++; + return n; + }); + const controller = new AbortController(); + controller.abort(); // aborted before the run starts + const ic = createIc({}, controller.signal); + + const {output} = await driveNode( + new ParallelWorker(inner), + [1, 2, 3, 4, 5], + ic, + ); + // No item was scheduled, and no wrong partial list was emitted. + expect(calls).toBe(0); + expect(output).toBeUndefined(); + }); }); describe('ParallelWorker registry factory', () => { From 3fc58ec2c2516a38a1c259b4e6ac9abd1c576a8e Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 14:33:01 -0700 Subject: [PATCH 4/4] refactor(workflow): use a condition loop in ParallelWorker's worker pool Replace the infinite worker loop with while(!failed && !isAborted()) so the termination conditions live in the loop header instead of an infinite loop with internal breaks. --- core/src/workflow/nodes/parallel_worker.ts | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/core/src/workflow/nodes/parallel_worker.ts b/core/src/workflow/nodes/parallel_worker.ts index e2a436615..453a41041 100644 --- a/core/src/workflow/nodes/parallel_worker.ts +++ b/core/src/workflow/nodes/parallel_worker.ts @@ -89,14 +89,13 @@ export class ParallelWorker extends BaseNode { ctx.abortSignal?.aborted === true || ctx.invocationContext.abortSignal?.aborted === true; + // Keep claiming the next item until the list is exhausted, an item fails, + // or the invocation is aborted. const worker = async (): Promise => { - for (;;) { - if (failed || isAborted()) { - return; - } + while (!failed && !isAborted()) { const i = nextIndex++; if (i >= items.length) { - return; + break; } try { // Key each child by its item index (not completion order): the runId @@ -115,7 +114,7 @@ export class ParallelWorker extends BaseNode { failed = true; firstError = err; } - return; + break; } } };