feat(workflow): event model extensions and shared node primitives (Part 1)

core/src/common.ts

@@ -143,7 +143,13 @@ export {
   pruneThoughts,
   stringifyContent,
 } from './events/event.js';
-export type {Event} 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';

core/src/events/event.ts

@@ -12,13 +12,64 @@ 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 `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');
+
+/**
+ * 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.
+ *
+ * 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.
 
   It is used to store the content of the conversation, as well as the actions
   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 `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.
@@ -63,6 +114,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?: Route;
+
+  /**
+   * 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<Partial<Event>, 'actions'> {
+  actions?: Partial<EventActions>;
 }
 
 /**
@@ -71,13 +157,14 @@ export interface Event extends LlmResponse {
  * @param params The partial event to create the event from.
  * @returns The event.
  */
-export function createEvent(params: Partial<Event> = {}): Event {
+export function createEvent(params: CreateEventParams = {}): Event {
   return {
     ...params,
+    [EVENT_SIGNATURE_SYMBOL]: true,
     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 +316,22 @@ 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 &&
+    EVENT_SIGNATURE_SYMBOL in obj &&
+    (obj as {[EVENT_SIGNATURE_SYMBOL]?: unknown})[EVENT_SIGNATURE_SYMBOL] ===
+      true
+  );
+}
+
 /**
  * Generates a new unique ID for the event.
  */
@@ -262,6 +365,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 +389,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',
 ];
 
 /**

core/src/events/event_actions.ts

@@ -56,6 +56,18 @@ export interface EventActions {
    * call id.
    */
   requestedToolConfirmations: {[key: string]: ToolConfirmation};
+
+  /**
+   * Workflow: a serialized node/agent state snapshot used for resumable
+   * checkpointing. Mirrors Python `EventActions.agent_state`.
+   */
+  agentState?: Record<string, unknown>;
+
+  /**
+   * 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;
 }
 
 /**

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<T> implements AsyncIterable<T> {
   private queue: T[] = [];
@@ -14,45 +27,72 @@ export class AsyncQueue<T> implements AsyncIterable<T> {
     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<T> {
     return {
-      next: () => {
-        if (this.errorVal) {
-          const err = this.errorVal;
-          this.errorVal = undefined;
-          return Promise.reject(err);
-        }
+      next: (): Promise<IteratorResult<T>> => {
         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});
         }