From 33d9a5c2b8e3e10bcf455592f22c9116938efe75 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 13:00:32 -0700 Subject: [PATCH 1/5] feat(workflow): add event model extensions and shared node primitives First slice of the workflow engine split (Part 1/9). Adds the leaf primitives the engine builds on, none of which depend on the engine core: - workflow/errors, node_status, node_state, retry_config, utils/retry_utils, trigger, branch_path, and utils/event_channel - Event model extensions mirroring adk-python: Event.{output, route, nodeInfo, isolationScope}, the NodeInfo type, CreateEventParams, an isEvent guard, and additive optional EventActions fields; PRESERVE_KEYS entries so node output and checkpointed state survive snake/camel round-trips - export the new event types from common.ts Bundled tests: foundations, event_channel, event_model, and event. Recomposed from the final feature/workflows tree onto current main (3-way merged event.ts to preserve main's relocated function-call-id helpers). Stacked PRs follow. --- core/src/common.ts | 2 +- core/src/events/event.ts | 85 ++++++++++- core/src/events/event_actions.ts | 26 ++++ core/src/workflow/branch_path.ts | 92 ++++++++++++ core/src/workflow/errors.ts | 78 ++++++++++ core/src/workflow/node_state.ts | 77 ++++++++++ core/src/workflow/node_status.ts | 29 ++++ core/src/workflow/retry_config.ts | 85 +++++++++++ core/src/workflow/trigger.ts | 26 ++++ core/src/workflow/utils/event_channel.ts | 108 ++++++++++++++ core/src/workflow/utils/retry_utils.ts | 112 ++++++++++++++ core/test/events/event_test.ts | 45 ++++++ core/test/workflow/event_channel_test.ts | 90 ++++++++++++ core/test/workflow/event_model_test.ts | 66 +++++++++ core/test/workflow/foundations_test.ts | 179 +++++++++++++++++++++++ 15 files changed, 1097 insertions(+), 3 deletions(-) create mode 100644 core/src/workflow/branch_path.ts create mode 100644 core/src/workflow/errors.ts create mode 100644 core/src/workflow/node_state.ts create mode 100644 core/src/workflow/node_status.ts create mode 100644 core/src/workflow/retry_config.ts create mode 100644 core/src/workflow/trigger.ts create mode 100644 core/src/workflow/utils/event_channel.ts create mode 100644 core/src/workflow/utils/retry_utils.ts create mode 100644 core/test/workflow/event_channel_test.ts create mode 100644 core/test/workflow/event_model_test.ts create mode 100644 core/test/workflow/foundations_test.ts diff --git a/core/src/common.ts b/core/src/common.ts index 23f628165..e75aa1881 100644 --- a/core/src/common.ts +++ b/core/src/common.ts @@ -143,7 +143,7 @@ export { pruneThoughts, stringifyContent, } from './events/event.js'; -export type {Event} from './events/event.js'; +export type {CreateEventParams, Event, NodeInfo} from './events/event.js'; export {createEventActions} from './events/event_actions.js'; export type {EventActions} from './events/event_actions.js'; export {EventType, toStructuredEvents} from './events/structured_events.js'; diff --git a/core/src/events/event.ts b/core/src/events/event.ts index 92387d8b1..313477853 100644 --- a/core/src/events/event.ts +++ b/core/src/events/event.ts @@ -12,6 +12,26 @@ import {randomUUID} from '../utils/env_aware_utils.js'; import {toCamelCase, toSnakeCase} from '../utils/object_notation_utils.js'; import {createEventActions, EventActions} from './event_actions.js'; +/** + * Workflow-node provenance attached to an event. + * + * Mirrors `google/adk-python` `Event.node_info`. Present only on events emitted + * from within a workflow node. + */ +export interface NodeInfo { + /** The workflow node path that produced this event (e.g. `wf.child.0`). */ + path?: string; + + /** The node run id this event's output should be attributed to. */ + outputFor?: string; + + /** + * Whether the event's textual content should be promoted to the node's + * structured output. + */ + messageAsOutput?: boolean; +} + /** * Represents an event in a conversation between agents and users. @@ -63,6 +83,41 @@ export interface Event extends LlmResponse { * The timestamp of the event. */ timestamp: number; + + /** + * Workflow: the structured output produced by the emitting node, if any. + * + * First-class field mirroring `google/adk-python` `Event.output`. Used by the + * workflow engine to carry a node's return value alongside its content. + */ + output?: unknown; + + /** + * Workflow: the route key(s) emitted by a routing node, used by the graph to + * select the matching outgoing edge(s). A single value fires one branch; an + * array fires every branch whose route matches any listed value (multi-route + * dispatch). Mirrors Python `Event.route`. + */ + route?: string | number | boolean | Array; + + /** + * Workflow: provenance of the emitting node. Mirrors Python `Event.node_info`. + */ + nodeInfo?: NodeInfo; + + /** + * Workflow: scope tag used to isolate multi-agent conversations so peer + * scopes don't see each other's events. Mirrors Python + * `Event.isolation_scope`. + */ + isolationScope?: string; +} + +/** + * Parameters for creating an event with partial fields. + */ +export interface CreateEventParams extends Omit, 'actions'> { + actions?: Partial; } /** @@ -71,13 +126,13 @@ export interface Event extends LlmResponse { * @param params The partial event to create the event from. * @returns The event. */ -export function createEvent(params: Partial = {}): Event { +export function createEvent(params: CreateEventParams = {}): Event { return { ...params, id: params.id || createNewEventId(), invocationId: params.invocationId || '', author: params.author, - actions: params.actions || createEventActions(), + actions: createEventActions(params.actions), longRunningToolIds: params.longRunningToolIds || [], branch: params.branch, timestamp: params.timestamp || Date.now(), @@ -229,6 +284,23 @@ export function pruneThoughts(event: Event): Event { const ASCII_LETTERS_AND_NUMBERS = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'; +/** + * Type guard to check if an object is an instance of Event. + * + * @param obj The object to check. + * @returns True if the object matches the Event structure. + */ +export function isEvent(obj: unknown): obj is Event { + return ( + typeof obj === 'object' && + obj !== null && + 'invocationId' in obj && + typeof (obj as Event).invocationId === 'string' && + 'actions' in obj && + typeof (obj as Event).actions === 'object' + ); +} + /** * Generates a new unique ID for the event. */ @@ -262,6 +334,11 @@ const PRESERVE_KEYS_CAMEL_CASE = [ 'customMetadata', 'content.parts.functionCall.args', 'content.parts.functionResponse.response', + // Workflow: arbitrary node output and checkpointed node state carry + // user-defined keys that must survive round-trips verbatim (a node's original + // input is stashed under `actions.agentState` for HITL resume). + 'output', + 'actions.agentState', ]; /** @@ -281,6 +358,10 @@ const PRESERVE_KEYS_SNAKE_CASE = [ 'custom_metadata', 'content.parts.function_call.args', 'content.parts.function_response.response', + // Workflow: arbitrary node output and checkpointed node state (see the + // camelCase list above). + 'output', + 'actions.agent_state', ]; /** diff --git a/core/src/events/event_actions.ts b/core/src/events/event_actions.ts index 0316ec290..54c89fa71 100644 --- a/core/src/events/event_actions.ts +++ b/core/src/events/event_actions.ts @@ -56,6 +56,32 @@ export interface EventActions { * call id. */ requestedToolConfirmations: {[key: string]: ToolConfirmation}; + + /** Workflow / custom event actions */ + output?: unknown; + joinCompleted?: unknown; + toolExecution?: unknown; + requestInput?: unknown; + nodeExecutionReplay?: unknown; + + /** + * Workflow: a serialized node/agent state snapshot used for resumable + * checkpointing. Mirrors Python `EventActions.agent_state`. + */ + agentState?: Record; + + /** + * Workflow: marks that the emitting agent/workflow has reached the end of its + * execution for this invocation. Mirrors Python `EventActions.end_of_agent`. + */ + endOfAgent?: boolean; + + /** + * Workflow: route key(s) selected by a routing node (alternative carrier to + * the top-level `Event.route`, used by callbacks/tools). A single value or an + * array for multi-route dispatch. + */ + route?: string | number | boolean | Array; } /** diff --git a/core/src/workflow/branch_path.ts b/core/src/workflow/branch_path.ts new file mode 100644 index 000000000..63fd58395 --- /dev/null +++ b/core/src/workflow/branch_path.ts @@ -0,0 +1,92 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Hierarchical, dot-separated path for execution branches. Each segment is a + * node run, typically formatted as `name@runId` (or just `name`). + * + * Ported from `google/adk-python` `events/_branch_path.py::_BranchPath`. + * + * @example 'parent_agent@1.collect_tool@2.sub_workflow' + */ +export class BranchPath { + private readonly segments: string[]; + + constructor(segments: string[]) { + this.segments = [...segments]; + } + + /** Parses a dot-separated string into a {@link BranchPath}. */ + static fromString(path?: string | null): BranchPath { + if (!path) { + return new BranchPath([]); + } + return new BranchPath(path.split('.')); + } + + toString(): string { + return this.segments.join('.'); + } + + /** Returns a copy of the path segments. */ + getSegments(): string[] { + return [...this.segments]; + } + + /** Whether this path is a strict descendant of `ancestor`. */ + isDescendantOf(ancestor: BranchPath): boolean { + if (this.segments.length <= ancestor.segments.length) { + return false; + } + return ancestor.segments.every((seg, i) => this.segments[i] === seg); + } + + /** Returns a new path with a `name@runId` (or `name`) segment appended. */ + append(name: string, runId?: string): BranchPath { + const segment = runId !== undefined ? `${name}@${runId}` : name; + return new BranchPath([...this.segments, segment]); + } + + /** Finds the common prefix across a list of paths. */ + static commonPrefix(paths: BranchPath[]): BranchPath { + if (paths.length === 0) { + return new BranchPath([]); + } + const common: string[] = []; + const minLen = Math.min(...paths.map((p) => p.segments.length)); + for (let i = 0; i < minLen; i++) { + const seg = paths[0].segments[i]; + if (paths.every((p) => p.segments[i] === seg)) { + common.push(seg); + } else { + break; + } + } + return new BranchPath(common); + } + + /** + * Creates a new dot-separated sub-branch string by appending a segment. + * + * @example createSubBranch('parent', {name: 'child', runId: '1'}) -> 'parent.child@1' + * @example createSubBranch(undefined, {name: 'agent'}) -> 'agent' + */ + static createSubBranch( + baseBranch: string | undefined | null, + options: {name: string; runId?: string}, + ): string { + return BranchPath.fromString(baseBranch) + .append(options.name, options.runId) + .toString(); + } + + /** Finds the common prefix of a list of dot-separated branch strings. */ + static commonPrefixOf(branches: string[]): string { + return BranchPath.commonPrefix( + branches.map((b) => BranchPath.fromString(b)), + ).toString(); + } +} diff --git a/core/src/workflow/errors.ts b/core/src/workflow/errors.ts new file mode 100644 index 000000000..14c73a44a --- /dev/null +++ b/core/src/workflow/errors.ts @@ -0,0 +1,78 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Errors raised by the workflow framework. + * + * Ported from `google/adk-python` `workflow/_errors.py`. + */ + +/** + * Internal: raised when a dynamic node interrupts (HITL). + * + * Used exclusively by `ctx.runNode()` to signal that the dynamic child has + * unresolved interrupt IDs. The parent's NodeRunner catches this and reads the + * interrupt IDs from the parent's ctx (set by `ctx.runNode()` before throwing). + * + * Internal to the framework — not part of the public API. + */ +export class NodeInterruptedError extends Error { + constructor(message = 'Node interrupted (awaiting resume input).') { + super(message); + this.name = 'NodeInterruptedError'; + // Restore prototype chain for `instanceof` across transpilation targets. + Object.setPrototypeOf(this, NodeInterruptedError.prototype); + } +} + +/** + * Raised when a node exceeds its configured timeout. + * + * This is a regular `Error` (retryable) so a timed-out node can be retried via + * `retryConfig`. + */ +export class NodeTimeoutError extends Error { + readonly nodeName: string; + readonly timeout: number; + + /** + * @param options.nodeName The name of the node that timed out. + * @param options.timeout The timeout, in seconds, that was exceeded. + */ + constructor(options: {nodeName: string; timeout: number}) { + super( + `Node '${options.nodeName}' timed out after ${options.timeout} seconds.`, + ); + this.name = 'NodeTimeoutError'; + this.nodeName = options.nodeName; + this.timeout = options.timeout; + Object.setPrototypeOf(this, NodeTimeoutError.prototype); + } +} + +/** + * Raised when a dynamic node fails. + * + * Caught by the parent node's NodeRunner to propagate the error. + * Internal to the framework — not part of the public API. + */ +export class DynamicNodeFailError extends Error { + readonly error: Error; + readonly errorNodePath: string; + + /** + * @param options.message Human-readable failure message. + * @param options.error The underlying error thrown by the dynamic node. + * @param options.errorNodePath The node path where the failure occurred. + */ + constructor(options: {message: string; error: Error; errorNodePath: string}) { + super(options.message); + this.name = 'DynamicNodeFailError'; + this.error = options.error; + this.errorNodePath = options.errorNodePath; + Object.setPrototypeOf(this, DynamicNodeFailError.prototype); + } +} diff --git a/core/src/workflow/node_state.ts b/core/src/workflow/node_state.ts new file mode 100644 index 000000000..8950bbaa5 --- /dev/null +++ b/core/src/workflow/node_state.ts @@ -0,0 +1,77 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {NodeStatus} from './node_status.js'; + +/** + * State of a node in the workflow. + * + * Ported from `google/adk-python` `workflow/_node_state.py`. Note that the + * node's *output* is intentionally NOT stored here — it is carried on emitted + * events / the node Context, not on the persisted node state. + */ +export interface NodeState { + /** The run status of the node. */ + status: NodeStatus; + + /** The input provided to the node. */ + input?: unknown; + + /** The attempt count for this node run (1-based). */ + attemptCount: number; + + /** The interrupt ids that are pending to be resolved. */ + interrupts: string[]; + + /** The responses for resuming the node, keyed by interrupt id. */ + resumeInputs: Record; + + /** + * Sequential counter incremented each time the node gets a fresh run. + * + * Preserving this count independently of `runId` prevents path collisions if + * a node switches between custom string IDs and auto-generated numeric IDs. + */ + runCounter: number; + + /** The run ID of this node run. */ + runId?: string; + + /** + * The run ID of the parent node which dynamically scheduled this node run. + */ + parentRunId?: string; +} + +/** + * Creates a {@link NodeState} with Python-aligned defaults, overlaying any + * provided partial values. + */ +export function createNodeState(partial?: Partial): NodeState { + return { + status: NodeStatus.INACTIVE, + attemptCount: 1, + interrupts: [], + resumeInputs: {}, + runCounter: 0, + ...partial, + }; +} + +/** + * Type guard for a {@link NodeState}-shaped object. + */ +export function isNodeState(obj: unknown): obj is NodeState { + return ( + typeof obj === 'object' && + obj !== null && + 'status' in obj && + typeof (obj as NodeState).status === 'number' && + 'attemptCount' in obj && + 'interrupts' in obj && + Array.isArray((obj as NodeState).interrupts) + ); +} diff --git a/core/src/workflow/node_status.ts b/core/src/workflow/node_status.ts new file mode 100644 index 000000000..0a6cec0eb --- /dev/null +++ b/core/src/workflow/node_status.ts @@ -0,0 +1,29 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * The status of a node in the workflow graph. + * + * Numeric values are aligned with `google/adk-python` + * `workflow/_node_status.py` so that persisted node state is portable across + * the Python and TypeScript runtimes. + */ +export enum NodeStatus { + /** The node is not ready to be executed. */ + INACTIVE = 0, + /** The node is ready to be executed. */ + PENDING = 1, + /** The node is being executed. */ + RUNNING = 2, + /** The node has been executed successfully. */ + COMPLETED = 3, + /** The node is waiting (e.g. for a user response or re-trigger). */ + WAITING = 4, + /** The node has failed. */ + FAILED = 5, + /** The node has been cancelled. */ + CANCELLED = 6, +} diff --git a/core/src/workflow/retry_config.ts b/core/src/workflow/retry_config.ts new file mode 100644 index 000000000..65ea24efc --- /dev/null +++ b/core/src/workflow/retry_config.ts @@ -0,0 +1,85 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * An error constructor usable in {@link RetryConfig.exceptions}. + */ +export type ErrorClass = new (...args: never[]) => Error; + +/** + * Configuration for retrying a node. + * + * Ported from `google/adk-python` `workflow/_retry_config.py`. Delays are + * expressed in **seconds** (fractions allowed) to match the Python semantics + * and keep configuration portable across runtimes. Unset fields fall back to + * the documented defaults inside the retry utilities. + */ +export interface RetryConfig { + /** + * Maximum number of attempts, including the original request. If 0 or 1, it + * means no retries. If not specified, defaults to 5. + */ + maxAttempts?: number; + + /** + * Initial delay before the first retry, in seconds. If not specified, + * defaults to 1.0 second. + */ + initialDelay?: number; + + /** + * Maximum delay between retries, in seconds. If not specified, defaults to + * 60.0 seconds. + */ + maxDelay?: number; + + /** + * Multiplier by which the delay increases after each attempt. If not + * specified, defaults to 2.0. + */ + backoffFactor?: number; + + /** + * Randomness factor for the delay. If not specified, defaults to 1.0. Use 0.0 + * to remove randomness. + */ + jitter?: number; + + /** + * Exceptions to retry on. Accepts error class names as strings (e.g. + * `['TypeError']`) or error classes directly (e.g. `[TypeError]`). + * `undefined`/`null` means retry on all errors. + */ + exceptions?: Array | null; +} + +/** + * Normalizes the `exceptions` field of a {@link RetryConfig} to a list of error + * class name strings, mirroring Python's `field_validator`. + * + * @returns The list of class-name strings, or `undefined` to mean "retry on all + * errors". + */ +export function normalizeRetryExceptions( + exceptions?: Array | null, +): string[] | undefined { + if (exceptions === undefined || exceptions === null) { + return undefined; + } + return exceptions.map((item) => { + if (typeof item === 'string') { + return item; + } + if (typeof item === 'function' && item.name) { + return item.name; + } + throw new Error( + `exceptions must contain error class names (string) or error classes, got: ${String( + item, + )}`, + ); + }); +} diff --git a/core/src/workflow/trigger.ts b/core/src/workflow/trigger.ts new file mode 100644 index 000000000..9ce2f307d --- /dev/null +++ b/core/src/workflow/trigger.ts @@ -0,0 +1,26 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * A buffered trigger for a downstream node. + * + * Ported from `google/adk-python` `workflow/_trigger.py`. Unlike the previous + * TypeScript `Trigger` (a route-matching predicate), this is a plain data record + * describing *how* a target node should be invoked when its turn comes. + */ +export interface Trigger { + /** The input to pass to the triggered node. */ + input?: unknown; + + /** Whether this trigger should run the node in an isolated sub-branch. */ + useSubBranch?: boolean; + + /** The branch inherited from the predecessor node. */ + branch?: string; + + /** Scope tag explicitly propagated to this trigger. */ + isolationScope?: string; +} diff --git a/core/src/workflow/utils/event_channel.ts b/core/src/workflow/utils/event_channel.ts new file mode 100644 index 000000000..48914a85d --- /dev/null +++ b/core/src/workflow/utils/event_channel.ts @@ -0,0 +1,108 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * A single-consumer async queue that bridges the workflow engine's *push* model + * (nodes/`ctx.runNode()` push events as they run) to the runtime's *pull* model + * (the workflow's outer async generator drains and re-yields them). + * + * Producers call {@link push} as events are produced and {@link close} (or + * {@link fail}) when finished. A single consumer drains the channel by + * `for await (const ev of channel)`. + * + * Semantics: + * - Buffered items are always delivered before an end/error signal. + * - {@link close} ends iteration cleanly (`done: true`). + * - {@link fail} surfaces the error to the consumer *after* any buffered items + * have been drained. + * - {@link push} after close/fail is ignored (the producer has already + * signalled completion). + */ +export class EventChannel implements AsyncIterable { + private readonly buffer: T[] = []; + private readonly waiters: Array<{ + resolve: (r: IteratorResult) => void; + reject: (e: unknown) => void; + }> = []; + private closed = false; + private failure?: {error: unknown}; + + /** Whether the channel has been closed or failed. */ + get isClosed(): boolean { + return this.closed; + } + + /** Number of items buffered and not yet consumed. */ + get size(): number { + return this.buffer.length; + } + + /** + * Enqueues an item. If a consumer is currently awaiting, it is resolved + * immediately; otherwise the item is buffered. No-op once closed/failed. + */ + push(item: T): void { + if (this.closed) { + return; + } + const waiter = this.waiters.shift(); + if (waiter) { + waiter.resolve({value: item, done: false}); + } else { + this.buffer.push(item); + } + } + + /** + * Signals that no more items will be produced. Any awaiting consumer receives + * `{done: true}`. Idempotent. + */ + close(): void { + if (this.closed) { + return; + } + this.closed = true; + while (this.waiters.length > 0) { + this.waiters.shift()!.resolve({value: undefined as never, done: true}); + } + } + + /** + * Signals that production failed. Buffered items are still delivered first; + * once the buffer drains, the consumer's next `next()` rejects with `error`. + * Idempotent (first failure wins). + */ + fail(error: unknown): void { + if (this.closed) { + return; + } + this.failure = {error}; + this.closed = true; + // If a consumer is awaiting, the buffer is empty, so surface the error now. + while (this.waiters.length > 0) { + this.waiters.shift()!.reject(error); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: (): Promise> => { + if (this.buffer.length > 0) { + return Promise.resolve({value: this.buffer.shift()!, done: false}); + } + if (this.failure) { + return Promise.reject(this.failure.error); + } + if (this.closed) { + return Promise.resolve({value: undefined as never, done: true}); + } + return new Promise>((resolve, reject) => { + this.waiters.push({resolve, reject}); + }); + }, + }; + } +} diff --git a/core/src/workflow/utils/retry_utils.ts b/core/src/workflow/utils/retry_utils.ts new file mode 100644 index 000000000..57680e801 --- /dev/null +++ b/core/src/workflow/utils/retry_utils.ts @@ -0,0 +1,112 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Utility functions for retrying nodes in a workflow. + * + * Ported from `google/adk-python` `workflow/utils/_retry_utils.py`. + */ + +import {NodeState} from '../node_state.js'; +import {RetryConfig, normalizeRetryExceptions} from '../retry_config.js'; + +const DEFAULT_MAX_ATTEMPTS = 5; +const DEFAULT_INITIAL_DELAY_SECONDS = 1.0; +const DEFAULT_MAX_DELAY_SECONDS = 60.0; +const DEFAULT_BACKOFF_FACTOR = 2.0; +const DEFAULT_JITTER = 1.0; + +/** + * Resolves the runtime name of a thrown value for exception-name matching. + * Mirrors Python's `type(exception).__name__`. + */ +function errorName(error: unknown): string { + if (error instanceof Error) { + // `name` is set by well-behaved Error subclasses; fall back to the + // constructor name for plain `throw new Error()` cases. + return error.name || error.constructor.name; + } + if (typeof error === 'object' && error !== null) { + return error.constructor.name; + } + return typeof error; +} + +/** + * Checks if a failed node should be retried based on its retry config. + * + * @param error The error thrown by the node. + * @param retryConfig The node's retry configuration, if any. + * @param nodeState The current node state (its `attemptCount` is 1-based). + */ +export function shouldRetryNode( + error: unknown, + retryConfig: RetryConfig | undefined, + nodeState: NodeState, +): boolean { + if (!retryConfig) { + return false; + } + + const attemptCount = nodeState.attemptCount; + const maxAttempts = retryConfig.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; + + // attemptCount starts at 1 for the original request; once it reaches + // maxAttempts, the limit is exhausted. + if (attemptCount >= maxAttempts) { + return false; + } + + const exceptions = normalizeRetryExceptions(retryConfig.exceptions); + if (exceptions !== undefined) { + if (!exceptions.includes(errorName(error))) { + return false; + } + } + + return true; +} + +/** + * Calculates the delay, in seconds, before retrying a node. + * + * @param retryConfig The node's retry configuration, if any. + * @param nodeState The current node state (its `attemptCount` is the 1-based + * attempt number that just failed). + * @param randomFn Injectable uniform RNG in [0, 1) for deterministic testing. + */ +export function getRetryDelaySeconds( + retryConfig: RetryConfig | undefined, + nodeState: NodeState, + randomFn: () => number = Math.random, +): number { + if (!retryConfig) { + return DEFAULT_INITIAL_DELAY_SECONDS; + } + + const initialDelay = + retryConfig.initialDelay ?? DEFAULT_INITIAL_DELAY_SECONDS; + const maxDelay = retryConfig.maxDelay ?? DEFAULT_MAX_DELAY_SECONDS; + const backoffFactor = retryConfig.backoffFactor ?? DEFAULT_BACKOFF_FACTOR; + const jitter = retryConfig.jitter ?? DEFAULT_JITTER; + + const attemptCount = nodeState.attemptCount || 1; + // attemptCount is the attempt number that just failed (1-based); the first + // failure (attempt 1) uses exponent 0. + const attemptForCalc = Math.max(0, attemptCount - 1); + + let delay = initialDelay * Math.pow(backoffFactor, attemptForCalc); + delay = Math.min(delay, maxDelay); + + if (jitter > 0.0) { + // random.uniform(-jitter*delay, jitter*delay) + const span = jitter * delay; + const randomOffset = -span + randomFn() * (2 * span); + delay = Math.max(0.0, delay + randomOffset); + } + + return delay; +} diff --git a/core/test/events/event_test.ts b/core/test/events/event_test.ts index b429ae754..2a14924cf 100644 --- a/core/test/events/event_test.ts +++ b/core/test/events/event_test.ts @@ -373,6 +373,51 @@ describe('Event Utils', () => { NestedKey: 'value2', }); }); + + it('preserves workflow output and agentState keys verbatim', () => { + const camelEvent = createEvent({ + id: '123', + invocationId: 'inv1', + output: {cityName: 'Paris', timeInfo: '10:10 AM'}, + actions: createEventActions({ + agentState: {input: {userId: 42, requestedItems: ['a']}}, + }), + }); + const snakeEvent = transformToSnakeCaseEvent(camelEvent); + // Arbitrary payloads must NOT be snake_cased. + expect(snakeEvent.output).toEqual({ + cityName: 'Paris', + timeInfo: '10:10 AM', + }); + expect( + (snakeEvent.actions as Record).agent_state, + ).toEqual({input: {userId: 42, requestedItems: ['a']}}); + }); + }); + + describe('event round-trip serialization', () => { + it('round-trips workflow output and agentState without mangling keys', () => { + const original = createEvent({ + id: '123', + invocationId: 'inv1', + output: {cityName: 'Paris', nested: {timeInfo: '10:10 AM'}}, + route: ['BUG', 'LOGISTICS'], + actions: createEventActions({ + agentState: {input: {userId: 42, camelKey: 'v'}}, + }), + }); + const restored = transformToCamelCaseEvent( + transformToSnakeCaseEvent(original), + ); + expect(restored.output).toEqual({ + cityName: 'Paris', + nested: {timeInfo: '10:10 AM'}, + }); + expect(restored.route).toEqual(['BUG', 'LOGISTICS']); + expect(restored.actions?.agentState).toEqual({ + input: {userId: 42, camelKey: 'v'}, + }); + }); }); describe('generateClientFunctionCallId', () => { diff --git a/core/test/workflow/event_channel_test.ts b/core/test/workflow/event_channel_test.ts new file mode 100644 index 000000000..e2b1a446e --- /dev/null +++ b/core/test/workflow/event_channel_test.ts @@ -0,0 +1,90 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import {EventChannel} from '../../src/workflow/utils/event_channel.js'; + +async function drain(ch: EventChannel): Promise { + const out: T[] = []; + for await (const item of ch) { + out.push(item); + } + return out; +} + +describe('Phase 1 — EventChannel', () => { + it('delivers items pushed before draining (buffered)', async () => { + const ch = new EventChannel(); + ch.push(1); + ch.push(2); + ch.push(3); + ch.close(); + expect(await drain(ch)).toEqual([1, 2, 3]); + }); + + it('delivers items pushed while a consumer is awaiting (interleaved)', async () => { + const ch = new EventChannel(); + const collected: number[] = []; + const consumer = (async () => { + for await (const item of ch) { + collected.push(item); + } + })(); + + // Push across turns of the event loop while the consumer is parked. + await Promise.resolve(); + ch.push(10); + await Promise.resolve(); + ch.push(20); + await Promise.resolve(); + ch.close(); + + await consumer; + expect(collected).toEqual([10, 20]); + }); + + it('close() terminates iteration cleanly', async () => { + const ch = new EventChannel(); + ch.push('a'); + ch.close(); + ch.push('ignored-after-close'); + expect(await drain(ch)).toEqual(['a']); + expect(ch.isClosed).toBe(true); + }); + + it('fail() surfaces the error to the consumer', async () => { + const ch = new EventChannel(); + const boom = new Error('boom'); + ch.fail(boom); + await expect(drain(ch)).rejects.toThrow('boom'); + }); + + it('fail() still drains buffered items before throwing', async () => { + const ch = new EventChannel(); + ch.push(1); + ch.push(2); + ch.fail(new Error('later')); + + const seen: number[] = []; + await expect( + (async () => { + for await (const item of ch) { + seen.push(item); + } + })(), + ).rejects.toThrow('later'); + expect(seen).toEqual([1, 2]); + }); + + it('reports buffered size and ignores push after close', async () => { + const ch = new EventChannel(); + ch.push(1); + expect(ch.size).toBe(1); + ch.close(); + ch.push(2); + expect(ch.size).toBe(1); + }); +}); diff --git a/core/test/workflow/event_model_test.ts b/core/test/workflow/event_model_test.ts new file mode 100644 index 000000000..b53dfe18a --- /dev/null +++ b/core/test/workflow/event_model_test.ts @@ -0,0 +1,66 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import { + createEvent, + transformToCamelCaseEvent, + transformToSnakeCaseEvent, +} from '../../src/events/event.js'; + +describe('Phase 0 — workflow event-model extensions', () => { + it('carries first-class workflow fields on Event', () => { + const ev = createEvent({ + author: 'node_a', + output: {value: 42}, + route: 'question', + nodeInfo: {path: 'wf.node_a', outputFor: 'run_1', messageAsOutput: true}, + isolationScope: 'wf:evt_123', + actions: {agentState: {status: 3}, endOfAgent: true}, + }); + expect(ev.output).toEqual({value: 42}); + expect(ev.route).toBe('question'); + expect(ev.nodeInfo?.path).toBe('wf.node_a'); + expect(ev.nodeInfo?.messageAsOutput).toBe(true); + expect(ev.isolationScope).toBe('wf:evt_123'); + expect(ev.actions.agentState).toEqual({status: 3}); + expect(ev.actions.endOfAgent).toBe(true); + }); + + it('round-trips new fields through snake_case <-> camelCase', () => { + const ev = createEvent({ + author: 'node_a', + output: {value: 42}, + route: 'question', + nodeInfo: {path: 'wf.node_a', outputFor: 'run_1', messageAsOutput: true}, + isolationScope: 'wf:evt_123', + actions: {agentState: {status: 3}, endOfAgent: true}, + }); + + const snake = transformToSnakeCaseEvent(ev); + // Verify Python-compatible key names on the wire. + expect(snake['node_info']).toBeDefined(); + expect((snake['node_info'] as Record)['output_for']).toBe( + 'run_1', + ); + expect( + (snake['node_info'] as Record)['message_as_output'], + ).toBe(true); + expect(snake['isolation_scope']).toBe('wf:evt_123'); + const snakeActions = snake['actions'] as Record; + expect(snakeActions['agent_state']).toEqual({status: 3}); + expect(snakeActions['end_of_agent']).toBe(true); + + const back = transformToCamelCaseEvent(snake); + expect(back.nodeInfo?.path).toBe('wf.node_a'); + expect(back.nodeInfo?.outputFor).toBe('run_1'); + expect(back.nodeInfo?.messageAsOutput).toBe(true); + expect(back.isolationScope).toBe('wf:evt_123'); + expect(back.route).toBe('question'); + expect(back.actions.agentState).toEqual({status: 3}); + expect(back.actions.endOfAgent).toBe(true); + }); +}); diff --git a/core/test/workflow/foundations_test.ts b/core/test/workflow/foundations_test.ts new file mode 100644 index 000000000..6d69e10bd --- /dev/null +++ b/core/test/workflow/foundations_test.ts @@ -0,0 +1,179 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import { + DynamicNodeFailError, + NodeInterruptedError, + NodeTimeoutError, +} from '../../src/workflow/errors.js'; +import { + createNodeState, + isNodeState, + NodeState, +} from '../../src/workflow/node_state.js'; +import {NodeStatus} from '../../src/workflow/node_status.js'; +import {normalizeRetryExceptions} from '../../src/workflow/retry_config.js'; +import { + getRetryDelaySeconds, + shouldRetryNode, +} from '../../src/workflow/utils/retry_utils.js'; + +describe('Phase 0 — errors', () => { + it('NodeTimeoutError carries nodeName/timeout and is instanceof Error', () => { + const err = new NodeTimeoutError({nodeName: 'n1', timeout: 2.5}); + expect(err).toBeInstanceOf(Error); + expect(err).toBeInstanceOf(NodeTimeoutError); + expect(err.name).toBe('NodeTimeoutError'); + expect(err.nodeName).toBe('n1'); + expect(err.timeout).toBe(2.5); + expect(err.message).toContain("Node 'n1' timed out after 2.5 seconds"); + }); + + it('NodeInterruptedError is a distinct catchable error', () => { + const err = new NodeInterruptedError(); + expect(err).toBeInstanceOf(NodeInterruptedError); + expect(err.name).toBe('NodeInterruptedError'); + }); + + it('DynamicNodeFailError wraps the underlying error + node path', () => { + const cause = new TypeError('boom'); + const err = new DynamicNodeFailError({ + message: 'dynamic node failed', + error: cause, + errorNodePath: 'wf.child.0', + }); + expect(err).toBeInstanceOf(DynamicNodeFailError); + expect(err.error).toBe(cause); + expect(err.errorNodePath).toBe('wf.child.0'); + }); +}); + +describe('Phase 0 — NodeStatus / NodeState', () => { + it('NodeStatus values match the Python enum ordinals', () => { + expect(NodeStatus.INACTIVE).toBe(0); + expect(NodeStatus.PENDING).toBe(1); + expect(NodeStatus.RUNNING).toBe(2); + expect(NodeStatus.COMPLETED).toBe(3); + expect(NodeStatus.WAITING).toBe(4); + expect(NodeStatus.FAILED).toBe(5); + expect(NodeStatus.CANCELLED).toBe(6); + }); + + it('createNodeState applies Python-aligned defaults', () => { + const s = createNodeState(); + expect(s.status).toBe(NodeStatus.INACTIVE); + expect(s.attemptCount).toBe(1); + expect(s.interrupts).toEqual([]); + expect(s.resumeInputs).toEqual({}); + expect(s.runCounter).toBe(0); + expect(s.runId).toBeUndefined(); + expect(s.parentRunId).toBeUndefined(); + }); + + it('createNodeState overlays partial values', () => { + const s = createNodeState({ + status: NodeStatus.RUNNING, + attemptCount: 3, + runId: 'r1', + }); + expect(s.status).toBe(NodeStatus.RUNNING); + expect(s.attemptCount).toBe(3); + expect(s.runId).toBe('r1'); + }); + + it('isNodeState recognizes valid/invalid shapes', () => { + expect(isNodeState(createNodeState())).toBe(true); + expect(isNodeState({})).toBe(false); + expect(isNodeState(null)).toBe(false); + expect(isNodeState({status: 'RUNNING'})).toBe(false); + }); +}); + +describe('Phase 0 — retry config normalization', () => { + it('returns undefined (retry-all) for null/undefined', () => { + expect(normalizeRetryExceptions(undefined)).toBeUndefined(); + expect(normalizeRetryExceptions(null)).toBeUndefined(); + }); + + it('normalizes error classes and strings to class-name strings', () => { + expect(normalizeRetryExceptions([TypeError, 'RangeError'])).toEqual([ + 'TypeError', + 'RangeError', + ]); + }); +}); + +describe('Phase 0 — shouldRetryNode', () => { + const state = (attemptCount: number): NodeState => + createNodeState({attemptCount}); + + it('never retries without a config', () => { + expect(shouldRetryNode(new Error('x'), undefined, state(1))).toBe(false); + }); + + it('retries until maxAttempts is reached (default 5)', () => { + expect(shouldRetryNode(new Error('x'), {}, state(1))).toBe(true); + expect(shouldRetryNode(new Error('x'), {}, state(4))).toBe(true); + expect(shouldRetryNode(new Error('x'), {}, state(5))).toBe(false); + }); + + it('respects an explicit maxAttempts', () => { + expect(shouldRetryNode(new Error('x'), {maxAttempts: 2}, state(1))).toBe( + true, + ); + expect(shouldRetryNode(new Error('x'), {maxAttempts: 2}, state(2))).toBe( + false, + ); + }); + + it('only retries listed exception types when provided', () => { + const cfg = {exceptions: [TypeError]}; + expect(shouldRetryNode(new TypeError('x'), cfg, state(1))).toBe(true); + expect(shouldRetryNode(new RangeError('x'), cfg, state(1))).toBe(false); + }); +}); + +describe('Phase 0 — getRetryDelaySeconds', () => { + it('defaults to 1.0s with no config', () => { + expect(getRetryDelaySeconds(undefined, createNodeState())).toBe(1.0); + }); + + it('applies exponential backoff (jitter disabled)', () => { + const cfg = {initialDelay: 1, backoffFactor: 2, jitter: 0}; + // attempt 1 -> exponent 0 -> 1s + expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}))).toBe( + 1, + ); + // attempt 3 -> exponent 2 -> 4s + expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 3}))).toBe( + 4, + ); + }); + + it('caps delay at maxDelay', () => { + const cfg = {initialDelay: 10, backoffFactor: 10, maxDelay: 30, jitter: 0}; + expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 5}))).toBe( + 30, + ); + }); + + it('applies bounded symmetric jitter using the injected RNG', () => { + const cfg = {initialDelay: 4, backoffFactor: 1, jitter: 1}; + // randomFn=0.5 -> offset 0 -> exactly base delay (4) + expect( + getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 0.5), + ).toBe(4); + // randomFn=0 -> offset -span -> max(0, 4-4)=0 + expect( + getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 0), + ).toBe(0); + // randomFn=1 -> offset +span -> 4+4=8 + expect( + getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 1), + ).toBe(8); + }); +}); From ac77d70afc565847e556e83a962b66d7b8d4b719 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 14:45:47 -0700 Subject: [PATCH 2/5] refactor(events): brand Event with a signature symbol for isEvent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR1 review feedback: identify Event objects via a Symbol.for('google.adk.event') brand — set by createEvent and checked by isEvent — matching the signature-symbol guards used across ADK (isBaseTool, isBaseAgent, ...) instead of structural duck-typing. The brand is declared optional on the Event interface and is a non-serializable runtime marker, so events reconstructed from storage/session payloads are intentionally unbranded (documented on the interface and covered by a round-trip test). Adds isEvent unit tests (branded event, impostor rejection, non-object rejection, round-trip brand drop). --- core/src/events/event.ts | 27 ++++++++++++++++++---- core/test/workflow/event_model_test.ts | 31 ++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/core/src/events/event.ts b/core/src/events/event.ts index 313477853..7e4316200 100644 --- a/core/src/events/event.ts +++ b/core/src/events/event.ts @@ -12,6 +12,16 @@ import {randomUUID} from '../utils/env_aware_utils.js'; import {toCamelCase, toSnakeCase} from '../utils/object_notation_utils.js'; import {createEventActions, EventActions} from './event_actions.js'; +/** + * A unique symbol identifying ADK Event objects. + * + * Events are plain objects produced by {@link createEvent} rather than class + * instances, so they carry this signature as a brand and {@link isEvent} checks + * for it — mirroring the `Symbol.for('google.adk.*')` guards used across ADK + * (e.g. {@link isBaseTool}, {@link isBaseAgent}). + */ +const EVENT_SIGNATURE_SYMBOL = Symbol.for('google.adk.event'); + /** * Workflow-node provenance attached to an event. * @@ -39,6 +49,15 @@ export interface NodeInfo { taken by the agents like function calls, etc. */ export interface Event extends LlmResponse { + /** + * Signature brand identifying this object as an ADK {@link Event}. + * + * Set by {@link createEvent} and checked by {@link isEvent}. Optional because + * events are also reconstructed from storage/session payloads, where the + * (non-serializable) brand is absent. + */ + readonly [EVENT_SIGNATURE_SYMBOL]?: true; + /** * The unique identifier of the event. * Do not assign the ID. It will be assigned by the session. @@ -129,6 +148,7 @@ export interface CreateEventParams extends Omit, 'actions'> { export function createEvent(params: CreateEventParams = {}): Event { return { ...params, + [EVENT_SIGNATURE_SYMBOL]: true, id: params.id || createNewEventId(), invocationId: params.invocationId || '', author: params.author, @@ -294,10 +314,9 @@ export function isEvent(obj: unknown): obj is Event { return ( typeof obj === 'object' && obj !== null && - 'invocationId' in obj && - typeof (obj as Event).invocationId === 'string' && - 'actions' in obj && - typeof (obj as Event).actions === 'object' + EVENT_SIGNATURE_SYMBOL in obj && + (obj as {[EVENT_SIGNATURE_SYMBOL]?: unknown})[EVENT_SIGNATURE_SYMBOL] === + true ); } diff --git a/core/test/workflow/event_model_test.ts b/core/test/workflow/event_model_test.ts index b53dfe18a..d78d54c71 100644 --- a/core/test/workflow/event_model_test.ts +++ b/core/test/workflow/event_model_test.ts @@ -7,6 +7,7 @@ import {describe, expect, it} from 'vitest'; import { createEvent, + isEvent, transformToCamelCaseEvent, transformToSnakeCaseEvent, } from '../../src/events/event.js'; @@ -64,3 +65,33 @@ describe('Phase 0 — workflow event-model extensions', () => { expect(back.actions.endOfAgent).toBe(true); }); }); + +describe('isEvent — signature-symbol brand', () => { + it('recognizes events built by createEvent', () => { + expect(isEvent(createEvent({author: 'node_a'}))).toBe(true); + }); + + it('rejects non-objects and null', () => { + expect(isEvent(null)).toBe(false); + expect(isEvent(undefined)).toBe(false); + expect(isEvent('event')).toBe(false); + expect(isEvent(42)).toBe(false); + }); + + it('rejects event-shaped impostors that lack the brand', () => { + // Structurally event-like, but not produced by createEvent: the old + // duck-typing guard would accept this; the brand-based guard does not. + const impostor = {invocationId: 'inv-1', actions: {}, output: {value: 1}}; + expect(isEvent(impostor)).toBe(false); + }); + + it('drops the (non-serializable) brand across a snake/camel round-trip', () => { + // The brand is a runtime marker only; a rehydrated event is not branded. + const branded = createEvent({author: 'node_a', output: {value: 42}}); + const rehydrated = transformToCamelCaseEvent( + transformToSnakeCaseEvent(branded), + ); + expect(isEvent(branded)).toBe(true); + expect(isEvent(rehydrated)).toBe(false); + }); +}); From 4c1a8172d8c281133f43c349cd2da4be4ae185e6 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 15:54:48 -0700 Subject: [PATCH 3/5] refactor(workflow): address PR1 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolves review comments on #587: - events: drop 6 unused EventActions fields (output, joinCompleted, toolExecution, requestInput, nodeExecutionReplay, route) that were dead across the whole branch and left arbitrary `unknown` payloads unpreserved; removing them also eliminates the round-trip hole and the ambiguous second `route` carrier. Add shared RouteKey/Route types for Event.route. - errors: add name-based guards (isNodeInterruptedError, isNodeTimeoutError, isDynamicNodeFailError) so catch sites stay correct across duplicate-package boundaries instead of relying on `instanceof`. - branch_path: convert createSubBranch/commonPrefixOf from static methods to standalone util functions (kept fromString/commonPrefix as factories). - async_queue: fold EventChannel into the existing AsyncQueue instead of shipping a duplicate queue — port drain-before-error, sticky failure, and error()-closes; add isClosed/size getters and a fail() method (error() retained as an alias). Delete event_channel.ts and move its tests. This also fixes the close()-then-fail() error-swallow bug. - retry: introduce prepareRetryConfig/PreparedRetryConfig so a node's exception filter is normalized+validated once at construction, not re-normalized (and potentially thrown) on every retry check; make shouldRetryNode/ getRetryDelaySeconds take a required prepared config and drop the unreachable no-config branches (and the misleading test that pinned one). Full core suite green (2338 tests). --- core/src/common.ts | 8 +- core/src/events/event.ts | 14 ++- core/src/events/event_actions.ts | 14 --- core/src/utils/async_queue.ts | 74 ++++++++++++---- core/src/workflow/branch_path.ts | 40 ++++----- core/src/workflow/errors.ts | 21 +++++ core/src/workflow/retry_config.ts | 37 ++++++++ core/src/workflow/utils/event_channel.ts | 108 ----------------------- core/src/workflow/utils/retry_utils.ts | 26 ++---- core/test/utils/async_queue_test.ts | 66 ++++++++++++++ core/test/workflow/event_channel_test.ts | 90 ------------------- core/test/workflow/foundations_test.ts | 62 ++++++++----- 12 files changed, 269 insertions(+), 291 deletions(-) delete mode 100644 core/src/workflow/utils/event_channel.ts delete mode 100644 core/test/workflow/event_channel_test.ts diff --git a/core/src/common.ts b/core/src/common.ts index e75aa1881..026811635 100644 --- a/core/src/common.ts +++ b/core/src/common.ts @@ -143,7 +143,13 @@ export { pruneThoughts, stringifyContent, } from './events/event.js'; -export type {CreateEventParams, Event, NodeInfo} from './events/event.js'; +export type { + CreateEventParams, + Event, + NodeInfo, + Route, + RouteKey, +} from './events/event.js'; export {createEventActions} from './events/event_actions.js'; export type {EventActions} from './events/event_actions.js'; export {EventType, toStructuredEvents} from './events/structured_events.js'; diff --git a/core/src/events/event.ts b/core/src/events/event.ts index 7e4316200..0f716a417 100644 --- a/core/src/events/event.ts +++ b/core/src/events/event.ts @@ -22,6 +22,18 @@ import {createEventActions, EventActions} from './event_actions.js'; */ const EVENT_SIGNATURE_SYMBOL = Symbol.for('google.adk.event'); +/** + * A single route key emitted by a routing node and matched against graph edge + * routes. Mirrors the graph's `RouteValue`. + */ +export type RouteKey = string | number | boolean; + +/** + * The route(s) a routing node emits: a single key fires one branch; an array + * fires every branch whose route matches any listed key (multi-route dispatch). + */ +export type Route = RouteKey | RouteKey[]; + /** * Workflow-node provenance attached to an event. * @@ -117,7 +129,7 @@ export interface Event extends LlmResponse { * array fires every branch whose route matches any listed value (multi-route * dispatch). Mirrors Python `Event.route`. */ - route?: string | number | boolean | Array; + route?: Route; /** * Workflow: provenance of the emitting node. Mirrors Python `Event.node_info`. diff --git a/core/src/events/event_actions.ts b/core/src/events/event_actions.ts index 54c89fa71..b89f494a2 100644 --- a/core/src/events/event_actions.ts +++ b/core/src/events/event_actions.ts @@ -57,13 +57,6 @@ export interface EventActions { */ requestedToolConfirmations: {[key: string]: ToolConfirmation}; - /** Workflow / custom event actions */ - output?: unknown; - joinCompleted?: unknown; - toolExecution?: unknown; - requestInput?: unknown; - nodeExecutionReplay?: unknown; - /** * Workflow: a serialized node/agent state snapshot used for resumable * checkpointing. Mirrors Python `EventActions.agent_state`. @@ -75,13 +68,6 @@ export interface EventActions { * execution for this invocation. Mirrors Python `EventActions.end_of_agent`. */ endOfAgent?: boolean; - - /** - * Workflow: route key(s) selected by a routing node (alternative carrier to - * the top-level `Event.route`, used by callbacks/tools). A single value or an - * array for multi-route dispatch. - */ - route?: string | number | boolean | Array; } /** diff --git a/core/src/utils/async_queue.ts b/core/src/utils/async_queue.ts index 0a2fb0f63..2a7cfcbd1 100644 --- a/core/src/utils/async_queue.ts +++ b/core/src/utils/async_queue.ts @@ -5,7 +5,20 @@ */ /** - * A generic, async-safe queue that implements AsyncIterable. + * A generic, single-consumer async queue that implements AsyncIterable, + * bridging a *push* producer to a *pull* consumer (`for await (const x of q)`). + * + * Producers call {@link push} as values are produced and {@link close} (or + * {@link fail}) when finished. A single consumer drains the queue. + * + * Semantics: + * - Buffered items are always delivered before an end/error signal. + * - {@link close} ends iteration cleanly (`done: true`); idempotent. + * - {@link fail} surfaces the error to the consumer *after* any buffered items + * have been drained; it is sticky (first failure wins) and also closes the + * queue, so a later `close()` can't discard the error. + * - {@link push} after close/fail is ignored (the producer has already + * signalled completion). */ export class AsyncQueue implements AsyncIterable { private queue: T[] = []; @@ -14,45 +27,72 @@ export class AsyncQueue implements AsyncIterable { reject: (reason?: unknown) => void; }> = []; private closed = false; - private errorVal?: unknown; + private failure?: {error: unknown}; + /** Whether the queue has been closed or failed. */ + get isClosed(): boolean { + return this.closed; + } + + /** Number of items buffered and not yet consumed. */ + get size(): number { + return this.queue.length; + } + + /** + * Enqueues a value. If a consumer is currently awaiting, it is resolved + * immediately; otherwise the value is buffered. No-op once closed/failed. + */ push(value: T) { if (this.closed) return; - if (this.resolvers.length > 0) { - const {resolve} = this.resolvers.shift()!; - resolve({value, done: false}); + const resolver = this.resolvers.shift(); + if (resolver) { + resolver.resolve({value, done: false}); } else { this.queue.push(value); } } - error(err: unknown) { - this.errorVal = err; + /** + * Signals that production failed. Buffered items are still delivered first; + * once the buffer drains, the consumer's next `next()` rejects with `error`. + * Sticky (first failure wins) and closes the queue. + */ + fail(error: unknown) { + if (this.failure) return; + this.failure = {error}; + this.closed = true; while (this.resolvers.length > 0) { - const {reject} = this.resolvers.shift()!; - reject(err); + this.resolvers.shift()!.reject(error); } } + /** @deprecated Alias for {@link fail}; kept for existing callers. */ + error(err: unknown) { + this.fail(err); + } + + /** + * Signals that no more items will be produced. Any awaiting consumer receives + * `{done: true}`. Idempotent. + */ close() { + if (this.closed) return; this.closed = true; while (this.resolvers.length > 0) { - const {resolve} = this.resolvers.shift()!; - resolve({value: undefined as never, done: true}); + this.resolvers.shift()!.resolve({value: undefined as never, done: true}); } } [Symbol.asyncIterator](): AsyncIterator { return { - next: () => { - if (this.errorVal) { - const err = this.errorVal; - this.errorVal = undefined; - return Promise.reject(err); - } + next: (): Promise> => { if (this.queue.length > 0) { return Promise.resolve({value: this.queue.shift()!, done: false}); } + if (this.failure) { + return Promise.reject(this.failure.error); + } if (this.closed) { return Promise.resolve({value: undefined as never, done: true}); } diff --git a/core/src/workflow/branch_path.ts b/core/src/workflow/branch_path.ts index 63fd58395..978a1d24d 100644 --- a/core/src/workflow/branch_path.ts +++ b/core/src/workflow/branch_path.ts @@ -67,26 +67,26 @@ export class BranchPath { } return new BranchPath(common); } +} - /** - * Creates a new dot-separated sub-branch string by appending a segment. - * - * @example createSubBranch('parent', {name: 'child', runId: '1'}) -> 'parent.child@1' - * @example createSubBranch(undefined, {name: 'agent'}) -> 'agent' - */ - static createSubBranch( - baseBranch: string | undefined | null, - options: {name: string; runId?: string}, - ): string { - return BranchPath.fromString(baseBranch) - .append(options.name, options.runId) - .toString(); - } +/** + * Creates a new dot-separated sub-branch string by appending a segment. + * + * @example createSubBranch('parent', {name: 'child', runId: '1'}) -> 'parent.child@1' + * @example createSubBranch(undefined, {name: 'agent'}) -> 'agent' + */ +export function createSubBranch( + baseBranch: string | undefined | null, + options: {name: string; runId?: string}, +): string { + return BranchPath.fromString(baseBranch) + .append(options.name, options.runId) + .toString(); +} - /** Finds the common prefix of a list of dot-separated branch strings. */ - static commonPrefixOf(branches: string[]): string { - return BranchPath.commonPrefix( - branches.map((b) => BranchPath.fromString(b)), - ).toString(); - } +/** Finds the common prefix of a list of dot-separated branch strings. */ +export function commonPrefixOf(branches: string[]): string { + return BranchPath.commonPrefix( + branches.map((b) => BranchPath.fromString(b)), + ).toString(); } diff --git a/core/src/workflow/errors.ts b/core/src/workflow/errors.ts index 14c73a44a..82b254c89 100644 --- a/core/src/workflow/errors.ts +++ b/core/src/workflow/errors.ts @@ -28,6 +28,17 @@ export class NodeInterruptedError extends Error { } } +/** + * Type guard for {@link NodeInterruptedError}. + * + * Matches on `name` rather than `instanceof` so it stays correct when errors + * cross a package boundary (two copies of adk-js in one runtime would fail an + * `instanceof` check between them). + */ +export function isNodeInterruptedError(e: unknown): e is NodeInterruptedError { + return e instanceof Error && e.name === 'NodeInterruptedError'; +} + /** * Raised when a node exceeds its configured timeout. * @@ -53,6 +64,11 @@ export class NodeTimeoutError extends Error { } } +/** Type guard for {@link NodeTimeoutError} (name-based; see above). */ +export function isNodeTimeoutError(e: unknown): e is NodeTimeoutError { + return e instanceof Error && e.name === 'NodeTimeoutError'; +} + /** * Raised when a dynamic node fails. * @@ -76,3 +92,8 @@ export class DynamicNodeFailError extends Error { Object.setPrototypeOf(this, DynamicNodeFailError.prototype); } } + +/** Type guard for {@link DynamicNodeFailError} (name-based; see above). */ +export function isDynamicNodeFailError(e: unknown): e is DynamicNodeFailError { + return e instanceof Error && e.name === 'DynamicNodeFailError'; +} diff --git a/core/src/workflow/retry_config.ts b/core/src/workflow/retry_config.ts index 65ea24efc..b020d5db9 100644 --- a/core/src/workflow/retry_config.ts +++ b/core/src/workflow/retry_config.ts @@ -83,3 +83,40 @@ export function normalizeRetryExceptions( ); }); } + +/** + * A {@link RetryConfig} whose `exceptions` filter has been normalized to error + * class-name strings once, up front. + * + * Produced by {@link prepareRetryConfig} when a node accepts its config, so the + * retry hot path neither re-normalizes on every failure nor throws on a + * malformed config from inside the retry loop. + */ +export interface PreparedRetryConfig { + readonly maxAttempts?: number; + readonly initialDelay?: number; + readonly maxDelay?: number; + readonly backoffFactor?: number; + readonly jitter?: number; + /** Normalized exception names; `undefined` means retry on all errors. */ + readonly exceptions?: readonly string[]; +} + +/** + * Validates and normalizes a {@link RetryConfig} once, at config-acceptance + * time (i.e. when a node is constructed). + * + * Throws if `exceptions` contains a malformed entry, surfacing the + * misconfiguration at construction rather than masking a node's real error from + * inside the retry path. + */ +export function prepareRetryConfig(config: RetryConfig): PreparedRetryConfig { + return { + maxAttempts: config.maxAttempts, + initialDelay: config.initialDelay, + maxDelay: config.maxDelay, + backoffFactor: config.backoffFactor, + jitter: config.jitter, + exceptions: normalizeRetryExceptions(config.exceptions), + }; +} diff --git a/core/src/workflow/utils/event_channel.ts b/core/src/workflow/utils/event_channel.ts deleted file mode 100644 index 48914a85d..000000000 --- a/core/src/workflow/utils/event_channel.ts +++ /dev/null @@ -1,108 +0,0 @@ -/** - * @license - * Copyright 2026 Google LLC - * SPDX-License-Identifier: Apache-2.0 - */ - -/** - * A single-consumer async queue that bridges the workflow engine's *push* model - * (nodes/`ctx.runNode()` push events as they run) to the runtime's *pull* model - * (the workflow's outer async generator drains and re-yields them). - * - * Producers call {@link push} as events are produced and {@link close} (or - * {@link fail}) when finished. A single consumer drains the channel by - * `for await (const ev of channel)`. - * - * Semantics: - * - Buffered items are always delivered before an end/error signal. - * - {@link close} ends iteration cleanly (`done: true`). - * - {@link fail} surfaces the error to the consumer *after* any buffered items - * have been drained. - * - {@link push} after close/fail is ignored (the producer has already - * signalled completion). - */ -export class EventChannel implements AsyncIterable { - private readonly buffer: T[] = []; - private readonly waiters: Array<{ - resolve: (r: IteratorResult) => void; - reject: (e: unknown) => void; - }> = []; - private closed = false; - private failure?: {error: unknown}; - - /** Whether the channel has been closed or failed. */ - get isClosed(): boolean { - return this.closed; - } - - /** Number of items buffered and not yet consumed. */ - get size(): number { - return this.buffer.length; - } - - /** - * Enqueues an item. If a consumer is currently awaiting, it is resolved - * immediately; otherwise the item is buffered. No-op once closed/failed. - */ - push(item: T): void { - if (this.closed) { - return; - } - const waiter = this.waiters.shift(); - if (waiter) { - waiter.resolve({value: item, done: false}); - } else { - this.buffer.push(item); - } - } - - /** - * Signals that no more items will be produced. Any awaiting consumer receives - * `{done: true}`. Idempotent. - */ - close(): void { - if (this.closed) { - return; - } - this.closed = true; - while (this.waiters.length > 0) { - this.waiters.shift()!.resolve({value: undefined as never, done: true}); - } - } - - /** - * Signals that production failed. Buffered items are still delivered first; - * once the buffer drains, the consumer's next `next()` rejects with `error`. - * Idempotent (first failure wins). - */ - fail(error: unknown): void { - if (this.closed) { - return; - } - this.failure = {error}; - this.closed = true; - // If a consumer is awaiting, the buffer is empty, so surface the error now. - while (this.waiters.length > 0) { - this.waiters.shift()!.reject(error); - } - } - - [Symbol.asyncIterator](): AsyncIterator { - return { - next: (): Promise> => { - if (this.buffer.length > 0) { - return Promise.resolve({value: this.buffer.shift()!, done: false}); - } - if (this.failure) { - return Promise.reject(this.failure.error); - } - if (this.closed) { - return Promise.resolve({value: undefined as never, done: true}); - } - return new Promise>((resolve, reject) => { - this.waiters.push({resolve, reject}); - }); - }, - }; - } -} diff --git a/core/src/workflow/utils/retry_utils.ts b/core/src/workflow/utils/retry_utils.ts index 57680e801..18e1481a6 100644 --- a/core/src/workflow/utils/retry_utils.ts +++ b/core/src/workflow/utils/retry_utils.ts @@ -11,7 +11,7 @@ */ import {NodeState} from '../node_state.js'; -import {RetryConfig, normalizeRetryExceptions} from '../retry_config.js'; +import {PreparedRetryConfig} from '../retry_config.js'; const DEFAULT_MAX_ATTEMPTS = 5; const DEFAULT_INITIAL_DELAY_SECONDS = 1.0; @@ -39,18 +39,14 @@ function errorName(error: unknown): string { * Checks if a failed node should be retried based on its retry config. * * @param error The error thrown by the node. - * @param retryConfig The node's retry configuration, if any. + * @param retryConfig The node's prepared (normalized) retry configuration. * @param nodeState The current node state (its `attemptCount` is 1-based). */ export function shouldRetryNode( error: unknown, - retryConfig: RetryConfig | undefined, + retryConfig: PreparedRetryConfig, nodeState: NodeState, ): boolean { - if (!retryConfig) { - return false; - } - const attemptCount = nodeState.attemptCount; const maxAttempts = retryConfig.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; @@ -60,11 +56,9 @@ export function shouldRetryNode( return false; } - const exceptions = normalizeRetryExceptions(retryConfig.exceptions); - if (exceptions !== undefined) { - if (!exceptions.includes(errorName(error))) { - return false; - } + const exceptions = retryConfig.exceptions; + if (exceptions !== undefined && !exceptions.includes(errorName(error))) { + return false; } return true; @@ -73,20 +67,16 @@ export function shouldRetryNode( /** * Calculates the delay, in seconds, before retrying a node. * - * @param retryConfig The node's retry configuration, if any. + * @param retryConfig The node's prepared (normalized) retry configuration. * @param nodeState The current node state (its `attemptCount` is the 1-based * attempt number that just failed). * @param randomFn Injectable uniform RNG in [0, 1) for deterministic testing. */ export function getRetryDelaySeconds( - retryConfig: RetryConfig | undefined, + retryConfig: PreparedRetryConfig, nodeState: NodeState, randomFn: () => number = Math.random, ): number { - if (!retryConfig) { - return DEFAULT_INITIAL_DELAY_SECONDS; - } - const initialDelay = retryConfig.initialDelay ?? DEFAULT_INITIAL_DELAY_SECONDS; const maxDelay = retryConfig.maxDelay ?? DEFAULT_MAX_DELAY_SECONDS; diff --git a/core/test/utils/async_queue_test.ts b/core/test/utils/async_queue_test.ts index a20a52c29..8d8670a85 100644 --- a/core/test/utils/async_queue_test.ts +++ b/core/test/utils/async_queue_test.ts @@ -86,3 +86,69 @@ describe('AsyncQueue', () => { expect(res.done).toBe(true); }); }); + +describe('AsyncQueue — failure & lifecycle', () => { + async function drain(queue: AsyncQueue): Promise { + const out: T[] = []; + for await (const item of queue) { + out.push(item); + } + return out; + } + + it('exposes isClosed and buffered size', () => { + const queue = new AsyncQueue(); + expect(queue.isClosed).toBe(false); + queue.push(1); + expect(queue.size).toBe(1); + queue.close(); + expect(queue.isClosed).toBe(true); + queue.push(2); // ignored after close + expect(queue.size).toBe(1); + }); + + it('fail() surfaces the error to the consumer', async () => { + const queue = new AsyncQueue(); + queue.fail(new Error('boom')); + await expect(drain(queue)).rejects.toThrow('boom'); + }); + + it('fail() drains buffered items before throwing (drain-before-error)', async () => { + const queue = new AsyncQueue(); + queue.push(1); + queue.push(2); + queue.fail(new Error('later')); + + const seen: number[] = []; + await expect( + (async () => { + for await (const item of queue) { + seen.push(item); + } + })(), + ).rejects.toThrow('later'); + expect(seen).toEqual([1, 2]); + }); + + it('keeps the failure sticky across repeated next() calls', async () => { + const queue = new AsyncQueue(); + const iterator = queue[Symbol.asyncIterator](); + queue.fail(new Error('sticky')); + await expect(iterator.next()).rejects.toThrow('sticky'); + await expect(iterator.next()).rejects.toThrow('sticky'); + }); + + it('does not swallow a fail() that lands after close()', async () => { + const queue = new AsyncQueue(); + queue.close(); + queue.fail(new Error('late failure')); + await expect(drain(queue)).rejects.toThrow('late failure'); + }); + + it('keeps the first failure (first failure wins)', async () => { + const queue = new AsyncQueue(); + queue.fail(new Error('first')); + queue.fail(new Error('second')); + await expect(drain(queue)).rejects.toThrow('first'); + }); +}); diff --git a/core/test/workflow/event_channel_test.ts b/core/test/workflow/event_channel_test.ts deleted file mode 100644 index e2b1a446e..000000000 --- a/core/test/workflow/event_channel_test.ts +++ /dev/null @@ -1,90 +0,0 @@ -/** - * @license - * Copyright 2026 Google LLC - * SPDX-License-Identifier: Apache-2.0 - */ - -import {describe, expect, it} from 'vitest'; -import {EventChannel} from '../../src/workflow/utils/event_channel.js'; - -async function drain(ch: EventChannel): Promise { - const out: T[] = []; - for await (const item of ch) { - out.push(item); - } - return out; -} - -describe('Phase 1 — EventChannel', () => { - it('delivers items pushed before draining (buffered)', async () => { - const ch = new EventChannel(); - ch.push(1); - ch.push(2); - ch.push(3); - ch.close(); - expect(await drain(ch)).toEqual([1, 2, 3]); - }); - - it('delivers items pushed while a consumer is awaiting (interleaved)', async () => { - const ch = new EventChannel(); - const collected: number[] = []; - const consumer = (async () => { - for await (const item of ch) { - collected.push(item); - } - })(); - - // Push across turns of the event loop while the consumer is parked. - await Promise.resolve(); - ch.push(10); - await Promise.resolve(); - ch.push(20); - await Promise.resolve(); - ch.close(); - - await consumer; - expect(collected).toEqual([10, 20]); - }); - - it('close() terminates iteration cleanly', async () => { - const ch = new EventChannel(); - ch.push('a'); - ch.close(); - ch.push('ignored-after-close'); - expect(await drain(ch)).toEqual(['a']); - expect(ch.isClosed).toBe(true); - }); - - it('fail() surfaces the error to the consumer', async () => { - const ch = new EventChannel(); - const boom = new Error('boom'); - ch.fail(boom); - await expect(drain(ch)).rejects.toThrow('boom'); - }); - - it('fail() still drains buffered items before throwing', async () => { - const ch = new EventChannel(); - ch.push(1); - ch.push(2); - ch.fail(new Error('later')); - - const seen: number[] = []; - await expect( - (async () => { - for await (const item of ch) { - seen.push(item); - } - })(), - ).rejects.toThrow('later'); - expect(seen).toEqual([1, 2]); - }); - - it('reports buffered size and ignores push after close', async () => { - const ch = new EventChannel(); - ch.push(1); - expect(ch.size).toBe(1); - ch.close(); - ch.push(2); - expect(ch.size).toBe(1); - }); -}); diff --git a/core/test/workflow/foundations_test.ts b/core/test/workflow/foundations_test.ts index 6d69e10bd..1189f7fbe 100644 --- a/core/test/workflow/foundations_test.ts +++ b/core/test/workflow/foundations_test.ts @@ -16,7 +16,10 @@ import { NodeState, } from '../../src/workflow/node_state.js'; import {NodeStatus} from '../../src/workflow/node_status.js'; -import {normalizeRetryExceptions} from '../../src/workflow/retry_config.js'; +import { + normalizeRetryExceptions, + prepareRetryConfig, +} from '../../src/workflow/retry_config.js'; import { getRetryDelaySeconds, shouldRetryNode, @@ -105,45 +108,51 @@ describe('Phase 0 — retry config normalization', () => { 'RangeError', ]); }); + + it('prepareRetryConfig validates exceptions eagerly (throws on malformed)', () => { + expect(() => + prepareRetryConfig({ + exceptions: [42 as unknown as string], + }), + ).toThrow(/error class names/i); + // Well-formed config normalizes the exception filter up front. + expect(prepareRetryConfig({exceptions: [TypeError]}).exceptions).toEqual([ + 'TypeError', + ]); + }); }); describe('Phase 0 — shouldRetryNode', () => { const state = (attemptCount: number): NodeState => createNodeState({attemptCount}); - it('never retries without a config', () => { - expect(shouldRetryNode(new Error('x'), undefined, state(1))).toBe(false); - }); - it('retries until maxAttempts is reached (default 5)', () => { - expect(shouldRetryNode(new Error('x'), {}, state(1))).toBe(true); - expect(shouldRetryNode(new Error('x'), {}, state(4))).toBe(true); - expect(shouldRetryNode(new Error('x'), {}, state(5))).toBe(false); + const cfg = prepareRetryConfig({}); + expect(shouldRetryNode(new Error('x'), cfg, state(1))).toBe(true); + expect(shouldRetryNode(new Error('x'), cfg, state(4))).toBe(true); + expect(shouldRetryNode(new Error('x'), cfg, state(5))).toBe(false); }); it('respects an explicit maxAttempts', () => { - expect(shouldRetryNode(new Error('x'), {maxAttempts: 2}, state(1))).toBe( - true, - ); - expect(shouldRetryNode(new Error('x'), {maxAttempts: 2}, state(2))).toBe( - false, - ); + const cfg = prepareRetryConfig({maxAttempts: 2}); + expect(shouldRetryNode(new Error('x'), cfg, state(1))).toBe(true); + expect(shouldRetryNode(new Error('x'), cfg, state(2))).toBe(false); }); it('only retries listed exception types when provided', () => { - const cfg = {exceptions: [TypeError]}; + const cfg = prepareRetryConfig({exceptions: [TypeError]}); expect(shouldRetryNode(new TypeError('x'), cfg, state(1))).toBe(true); expect(shouldRetryNode(new RangeError('x'), cfg, state(1))).toBe(false); }); }); describe('Phase 0 — getRetryDelaySeconds', () => { - it('defaults to 1.0s with no config', () => { - expect(getRetryDelaySeconds(undefined, createNodeState())).toBe(1.0); - }); - it('applies exponential backoff (jitter disabled)', () => { - const cfg = {initialDelay: 1, backoffFactor: 2, jitter: 0}; + const cfg = prepareRetryConfig({ + initialDelay: 1, + backoffFactor: 2, + jitter: 0, + }); // attempt 1 -> exponent 0 -> 1s expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}))).toBe( 1, @@ -155,14 +164,23 @@ describe('Phase 0 — getRetryDelaySeconds', () => { }); it('caps delay at maxDelay', () => { - const cfg = {initialDelay: 10, backoffFactor: 10, maxDelay: 30, jitter: 0}; + const cfg = prepareRetryConfig({ + initialDelay: 10, + backoffFactor: 10, + maxDelay: 30, + jitter: 0, + }); expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 5}))).toBe( 30, ); }); it('applies bounded symmetric jitter using the injected RNG', () => { - const cfg = {initialDelay: 4, backoffFactor: 1, jitter: 1}; + const cfg = prepareRetryConfig({ + initialDelay: 4, + backoffFactor: 1, + jitter: 1, + }); // randomFn=0.5 -> offset 0 -> exactly base delay (4) expect( getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 0.5), From 9aee4898c3f2f469906eb6256da95938de8638e6 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 18:43:50 -0700 Subject: [PATCH 4/5] docs(events): drop @link to internal isEvent in the Event brand comment typedoc `docs:check` runs with --treatWarningsAsErrors and flagged the [EVENT_SIGNATURE_SYMBOL] doc comment (inherited by Event, CreateEventParams and CompactedEvent) for linking to `isEvent`, which is not part of the exported/ documented API. Reference it as inline code instead of a doc link. No API or behavior change. --- core/src/events/event.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/events/event.ts b/core/src/events/event.ts index 0f716a417..9e20794e5 100644 --- a/core/src/events/event.ts +++ b/core/src/events/event.ts @@ -16,9 +16,9 @@ import {createEventActions, EventActions} from './event_actions.js'; * A unique symbol identifying ADK Event objects. * * Events are plain objects produced by {@link createEvent} rather than class - * instances, so they carry this signature as a brand and {@link isEvent} checks - * for it — mirroring the `Symbol.for('google.adk.*')` guards used across ADK - * (e.g. {@link isBaseTool}, {@link isBaseAgent}). + * instances, so they carry this signature as a brand and `isEvent` checks for + * it — mirroring the `Symbol.for('google.adk.*')` guards used across ADK (e.g. + * `isBaseTool`, `isBaseAgent`). */ const EVENT_SIGNATURE_SYMBOL = Symbol.for('google.adk.event'); @@ -64,7 +64,7 @@ export interface Event extends LlmResponse { /** * Signature brand identifying this object as an ADK {@link Event}. * - * Set by {@link createEvent} and checked by {@link isEvent}. Optional because + * Set by {@link createEvent} and checked by `isEvent`. Optional because * events are also reconstructed from storage/session payloads, where the * (non-serializable) brand is absent. */ From 3e927fd38e224ab1972f62eff8bfd96b4145f6c1 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 13:22:43 -0700 Subject: [PATCH 5/5] refactor(workflow): apply workflow conventions to part 1 primitives Aligns the Part 1 shared primitives with the workflow conventions applied across the stack. - branch_path: convert the BranchPath.fromString / BranchPath.commonPrefix static methods to plain functions (branchPathFromString / commonPrefixOfPaths); keep the instance methods. Internal callers (createSubBranch, commonPrefixOf) updated. - retry_utils: shouldRetryNode and getRetryDelaySeconds now take a single params object; errorName is duck-typed (no `instanceof Error`) so it works for error-like values across package copies. - Add branch_path unit coverage (previously untested) and update the retry_utils call sites in foundations_test to the params-object form. Error type guards keep the name-based `instanceof Error` form by design (the one sanctioned use of instanceof). --- core/src/workflow/branch_path.ts | 53 ++++++------ core/src/workflow/utils/retry_utils.ts | 69 ++++++++++------ core/test/workflow/branch_path_test.ts | 108 +++++++++++++++++++++++++ core/test/workflow/foundations_test.ts | 75 +++++++++++------ 4 files changed, 229 insertions(+), 76 deletions(-) create mode 100644 core/test/workflow/branch_path_test.ts diff --git a/core/src/workflow/branch_path.ts b/core/src/workflow/branch_path.ts index 978a1d24d..0a5d09ca8 100644 --- a/core/src/workflow/branch_path.ts +++ b/core/src/workflow/branch_path.ts @@ -19,14 +19,6 @@ export class BranchPath { this.segments = [...segments]; } - /** Parses a dot-separated string into a {@link BranchPath}. */ - static fromString(path?: string | null): BranchPath { - if (!path) { - return new BranchPath([]); - } - return new BranchPath(path.split('.')); - } - toString(): string { return this.segments.join('.'); } @@ -49,24 +41,33 @@ export class BranchPath { const segment = runId !== undefined ? `${name}@${runId}` : name; return new BranchPath([...this.segments, segment]); } +} - /** Finds the common prefix across a list of paths. */ - static commonPrefix(paths: BranchPath[]): BranchPath { - if (paths.length === 0) { - return new BranchPath([]); - } - const common: string[] = []; - const minLen = Math.min(...paths.map((p) => p.segments.length)); - for (let i = 0; i < minLen; i++) { - const seg = paths[0].segments[i]; - if (paths.every((p) => p.segments[i] === seg)) { - common.push(seg); - } else { - break; - } +/** Parses a dot-separated string into a {@link BranchPath}. */ +export function branchPathFromString(path?: string | null): BranchPath { + if (!path) { + return new BranchPath([]); + } + return new BranchPath(path.split('.')); +} + +/** Finds the common prefix across a list of {@link BranchPath}s. */ +export function commonPrefixOfPaths(paths: BranchPath[]): BranchPath { + if (paths.length === 0) { + return new BranchPath([]); + } + const allSegments = paths.map((p) => p.getSegments()); + const common: string[] = []; + const minLen = Math.min(...allSegments.map((s) => s.length)); + for (let i = 0; i < minLen; i++) { + const seg = allSegments[0][i]; + if (allSegments.every((s) => s[i] === seg)) { + common.push(seg); + } else { + break; } - return new BranchPath(common); } + return new BranchPath(common); } /** @@ -79,14 +80,14 @@ export function createSubBranch( baseBranch: string | undefined | null, options: {name: string; runId?: string}, ): string { - return BranchPath.fromString(baseBranch) + return branchPathFromString(baseBranch) .append(options.name, options.runId) .toString(); } /** Finds the common prefix of a list of dot-separated branch strings. */ export function commonPrefixOf(branches: string[]): string { - return BranchPath.commonPrefix( - branches.map((b) => BranchPath.fromString(b)), + return commonPrefixOfPaths( + branches.map((b) => branchPathFromString(b)), ).toString(); } diff --git a/core/src/workflow/utils/retry_utils.ts b/core/src/workflow/utils/retry_utils.ts index 18e1481a6..7afe9cbd9 100644 --- a/core/src/workflow/utils/retry_utils.ts +++ b/core/src/workflow/utils/retry_utils.ts @@ -22,31 +22,42 @@ const DEFAULT_JITTER = 1.0; /** * Resolves the runtime name of a thrown value for exception-name matching. * Mirrors Python's `type(exception).__name__`. + * + * Duck-typed (no `instanceof`) so it works for error-like values from another + * package copy: it prefers an explicit `name` string, then the constructor + * name, then the primitive `typeof`. */ function errorName(error: unknown): string { - if (error instanceof Error) { - // `name` is set by well-behaved Error subclasses; fall back to the - // constructor name for plain `throw new Error()` cases. - return error.name || error.constructor.name; - } if (typeof error === 'object' && error !== null) { - return error.constructor.name; + const named = error as {name?: unknown; constructor?: {name?: string}}; + if (typeof named.name === 'string' && named.name) { + // `name` is set by well-behaved Error subclasses. + return named.name; + } + // Fall back to the constructor name for plain `throw new Error()` cases. + return named.constructor?.name ?? 'Object'; } return typeof error; } +/** Parameters for {@link shouldRetryNode}. */ +export interface ShouldRetryNodeParams { + /** The error thrown by the node. */ + error: unknown; + /** The node's prepared (normalized) retry configuration. */ + retryConfig: PreparedRetryConfig; + /** The current node state (its `attemptCount` is 1-based). */ + nodeState: NodeState; +} + /** * Checks if a failed node should be retried based on its retry config. - * - * @param error The error thrown by the node. - * @param retryConfig The node's prepared (normalized) retry configuration. - * @param nodeState The current node state (its `attemptCount` is 1-based). */ -export function shouldRetryNode( - error: unknown, - retryConfig: PreparedRetryConfig, - nodeState: NodeState, -): boolean { +export function shouldRetryNode({ + error, + retryConfig, + nodeState, +}: ShouldRetryNodeParams): boolean { const attemptCount = nodeState.attemptCount; const maxAttempts = retryConfig.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; @@ -64,19 +75,27 @@ export function shouldRetryNode( return true; } +/** Parameters for {@link getRetryDelaySeconds}. */ +export interface GetRetryDelaySecondsParams { + /** The node's prepared (normalized) retry configuration. */ + retryConfig: PreparedRetryConfig; + /** + * The current node state (its `attemptCount` is the 1-based attempt number + * that just failed). + */ + nodeState: NodeState; + /** Injectable uniform RNG in [0, 1) for deterministic testing. */ + randomFn?: () => number; +} + /** * Calculates the delay, in seconds, before retrying a node. - * - * @param retryConfig The node's prepared (normalized) retry configuration. - * @param nodeState The current node state (its `attemptCount` is the 1-based - * attempt number that just failed). - * @param randomFn Injectable uniform RNG in [0, 1) for deterministic testing. */ -export function getRetryDelaySeconds( - retryConfig: PreparedRetryConfig, - nodeState: NodeState, - randomFn: () => number = Math.random, -): number { +export function getRetryDelaySeconds({ + retryConfig, + nodeState, + randomFn = Math.random, +}: GetRetryDelaySecondsParams): number { const initialDelay = retryConfig.initialDelay ?? DEFAULT_INITIAL_DELAY_SECONDS; const maxDelay = retryConfig.maxDelay ?? DEFAULT_MAX_DELAY_SECONDS; diff --git a/core/test/workflow/branch_path_test.ts b/core/test/workflow/branch_path_test.ts new file mode 100644 index 000000000..44f0c6ceb --- /dev/null +++ b/core/test/workflow/branch_path_test.ts @@ -0,0 +1,108 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import { + BranchPath, + branchPathFromString, + commonPrefixOf, + commonPrefixOfPaths, + createSubBranch, +} from '../../src/workflow/branch_path.js'; + +describe('branchPathFromString', () => { + it('parses undefined / null / empty into an empty path', () => { + expect(branchPathFromString(undefined).getSegments()).toEqual([]); + expect(branchPathFromString(null).getSegments()).toEqual([]); + expect(branchPathFromString('').getSegments()).toEqual([]); + }); + + it('splits a dotted string into segments', () => { + const path = branchPathFromString('a.b@1.c'); + expect(path.getSegments()).toEqual(['a', 'b@1', 'c']); + expect(path.toString()).toBe('a.b@1.c'); + }); +}); + +describe('BranchPath.append', () => { + it('appends a bare name segment', () => { + expect(new BranchPath([]).append('x').toString()).toBe('x'); + }); + + it('appends a name@runId segment when a runId is given', () => { + expect(new BranchPath(['a']).append('x', '1').toString()).toBe('a.x@1'); + }); + + it('does not mutate the original path', () => { + const base = new BranchPath(['a']); + base.append('b'); + expect(base.getSegments()).toEqual(['a']); + }); +}); + +describe('BranchPath.isDescendantOf', () => { + it('is true for a strict descendant', () => { + expect( + branchPathFromString('a.b.c').isDescendantOf(branchPathFromString('a.b')), + ).toBe(true); + }); + + it('is false for an identical path (strict)', () => { + expect( + branchPathFromString('a.b').isDescendantOf(branchPathFromString('a.b')), + ).toBe(false); + }); + + it('is false for a divergent path', () => { + expect( + branchPathFromString('a.c').isDescendantOf(branchPathFromString('a.b')), + ).toBe(false); + }); +}); + +describe('commonPrefixOfPaths', () => { + it('returns an empty path for no inputs', () => { + expect(commonPrefixOfPaths([]).getSegments()).toEqual([]); + }); + + it('returns the shared prefix', () => { + const prefix = commonPrefixOfPaths([ + branchPathFromString('a.b.c'), + branchPathFromString('a.b.d'), + ]); + expect(prefix.toString()).toBe('a.b'); + }); + + it('returns empty when there is no shared prefix', () => { + const prefix = commonPrefixOfPaths([ + branchPathFromString('a.b'), + branchPathFromString('x.y'), + ]); + expect(prefix.toString()).toBe(''); + }); +}); + +describe('createSubBranch', () => { + it('creates a root segment from an empty base', () => { + expect(createSubBranch(undefined, {name: 'agent'})).toBe('agent'); + }); + + it('appends a name@runId segment to an existing branch', () => { + expect(createSubBranch('parent', {name: 'child', runId: '1'})).toBe( + 'parent.child@1', + ); + }); +}); + +describe('commonPrefixOf', () => { + it('finds the common prefix of dotted branch strings', () => { + expect(commonPrefixOf(['a.b.c', 'a.b.d'])).toBe('a.b'); + }); + + it('returns an empty string for no inputs', () => { + expect(commonPrefixOf([])).toBe(''); + }); +}); diff --git a/core/test/workflow/foundations_test.ts b/core/test/workflow/foundations_test.ts index 1189f7fbe..195ba0869 100644 --- a/core/test/workflow/foundations_test.ts +++ b/core/test/workflow/foundations_test.ts @@ -128,21 +128,37 @@ describe('Phase 0 — shouldRetryNode', () => { it('retries until maxAttempts is reached (default 5)', () => { const cfg = prepareRetryConfig({}); - expect(shouldRetryNode(new Error('x'), cfg, state(1))).toBe(true); - expect(shouldRetryNode(new Error('x'), cfg, state(4))).toBe(true); - expect(shouldRetryNode(new Error('x'), cfg, state(5))).toBe(false); + const check = (nodeState: NodeState) => + shouldRetryNode({error: new Error('x'), retryConfig: cfg, nodeState}); + expect(check(state(1))).toBe(true); + expect(check(state(4))).toBe(true); + expect(check(state(5))).toBe(false); }); it('respects an explicit maxAttempts', () => { const cfg = prepareRetryConfig({maxAttempts: 2}); - expect(shouldRetryNode(new Error('x'), cfg, state(1))).toBe(true); - expect(shouldRetryNode(new Error('x'), cfg, state(2))).toBe(false); + const check = (nodeState: NodeState) => + shouldRetryNode({error: new Error('x'), retryConfig: cfg, nodeState}); + expect(check(state(1))).toBe(true); + expect(check(state(2))).toBe(false); }); it('only retries listed exception types when provided', () => { const cfg = prepareRetryConfig({exceptions: [TypeError]}); - expect(shouldRetryNode(new TypeError('x'), cfg, state(1))).toBe(true); - expect(shouldRetryNode(new RangeError('x'), cfg, state(1))).toBe(false); + expect( + shouldRetryNode({ + error: new TypeError('x'), + retryConfig: cfg, + nodeState: state(1), + }), + ).toBe(true); + expect( + shouldRetryNode({ + error: new RangeError('x'), + retryConfig: cfg, + nodeState: state(1), + }), + ).toBe(false); }); }); @@ -154,13 +170,19 @@ describe('Phase 0 — getRetryDelaySeconds', () => { jitter: 0, }); // attempt 1 -> exponent 0 -> 1s - expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}))).toBe( - 1, - ); + expect( + getRetryDelaySeconds({ + retryConfig: cfg, + nodeState: createNodeState({attemptCount: 1}), + }), + ).toBe(1); // attempt 3 -> exponent 2 -> 4s - expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 3}))).toBe( - 4, - ); + expect( + getRetryDelaySeconds({ + retryConfig: cfg, + nodeState: createNodeState({attemptCount: 3}), + }), + ).toBe(4); }); it('caps delay at maxDelay', () => { @@ -170,9 +192,12 @@ describe('Phase 0 — getRetryDelaySeconds', () => { maxDelay: 30, jitter: 0, }); - expect(getRetryDelaySeconds(cfg, createNodeState({attemptCount: 5}))).toBe( - 30, - ); + expect( + getRetryDelaySeconds({ + retryConfig: cfg, + nodeState: createNodeState({attemptCount: 5}), + }), + ).toBe(30); }); it('applies bounded symmetric jitter using the injected RNG', () => { @@ -181,17 +206,17 @@ describe('Phase 0 — getRetryDelaySeconds', () => { backoffFactor: 1, jitter: 1, }); + const delayWith = (randomFn: () => number) => + getRetryDelaySeconds({ + retryConfig: cfg, + nodeState: createNodeState({attemptCount: 1}), + randomFn, + }); // randomFn=0.5 -> offset 0 -> exactly base delay (4) - expect( - getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 0.5), - ).toBe(4); + expect(delayWith(() => 0.5)).toBe(4); // randomFn=0 -> offset -span -> max(0, 4-4)=0 - expect( - getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 0), - ).toBe(0); + expect(delayWith(() => 0)).toBe(0); // randomFn=1 -> offset +span -> 4+4=8 - expect( - getRetryDelaySeconds(cfg, createNodeState({attemptCount: 1}), () => 1), - ).toBe(8); + expect(delayWith(() => 1)).toBe(8); }); });