From 21a168543f38e0edf91f5c9a6affc79e023fd21d Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Thu, 30 Jul 2026 19:35:10 -0700 Subject: [PATCH 1/6] feat(workflow): add LLM-agent-as-node, task mode, and node-as-tool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 6/9 of the feature/workflows split. Lets agents participate in workflows. - nodes/llm_agent_wrapper: runs a BaseAgent as a workflow node — streaming, transfer_to_agent hand-offs, workflow instruction scope, and task mode (loops until the agent calls finish_task, whose args become the node output). It registers the agent node-builder, explicitly excluding BaseTool (which also exposes runAsync) so tool-before-agent precedence holds regardless of registration order. - nodes/node_tool: exposes a node/workflow as a tool an agent can call. - tools/finish_task_tool: the finish_task tool backing task mode. - agents/llm_agent: task mode (mode/finishTaskTool) and registers the request-input + request-confirmation LLM request processors. - agents/processors/request_input_llm_request_processor: agent-side HITL (request user input mid-run). - agents/{invocation_context,instructions}, basic_llm_request_processor: workflow instruction scope and {Class.field}/ placeholder resolution. Wired into the public barrel (LLMAgentWrapper, NodeTool) and register_builtin_nodes; llm_agent 3-way merged onto current main. Tests: node_api (16), llm_agent (7), multi_agent (3), instructions (37), plus the workflow integration suite (15 files / 35 tests, recorded model responses). Full core suite green (2476), integration workflows green, docs:check + tsc clean. --- core/src/agents/instructions.ts | 68 +++- core/src/agents/invocation_context.ts | 48 +++ core/src/agents/llm_agent.ts | 84 ++++- .../processors/basic_llm_request_processor.ts | 4 + .../request_input_llm_request_processor.ts | 205 +++++++++++ core/src/common.ts | 5 + core/src/tools/finish_task_tool.ts | 140 ++++++++ core/src/workflow/index.ts | 4 +- core/src/workflow/nodes/llm_agent_wrapper.ts | 321 ++++++++++++++++++ core/src/workflow/nodes/node_tool.ts | 143 ++++++++ core/test/agents/instructions_test.ts | 61 ++++ core/test/workflow/llm_agent_test.ts | 268 +++++++++++++++ core/test/workflow/multi_agent_test.ts | 150 ++++++++ core/test/workflow/node_api_test.ts | 231 +++++++++++++ .../workflows/advanced_workflows_test.ts | 170 ++++++++++ .../agent_pipeline.model_responses.json | 41 +++ .../workflows/agent_pipeline_test.ts | 81 +++++ .../workflows/auth_workflow_test.ts | 104 ++++++ .../workflows/core_workflows_test.ts | 306 +++++++++++++++++ .../workflows/llm_loop.model_responses.json | 28 ++ tests/integration/workflows/llm_loop_test.ts | 62 ++++ .../llm_tool_agent.model_responses.json | 34 ++ .../workflows/llm_tool_agent_test.ts | 74 ++++ .../workflows/loop_and_trigger_test.ts | 113 ++++++ .../workflows/node_as_tool_hitl_test.ts | 121 +++++++ .../workflows/node_as_tool_test.ts | 80 +++++ .../parallel_llm.model_responses.json | 28 ++ .../workflows/parallel_llm_test.ts | 51 +++ .../workflows/plain_text_resume_test.ts | 60 ++++ .../workflows/route.model_responses.json | 28 ++ tests/integration/workflows/route_llm_test.ts | 106 ++++++ .../workflows/sequence.model_responses.json | 25 ++ .../workflows/sequence_llm_test.ts | 60 ++++ tests/integration/workflows/task_mode_test.ts | 69 ++++ .../workflows/tool_and_resilience_test.ts | 119 +++++++ .../workflows/workflow_test_utils.ts | 125 +++++++ typedoc.json | 3 +- 37 files changed, 3599 insertions(+), 21 deletions(-) create mode 100644 core/src/agents/processors/request_input_llm_request_processor.ts create mode 100644 core/src/tools/finish_task_tool.ts create mode 100644 core/src/workflow/nodes/llm_agent_wrapper.ts create mode 100644 core/src/workflow/nodes/node_tool.ts create mode 100644 core/test/workflow/llm_agent_test.ts create mode 100644 core/test/workflow/multi_agent_test.ts create mode 100644 core/test/workflow/node_api_test.ts create mode 100644 tests/integration/workflows/advanced_workflows_test.ts create mode 100644 tests/integration/workflows/agent_pipeline.model_responses.json create mode 100644 tests/integration/workflows/agent_pipeline_test.ts create mode 100644 tests/integration/workflows/auth_workflow_test.ts create mode 100644 tests/integration/workflows/core_workflows_test.ts create mode 100644 tests/integration/workflows/llm_loop.model_responses.json create mode 100644 tests/integration/workflows/llm_loop_test.ts create mode 100644 tests/integration/workflows/llm_tool_agent.model_responses.json create mode 100644 tests/integration/workflows/llm_tool_agent_test.ts create mode 100644 tests/integration/workflows/loop_and_trigger_test.ts create mode 100644 tests/integration/workflows/node_as_tool_hitl_test.ts create mode 100644 tests/integration/workflows/node_as_tool_test.ts create mode 100644 tests/integration/workflows/parallel_llm.model_responses.json create mode 100644 tests/integration/workflows/parallel_llm_test.ts create mode 100644 tests/integration/workflows/plain_text_resume_test.ts create mode 100644 tests/integration/workflows/route.model_responses.json create mode 100644 tests/integration/workflows/route_llm_test.ts create mode 100644 tests/integration/workflows/sequence.model_responses.json create mode 100644 tests/integration/workflows/sequence_llm_test.ts create mode 100644 tests/integration/workflows/task_mode_test.ts create mode 100644 tests/integration/workflows/tool_and_resilience_test.ts create mode 100644 tests/integration/workflows/workflow_test_utils.ts diff --git a/core/src/agents/instructions.ts b/core/src/agents/instructions.ts index 093586d3a..1f907a381 100644 --- a/core/src/agents/instructions.ts +++ b/core/src/agents/instructions.ts @@ -5,10 +5,36 @@ */ import {State} from '../sessions/state.js'; +import type {WorkflowInstructionScope} from './invocation_context.js'; import {ReadonlyContext} from './readonly_context.js'; const ARTIFACT_PREFIX = 'artifact.'; +/** Matches a `{Class.field}` workflow placeholder key (dotted identifier pair). */ +const WORKFLOW_FIELD_KEY = /^[A-Za-z_]\w*\.[A-Za-z_]\w*$/; + +/** Matches a `` workflow placeholder. */ +const SOURCE_NODE_PLACEHOLDER = + /<\s*[A-Za-z_]\w*\.([A-Za-z_]\w*)\s+from\s+([A-Za-z_]\w*)\s*>/g; + +/** + * Resolves `` placeholders against a workflow + * scope (predecessor outputs by node name). Synchronous; unresolved placeholders + * are left untouched. Mirrors Python's source-node-qualified data selection. + */ +function resolveSourceNodePlaceholders( + template: string, + scope: WorkflowInstructionScope, +): string { + return template.replace(SOURCE_NODE_PLACEHOLDER, (raw, field, nodeName) => { + const out = scope.outputsByNode?.[nodeName]; + if (out && typeof out === 'object' && field in (out as object)) { + return formatValue((out as Record)[field], false); + } + return raw; + }); +} + /** * Resolves a single key from the context (state or artifact). */ @@ -39,19 +65,30 @@ async function resolveKey( } // Step 3: Handle state variable injection. - if (!isValidStateName(key)) { - return rawMatch; - } - - if (key in invocationContext.session.state) { - return formatValue(invocationContext.session.state[key], false); + if (isValidStateName(key)) { + if (key in invocationContext.session.state) { + return formatValue(invocationContext.session.state[key], false); + } + if (isOptional) { + return ''; + } + throw new Error(`Context variable not found: \`${key}\`.`); } - if (isOptional) { - return ''; + // Step 4: Workflow — resolve `{Class.field}` from the current node input. + const scope = invocationContext.workflowInstructionScope; + if (scope && WORKFLOW_FIELD_KEY.test(key)) { + const field = key.slice(key.indexOf('.') + 1); + const input = scope.input; + if (input && typeof input === 'object' && field in (input as object)) { + return formatValue((input as Record)[field], false); + } + if (isOptional) { + return ''; + } } - throw new Error(`Context variable not found: \`${key}\`.`); + return rawMatch; } /** @@ -115,6 +152,14 @@ export async function injectSessionState( template: string, readonlyContext: ReadonlyContext, ): Promise { + // Workflow: first resolve `` placeholders, and + // enable `{Class.field}` resolution below. Both are no-ops (placeholders left + // untouched) for ordinary agents, which have no workflow scope. + const scope = readonlyContext.invocationContext.workflowInstructionScope; + if (scope) { + template = resolveSourceNodePlaceholders(template, scope); + } + const pattern = /\{+[^{}]*}+/g; const matches = Array.from(template.matchAll(pattern)); @@ -130,7 +175,10 @@ export async function injectSessionState( if (isOptional) { key = key.slice(0, -1); } - const isValid = key.startsWith(ARTIFACT_PREFIX) || isValidStateName(key); + const isValid = + key.startsWith(ARTIFACT_PREFIX) || + isValidStateName(key) || + (!!scope && WORKFLOW_FIELD_KEY.test(key)); return { raw, key, diff --git a/core/src/agents/invocation_context.ts b/core/src/agents/invocation_context.ts index 8a6c4e19d..d09c5f7aa 100644 --- a/core/src/agents/invocation_context.ts +++ b/core/src/agents/invocation_context.ts @@ -8,10 +8,12 @@ import {Content} from '@google/genai'; import {SessionArtifactService} from '../artifacts/session_artifact_service.js'; import {BaseCredentialService} from '../auth/credential_service/base_credential_service.js'; +import {Event} from '../events/event.js'; import {BaseMemoryService} from '../memory/base_memory_service.js'; import {PluginManager} from '../plugins/plugin_manager.js'; import {BaseSessionService} from '../sessions/base_session_service.js'; import {Session} from '../sessions/session.js'; +import {AsyncQueue} from '../utils/async_queue.js'; import {randomUUID} from '../utils/env_aware_utils.js'; import {ActiveStreamingTool} from './active_streaming_tool.js'; @@ -19,6 +21,19 @@ import {BaseAgent} from './base_agent.js'; import {RunConfig} from './run_config.js'; import {TranscriptionEntry} from './transcription_entry.js'; +/** + * Workflow: data exposed to `{Class.field}` and `` + * instruction placeholders when an LlmAgent runs as a workflow node. Populated by + * `LLMAgentWrapper`; absent for ordinary (non-workflow) agent runs, in which case + * those placeholders are left untouched. + */ +export interface WorkflowInstructionScope { + /** The current node's input, exposing fields for `{Class.field}`. */ + input?: unknown; + /** Predecessor node outputs keyed by node name, for ``. */ + outputsByNode?: Record; +} + /** * The parameters for creating an invocation context. */ @@ -38,6 +53,9 @@ export interface InvocationContextParams { activeStreamingTools?: Record; pluginManager: PluginManager; abortSignal?: AbortSignal; + agentStates?: Record; + endOfAgents?: Record; + workflowInstructionScope?: WorkflowInstructionScope; } /** @@ -185,6 +203,32 @@ export class InvocationContext { readonly abortSignal?: AbortSignal; + /** + * An optional channel into which a running tool can push events to be + * interleaved into the agent's output stream. Set by the LLM flow around tool + * execution so a {@link NodeTool} (running a node/workflow) can surface the + * node's intermediate and interrupt events. Cleared once tools finish. + */ + eventQueue?: AsyncQueue; + + /** + * Checkpointed states for workflow nodes under this invocation. + */ + agentStates: Record; + + /** + * Tracks whether specific agents or workflows have reached the end of their execution. + */ + + endOfAgents: Record; + + /** + * Workflow: field-resolution scope for `{Class.field}` / + * `` instruction placeholders (set by + * `LLMAgentWrapper`). + */ + workflowInstructionScope?: WorkflowInstructionScope; + /** * @param params The parameters for creating an invocation context. */ @@ -203,7 +247,11 @@ export class InvocationContext { this.activeStreamingTools = params.activeStreamingTools; this.pluginManager = params.pluginManager; this.abortSignal = params.abortSignal; + this.agentStates = params.agentStates ?? {}; + this.endOfAgents = params.endOfAgents ?? {}; + this.workflowInstructionScope = params.workflowInstructionScope; // Inherit the parent invocation's cost manager when one is available. + // Child contexts created for sub-agents, agent transfers and loop // iterations (via createInvocationContext / createBranchCtxForSubAgent) // carry the parent context's fields over, so reusing its cost manager diff --git a/core/src/agents/llm_agent.ts b/core/src/agents/llm_agent.ts index 330629b23..ab1828caf 100644 --- a/core/src/agents/llm_agent.ts +++ b/core/src/agents/llm_agent.ts @@ -6,7 +6,11 @@ import {GenerateContentConfig, Schema} from '@google/genai'; import {context, trace} from '@opentelemetry/api'; +import {FinishTaskTool} from '../tools/finish_task_tool.js'; import {FunctionTool} from '../tools/function_tool.js'; +import {AsyncQueue} from '../utils/async_queue.js'; +import {BaseNode} from '../workflow/base_node.js'; +import {NodeTool} from '../workflow/nodes/node_tool.js'; import {z as z3} from 'zod/v3'; import {z as z4} from 'zod/v4'; @@ -68,6 +72,7 @@ import {IDENTITY_LLM_REQUEST_PROCESSOR} from './processors/identity_llm_request_ import {INSTRUCTIONS_LLM_REQUEST_PROCESSOR} from './processors/instructions_llm_request_processor.js'; import {INTERACTIONS_REQUEST_PROCESSOR} from './processors/interactions_request_processor.js'; import {REQUEST_CONFIRMATION_LLM_REQUEST_PROCESSOR} from './processors/request_confirmation_llm_request_processor.js'; +import {REQUEST_INPUT_LLM_REQUEST_PROCESSOR} from './processors/request_input_llm_request_processor.js'; import {TOOL_FILTER_REQUEST_PROCESSOR} from './processors/tool_filter_request_processor.js'; import {ReadonlyContext} from './readonly_context.js'; import {StreamingMode} from './run_config.js'; @@ -193,7 +198,7 @@ export type AfterToolCallback = export type ExamplesUnion = Example[] | BaseExampleProvider; /** A union of tool types that can be provided to an agent. */ -export type ToolUnion = BaseTool | BaseToolset; +export type ToolUnion = BaseTool | BaseToolset | BaseNode; const ADK_AGENT_NAME_LABEL_KEY = 'adk_agent_name'; @@ -259,6 +264,16 @@ export interface LlmAgentConfig extends BaseAgentConfig { */ includeContents?: 'default' | 'none'; + /** + * The agent's execution mode when run as a workflow node. + * + * - `single_turn` (default): the agent runs once against the node input. + * - `task`: the agent is given a `finish_task` tool and runs a multi-round + * loop until it calls `finish_task`, whose arguments (conforming to + * `outputSchema`) become the node output. Mirrors Python's `Agent(mode=...)`. + */ + mode?: 'single_turn' | 'task'; + /** The input schema when agent is used as a tool. */ inputSchema?: LlmAgentSchema; @@ -323,6 +338,11 @@ async function convertToolUnionToTools( if (isBaseTool(toolUnion)) { return [toolUnion]; } + if (toolUnion instanceof BaseNode) { + // A node/Workflow passed as a tool is auto-wrapped as a NodeTool so the + // model can call it (mirrors Python's Agent(tools=[node/workflow])). + return [new NodeTool(toolUnion)]; + } return await toolUnion.getTools(context); } @@ -362,9 +382,11 @@ export class LlmAgent extends BaseAgent { disallowTransferToParent: boolean; disallowTransferToPeers: boolean; includeContents: 'default' | 'none'; + mode?: 'single_turn' | 'task'; inputSchema?: Schema; outputSchema?: Schema; outputKey?: string; + private _finishTaskTool?: FinishTaskTool; beforeModelCallback?: BeforeModelCallback; afterModelCallback?: AfterModelCallback; beforeToolCallback?: BeforeToolCallback; @@ -389,6 +411,7 @@ export class LlmAgent extends BaseAgent { this.outputSchema = isZodObject(config.outputSchema) ? zodObjectToSchema(config.outputSchema) : config.outputSchema; + this.mode = config.mode; this.outputKey = config.outputKey; this.beforeModelCallback = config.beforeModelCallback; this.afterModelCallback = config.afterModelCallback; @@ -404,6 +427,7 @@ export class LlmAgent extends BaseAgent { IDENTITY_LLM_REQUEST_PROCESSOR, INSTRUCTIONS_LLM_REQUEST_PROCESSOR, REQUEST_CONFIRMATION_LLM_REQUEST_PROCESSOR, + REQUEST_INPUT_LLM_REQUEST_PROCESSOR, CONTENT_REQUEST_PROCESSOR, INTERACTIONS_REQUEST_PROCESSOR, CODE_EXECUTION_REQUEST_PROCESSOR, @@ -500,6 +524,17 @@ export class LlmAgent extends BaseAgent { throw new Error(`No model found for ${this.name}.`); } + /** + * The `finish_task` tool for this agent (task mode). Lazily created and cached + * so its declaration (derived from `outputSchema`) is stable across turns. + */ + get finishTaskTool(): FinishTaskTool { + if (!this._finishTaskTool) { + this._finishTaskTool = new FinishTaskTool(this.outputSchema); + } + return this._finishTaskTool; + } + /** * The resolved instruction field to construct instruction for this * agent. @@ -788,7 +823,11 @@ export class LlmAgent extends BaseAgent { // TODO - b/425992518: check if tool preprocessors can be simplified. // Run pre-processors for tools. const allTools = [...this.tools]; - if ( + if (this.mode === 'task') { + // Task mode: the agent completes by calling `finish_task` (whose params + // mirror the output schema) rather than emitting structured output. + allTools.push(this.finishTaskTool); + } else if ( this.outputSchema && allTools.length > 0 && !canUseOutputSchemaWithTools(this.canonicalModel.model) @@ -978,13 +1017,40 @@ export class LlmAgent extends BaseAgent { // Call functions // TODO - b/425992518: bloated funciton input, fix. // Tool callback passed to get rid of cyclic dependency. - const functionResponseEvent = await handleFunctionCallsAsync({ - invocationContext: invocationContext, - functionCallEvent: mergedEvent, - toolsDict: llmRequest.toolsDict, - beforeToolCallbacks: this.canonicalBeforeToolCallbacks, - afterToolCallbacks: this.canonicalAfterToolCallbacks, - }); + // A NodeTool (running a node/workflow) streams the node's intermediate and + // interrupt events into `invocationContext.eventQueue`; drain it concurrently + // so those events interleave into this agent's output stream. The tool runs + // in a self-contained task that captures its result/error and always closes + // the queue, so there is a single error path (no unhandled rejection). + const eventQueue = new AsyncQueue(); + invocationContext.eventQueue = eventQueue; + const toolTask = (async (): Promise<{ + event: Event | null; + error?: unknown; + }> => { + try { + const event = await handleFunctionCallsAsync({ + invocationContext: invocationContext, + functionCallEvent: mergedEvent, + toolsDict: llmRequest.toolsDict, + beforeToolCallbacks: this.canonicalBeforeToolCallbacks, + afterToolCallbacks: this.canonicalAfterToolCallbacks, + }); + return {event}; + } catch (error) { + return {event: null, error}; + } finally { + eventQueue.close(); + } + })(); + for await (const queuedEvent of eventQueue) { + yield queuedEvent; + } + const {event: functionResponseEvent, error: toolError} = await toolTask; + invocationContext.eventQueue = undefined; + if (toolError) { + throw toolError; + } if (!functionResponseEvent || invocationContext.abortSignal?.aborted) { return; diff --git a/core/src/agents/processors/basic_llm_request_processor.ts b/core/src/agents/processors/basic_llm_request_processor.ts index 7ed2c66c5..76fcdb12f 100644 --- a/core/src/agents/processors/basic_llm_request_processor.ts +++ b/core/src/agents/processors/basic_llm_request_processor.ts @@ -41,8 +41,12 @@ export class BasicLlmRequestProcessor extends BaseLlmRequestProcessor { // Models that cannot take an output schema alongside tools get the // prompt-based `set_model_response` workaround instead, injected by // `LlmAgent.runOneStepAsync` and the instructions processor. + // Task-mode agents complete via the `finish_task` tool, so the JSON response + // mode must not be set (function calling is incompatible with a JSON + // response mime type). if ( agent.outputSchema && + agent.mode !== 'task' && (!agent.tools?.length || canUseOutputSchemaWithTools(agent.canonicalModel.model)) ) { diff --git a/core/src/agents/processors/request_input_llm_request_processor.ts b/core/src/agents/processors/request_input_llm_request_processor.ts new file mode 100644 index 000000000..aca84e899 --- /dev/null +++ b/core/src/agents/processors/request_input_llm_request_processor.ts @@ -0,0 +1,205 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {FunctionCall} from '@google/genai'; + +import { + Event, + getFunctionCalls, + getFunctionResponses, +} from '../../events/event.js'; +import {ToolConfirmation} from '../../tools/tool_confirmation.js'; +import {AsyncQueue} from '../../utils/async_queue.js'; +import {NodeTool} from '../../workflow/nodes/node_tool.js'; +import {REQUEST_INPUT_FUNCTION_CALL_NAME} from '../../workflow/utils/hitl_utils.js'; +import {unwrapResponse} from '../../workflow/utils/rehydration_utils.js'; +import {handleFunctionCallList} from '../functions.js'; +import {InvocationContext} from '../invocation_context.js'; +import {isLlmAgent} from '../llm_agent.js'; +import {ReadonlyContext} from '../readonly_context.js'; +import {BaseLlmRequestProcessor} from './base_llm_processor.js'; + +/** + * Resumes a {@link NodeTool} call that paused for input. When a node/workflow + * run inside a node-tool raises a `RequestInput` interrupt, the node-tool's + * function call is left pending (no response) and the invocation pauses. On the + * next turn, once the user answers the `adk_request_input` interrupt, this + * processor re-runs the pending node-tool with the answer(s) threaded as + * `resumeInputs`, then emits the tool's function response so the agent can + * continue. Analogous to {@link RequestConfirmationLlmRequestProcessor}. + */ +export class RequestInputLlmRequestProcessor extends BaseLlmRequestProcessor { + override async *runAsync( + invocationContext: InvocationContext, + ): AsyncGenerator { + const agent = invocationContext.agent; + if (!isLlmAgent(agent)) { + return; + } + const events = invocationContext.session.events; + if (!events || events.length === 0) { + return; + } + + // 1. Collect resume inputs (interruptId -> value): prefer structured + // `adk_request_input` responses, else map a plain-text reply to any + // pending interrupt (so an interactive client can resume by typing). + const resumeInputs = collectResumeInputs(events); + if (Object.keys(resumeInputs).length === 0) { + return; + } + + // 2. Resolve the agent's node-tools. + const toolsList = await agent.canonicalTools( + new ReadonlyContext(invocationContext), + ); + const toolsDict = Object.fromEntries(toolsList.map((t) => [t.name, t])); + const nodeToolNames = new Set( + toolsList.filter((t) => t instanceof NodeTool).map((t) => t.name), + ); + if (nodeToolNames.size === 0) { + return; + } + + // 3. Find pending node-tool function calls (raised but not yet answered). + const answeredIds = new Set(); + for (const event of events) { + for (const fr of getFunctionResponses(event)) { + if (fr.id) { + answeredIds.add(fr.id); + } + } + } + const pending: Record = {}; + for (const event of events) { + for (const fc of getFunctionCalls(event)) { + if ( + fc.id && + fc.name && + nodeToolNames.has(fc.name) && + !answeredIds.has(fc.id) + ) { + pending[fc.id] = fc; + } + } + } + if (Object.keys(pending).length === 0) { + return; + } + + // 4. Re-run each pending node-tool, threading the resume inputs through the + // tool confirmation payload (read by NodeTool as the node's resumeInputs). + const toolConfirmationDict: Record = {}; + for (const id of Object.keys(pending)) { + toolConfirmationDict[id] = new ToolConfirmation({ + confirmed: true, + payload: resumeInputs, + }); + } + + const eventQueue = new AsyncQueue(); + invocationContext.eventQueue = eventQueue; + const task = (async (): Promise => { + try { + return await handleFunctionCallList({ + invocationContext, + functionCalls: Object.values(pending), + toolsDict, + beforeToolCallbacks: agent.canonicalBeforeToolCallbacks, + afterToolCallbacks: agent.canonicalAfterToolCallbacks, + filters: new Set(Object.keys(pending)), + toolConfirmationDict, + }); + } finally { + eventQueue.close(); + } + })(); + for await (const queuedEvent of eventQueue) { + yield queuedEvent; + } + const functionResponseEvent = await task; + invocationContext.eventQueue = undefined; + if (functionResponseEvent) { + yield functionResponseEvent; + } + } +} + +/** + * Collects resume inputs from the session: structured `adk_request_input` + * function responses take precedence; otherwise a plain-text reply is mapped to + * every still-pending interrupt id. + */ +function collectResumeInputs(events: Event[]): Record { + for (let i = events.length - 1; i >= 0; i--) { + const event = events[i]; + if (event.author !== 'user') { + continue; + } + const structured: Record = {}; + let found = false; + for (const fr of getFunctionResponses(event)) { + if (fr.name === REQUEST_INPUT_FUNCTION_CALL_NAME && fr.id) { + structured[fr.id] = unwrapResponse(fr.response); + found = true; + } + } + if (found) { + return structured; + } + } + + // Plain-text fallback: map the latest plain-text user turn to pending + // interrupts (mirrors WorkflowAgent's interactive resume). + const pending = pendingInterruptIds(events); + if (pending.size === 0) { + return {}; + } + const lastUser = [...events].reverse().find((e) => e.author === 'user'); + const parts = lastUser?.content?.parts ?? []; + const isPlainText = + parts.length > 0 && parts.every((p) => typeof p.text === 'string'); + if (!isPlainText) { + return {}; + } + const text = parts.map((p) => p.text).join(''); + const inputs: Record = {}; + for (const id of pending) { + inputs[id] = text; + } + return inputs; +} + +/** Interrupt ids raised via `adk_request_input` that have no user response. */ +function pendingInterruptIds(events: Event[]): Set { + const answered = new Set(); + for (const event of events) { + if (event.author !== 'user') { + continue; + } + for (const fr of getFunctionResponses(event)) { + if (fr.id) { + answered.add(fr.id); + } + } + } + const pending = new Set(); + for (const event of events) { + for (const fc of getFunctionCalls(event)) { + if ( + fc.name === REQUEST_INPUT_FUNCTION_CALL_NAME && + fc.id && + !answered.has(fc.id) + ) { + pending.add(fc.id); + } + } + } + return pending; +} + +export const REQUEST_INPUT_LLM_REQUEST_PROCESSOR = + new RequestInputLlmRequestProcessor(); diff --git a/core/src/common.ts b/core/src/common.ts index f5fb03658..a879e1f54 100644 --- a/core/src/common.ts +++ b/core/src/common.ts @@ -248,6 +248,11 @@ export { } from './tools/enterprise_web_search_tool.js'; export {ExampleTool} from './tools/example_tool.js'; export {EXIT_LOOP, ExitLoopTool} from './tools/exit_loop_tool.js'; +export { + FINISH_TASK_SUCCESS_RESULT, + FINISH_TASK_TOOL_NAME, + FinishTaskTool, +} from './tools/finish_task_tool.js'; export {FunctionTool, isFunctionTool} from './tools/function_tool.js'; export type { ToolExecuteArgument, diff --git a/core/src/tools/finish_task_tool.ts b/core/src/tools/finish_task_tool.ts new file mode 100644 index 000000000..203992a0f --- /dev/null +++ b/core/src/tools/finish_task_tool.ts @@ -0,0 +1,140 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {FunctionDeclaration, Schema, Type} from '@google/genai'; + +import {appendInstructions} from '../models/llm_request.js'; +import { + BaseTool, + RunAsyncToolRequest, + ToolProcessLlmRequest, +} from './base_tool.js'; + +/** The name of the finish_task tool. */ +export const FINISH_TASK_TOOL_NAME = 'finish_task'; + +/** + * The result returned by {@link FinishTaskTool.runAsync} when validation passes. + * The task-mode wrapper uses this to distinguish a successful completion from a + * validation-error retry signal. + */ +export const FINISH_TASK_SUCCESS_RESULT = 'Task completed.'; + +/** The default output schema when the task agent declares none. */ +const DEFAULT_TASK_OUTPUT_SCHEMA: Schema = { + type: Type.OBJECT, + properties: { + result: { + type: Type.STRING, + description: 'A brief summary of what the agent accomplished.', + }, + }, + required: ['result'], +}; + +/** + * Tool for signaling that a task-mode {@link LlmAgent} has completed its task. + * + * The tool's parameters mirror the agent's `outputSchema` (or a default single + * `result` string). The task-mode wrapper sniffs the `finish_task` function call + * and, on a successful function response, promotes the call's arguments to the + * node's output. + * + * Ported from `google/adk-python` + * `agents/llm/task/_finish_task_tool.py::FinishTaskTool`. + */ +export class FinishTaskTool extends BaseTool { + /** The schema describing the expected task output. */ + private readonly outputSchema: Schema; + /** + * When the output schema is a non-object (primitive/array), the value is + * wrapped under this key (the GenAI API requires object-typed parameters). + * `undefined` for object schemas (the value lives at the top level of args). + */ + readonly wrapperKey?: string; + + constructor(outputSchema?: Schema) { + const schema = outputSchema ?? DEFAULT_TASK_OUTPUT_SCHEMA; + let description = + 'Signal that this agent has completed its delegated task. Call this' + + ' when you have finished your delegated task.'; + if (outputSchema) { + description += ' Pass the required output data in the parameters.'; + } + super({name: FINISH_TASK_TOOL_NAME, description}); + this.outputSchema = schema; + this.wrapperKey = schema.type === Type.OBJECT ? undefined : 'result'; + } + + override _getDeclaration(): FunctionDeclaration { + const parameters: Schema = this.wrapperKey + ? { + type: Type.OBJECT, + properties: {[this.wrapperKey]: this.outputSchema}, + required: [this.wrapperKey], + } + : this.outputSchema; + return {name: this.name, description: this.description, parameters}; + } + + override async processLlmRequest( + request: ToolProcessLlmRequest, + ): Promise { + await super.processLlmRequest(request); + // Tell the model when to call finish_task (mirrors Python's tool + // instruction), so it completes the task deliberately. + appendInstructions(request.llmRequest, [ + 'Do NOT call `finish_task` prematurely. Use your available tools to fully' + + ' complete every aspect of the task first. If the task is unclear, ask' + + ' the user for clarification before proceeding. Once the task is fully' + + ' complete, call `finish_task` by itself with no accompanying text' + + ' output.', + ]); + } + + /** + * Extracts the task output from a `finish_task` call's arguments, applying the + * wrapper-key unwrapping when the schema is a non-object. + */ + extractOutput(args: Record): unknown { + if (this.wrapperKey) { + return args[this.wrapperKey]; + } + return args; + } + + override async runAsync({args}: RunAsyncToolRequest): Promise { + const value = this.wrapperKey ? args[this.wrapperKey] : args; + const missing = this.missingRequiredKeys(value); + if (missing.length > 0) { + return { + error: + `Invoking \`${this.name}()\` failed due to missing required ` + + `parameters: ${missing.join(', ')}. You could retry calling this ` + + 'tool, but it is IMPORTANT for you to provide all the mandatory ' + + 'parameters with correct types.', + }; + } + return FINISH_TASK_SUCCESS_RESULT; + } + + /** Returns any `required` keys the schema declares that are absent. */ + private missingRequiredKeys(value: unknown): string[] { + const required = this.wrapperKey + ? value === undefined || value === null + ? [this.wrapperKey] + : [] + : (this.outputSchema.required ?? []); + if (this.wrapperKey) { + return required; + } + if (typeof value !== 'object' || value === null) { + return required; + } + const obj = value as Record; + return required.filter((key) => obj[key] === undefined); + } +} diff --git a/core/src/workflow/index.ts b/core/src/workflow/index.ts index ced77ea5b..47deb7414 100644 --- a/core/src/workflow/index.ts +++ b/core/src/workflow/index.ts @@ -29,12 +29,14 @@ export type { FunctionNodeResult, } from './nodes/function_node.js'; export {JoinNode} from './nodes/join_node.js'; +export {LLMAgentWrapper} from './nodes/llm_agent_wrapper.js'; +export type {LLMAgentWrapperConfig} from './nodes/llm_agent_wrapper.js'; +export {NodeTool} from './nodes/node_tool.js'; export {ParallelWorker} from './nodes/parallel_worker.js'; export type {ParallelWorkerConfig} from './nodes/parallel_worker.js'; export {ToolNode} from './nodes/tool_node.js'; export type {ToolNodeConfig} from './nodes/tool_node.js'; export type {BuildNodeOptions} from './utils/workflow_graph_utils.js'; -// LLMAgentWrapper and NodeTool are exported by Part 7 (LLM node). // --- Graph model --- export {DEFAULT_ROUTE, Edge, Graph} from './graph.js'; diff --git a/core/src/workflow/nodes/llm_agent_wrapper.ts b/core/src/workflow/nodes/llm_agent_wrapper.ts new file mode 100644 index 000000000..a8c64fcca --- /dev/null +++ b/core/src/workflow/nodes/llm_agent_wrapper.ts @@ -0,0 +1,321 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {Content} from '@google/genai'; +import {BaseAgent, isBaseAgent} from '../../agents/base_agent.js'; +import { + InvocationContext, + InvocationContextParams, + WorkflowInstructionScope, +} from '../../agents/invocation_context.js'; +import {isLlmAgent, LlmAgent} from '../../agents/llm_agent.js'; +import { + createEvent, + Event, + getFunctionCalls, + getFunctionResponses, +} from '../../events/event.js'; +import {isBaseTool} from '../../tools/base_tool.js'; +import { + FINISH_TASK_SUCCESS_RESULT, + FINISH_TASK_TOOL_NAME, +} from '../../tools/finish_task_tool.js'; +import {BaseNode, BaseNodeConfig, isContent} from '../base_node.js'; +import {NodeContext} from '../node_context.js'; +import {registerNodeBuilder} from '../utils/workflow_graph_utils.js'; + +/** Safety cap on chained `transfer_to_agent` hand-offs. */ +const MAX_TRANSFER_DEPTH = 10; + +/** Options for an {@link LLMAgentWrapper}. */ +export interface LLMAgentWrapperConfig extends Partial< + Omit +> { + name?: string; +} + +/** + * Runs a {@link BaseAgent} (typically an `LlmAgent`) as a workflow node in + * `single_turn` mode: the node input is appended as a user turn, the agent runs + * once, and its final model text becomes the node output. + * + * Ported (single_turn subset) from `google/adk-python` + * `workflow/_llm_agent_wrapper.py`. The `task` and `chat` modes (FinishTaskTool, + * task delegation, transfer, isolation scopes) are a Phase 7b continuation. + */ +export class LLMAgentWrapper extends BaseNode { + readonly agent: BaseAgent; + + constructor(agent: BaseAgent, config: LLMAgentWrapperConfig = {}) { + super({ + name: config.name ?? agent.name, + description: agent.description, + ...config, + }); + this.agent = agent; + } + + protected async *runImpl( + ctx: NodeContext, + input: unknown, + ): AsyncGenerator { + // Append the node input as a user turn so the agent responds to it. + if (input !== undefined && input !== null) { + const userEvent = createEvent({ + author: 'user', + invocationId: ctx.invocationId, + branch: ctx.branch, + content: toUserContent(input), + }); + if (ctx.isolationScope) { + userEvent.isolationScope = ctx.isolationScope; + } + // Persist the injected user turn (not just push it in-memory) so it + // survives on DB/Vertex session backends and is present on resume. + // appendEvent also adds it to `session.events` (deduped by id), which the + // agent reads synchronously to build its request. Falls back to a direct + // push when no session service is wired (e.g. in unit tests). + const sessionService = ctx.invocationContext.sessionService; + if (sessionService) { + await sessionService.appendEvent({ + session: ctx.session, + event: userEvent, + }); + } else { + ctx.session.events.push(userEvent); + } + } + + // Expose the node input and predecessor outputs to `{Class.field}` and + // `` instruction placeholders (Python's + // data-selection syntax). Carried on a child context so it never leaks to + // sibling/ordinary agent runs. + const agentIc = withWorkflowInstructionScope(ctx.invocationContext, { + input, + outputsByNode: collectPredecessorOutputs(ctx), + }); + + // Task mode: run a multi-round loop until the agent calls `finish_task`, + // whose arguments become the node output. + if (isLlmAgent(this.agent) && this.agent.mode === 'task') { + yield* this.runTaskMode(ctx, agentIc, this.agent); + return; + } + + // Run the agent, following any transfer_to_agent hand-offs to peers. + yield* this.runWithTransfers(ctx, agentIc, this.agent, 0); + } + + /** + * Runs a `task`-mode agent: the agent loops (LLM ↔ tools) until it calls the + * `finish_task` tool. The wrapper sniffs the `finish_task` function call and, + * on its successful function response, promotes the call's arguments to the + * node output (and to `outputKey` state, if set). Mirrors Python's + * `run_llm_agent_as_node` task branch. + */ + private async *runTaskMode( + ctx: NodeContext, + agentIc: InvocationContext, + agent: LlmAgent, + ): AsyncGenerator { + const finishTool = agent.finishTaskTool; + let pendingArgs: Record | undefined; + + for await (const event of agent.runAsync(agentIc)) { + const finishCall = getFunctionCalls(event).find( + (fc) => fc.name === FINISH_TASK_TOOL_NAME, + ); + if (finishCall) { + // Remember the latest finish_task args; wait for the success function + // response before terminating (a validation error lets the LLM retry). + pendingArgs = {...(finishCall.args ?? {})}; + yield event; + continue; + } + + if (pendingArgs !== undefined && isFinishTaskSuccessResponse(event)) { + const output = finishTool.extractOutput(pendingArgs); + event.output = output; + event.nodeInfo = {...(event.nodeInfo ?? {}), messageAsOutput: true}; + if (agent.outputKey && output !== undefined) { + ctx.actions.stateDelta[agent.outputKey] = output; + } + yield event; + return; + } + + yield event; + } + } + + /** + * Runs `agent`; if it emits a `transfer_to_agent` action, resolves the target + * in the agent tree and continues with it (multi-agent hand-off). This is the + * portable slice of Python's chat mode; autonomous task delegation + * (FinishTaskTool / task tools / isolation scopes) is not yet supported. + */ + private async *runWithTransfers( + ctx: NodeContext, + agentIc: InvocationContext, + agent: BaseAgent, + depth: number, + ): AsyncGenerator { + if (depth > MAX_TRANSFER_DEPTH) { + throw new Error( + `LLMAgentWrapper: transfer_to_agent depth exceeded ${MAX_TRANSFER_DEPTH} ` + + `(possible transfer loop starting at '${this.agent.name}').`, + ); + } + + let transferTarget: string | undefined; + for await (const event of agent.runAsync(agentIc)) { + this.maybeSetOutput(event); + yield event; + if (event.actions?.transferToAgent) { + transferTarget = event.actions.transferToAgent; + break; + } + } + + if (transferTarget) { + const target = agent.rootAgent.findAgent(transferTarget); + if (!target) { + throw new Error( + `LLMAgentWrapper: transfer target agent '${transferTarget}' not found.`, + ); + } + yield* this.runWithTransfers(ctx, agentIc, target, depth + 1); + } + } + + /** + * Promotes the final model text of an event to the node output (mirroring + * Python `process_llm_agent_output`). + */ + private maybeSetOutput(event: Event): void { + if (event.partial) { + return; + } + if (hasFunctionCalls(event)) { + return; + } + const content = event.content; + if (!content || content.role !== 'model' || !content.parts) { + return; + } + const text = content.parts + .filter((p) => p.text && !p.thought) + .map((p) => p.text) + .join(''); + + // If the agent declares an output schema, its text is structured JSON; + // surface the parsed object as the node output (matching Python). + let output: unknown = text; + const hasOutputSchema = !!(this.agent as {outputSchema?: unknown}) + .outputSchema; + if (hasOutputSchema && text.trim()) { + try { + output = JSON.parse(text); + } catch { + output = text; + } + } + + event.output = output; + event.nodeInfo = {...(event.nodeInfo ?? {}), messageAsOutput: true}; + } +} + +/** + * Creates a child InvocationContext carrying a workflow instruction scope, + * preserving the shared session/services/cost manager (like `withBranch`). + */ +function withWorkflowInstructionScope( + ic: InvocationContext, + scope: WorkflowInstructionScope, +): InvocationContext { + return new InvocationContext({ + ...(ic as unknown as InvocationContextParams), + workflowInstructionScope: scope, + }); +} + +/** + * Collects predecessor node outputs (keyed by node name) for the current + * invocation from the session events, for `` + * resolution. Node names are the leaf of each event's `nodeInfo.path` (with any + * `@runId` suffix stripped). + */ +function collectPredecessorOutputs(ctx: NodeContext): Record { + const outputs: Record = {}; + for (const event of ctx.session.events) { + if (event.invocationId !== ctx.invocationId || event.output === undefined) { + continue; + } + const path = event.nodeInfo?.path; + if (!path) { + continue; + } + const leaf = path.slice(path.lastIndexOf('.') + 1); + const name = leaf.includes('@') ? leaf.slice(0, leaf.indexOf('@')) : leaf; + outputs[name] = event.output; + } + return outputs; +} + +function hasFunctionCalls(event: Event): boolean { + return (event.content?.parts ?? []).some((p) => p.functionCall); +} + +/** + * Whether an event carries the success function response from `finish_task`. + * A non-success response (e.g. a validation error) returns false so the caller + * keeps iterating and the LLM gets a chance to retry. + */ +function isFinishTaskSuccessResponse(event: Event): boolean { + return getFunctionResponses(event).some((fr) => { + if (fr.name !== FINISH_TASK_TOOL_NAME) { + return false; + } + const response = (fr.response ?? {}) as {result?: unknown}; + return response.result === FINISH_TASK_SUCCESS_RESULT; + }); +} + +/** Converts an arbitrary node input into a user-role `Content`. */ +function toUserContent(input: unknown): Content { + if (isContent(input)) { + return {...input, role: 'user'}; + } + if (typeof input === 'string') { + return {role: 'user', parts: [{text: input}]}; + } + return {role: 'user', parts: [{text: JSON.stringify(input)}]}; +} + +/** Heuristic: an agent-like value exposes a `runAsync` method. */ +function isAgentLike(value: unknown): boolean { + return ( + typeof value === 'object' && + value !== null && + 'runAsync' in value && + typeof (value as {runAsync?: unknown}).runAsync === 'function' + ); +} + +/** + * Registers the builder that wraps a {@link BaseAgent} (or agent-like object) in + * an {@link LLMAgentWrapper}. + * + * Tools are excluded explicitly: a {@link BaseTool} also exposes `runAsync`, so + * this preserves the original tool-before-agent precedence regardless of the + * order in which node builders happen to be registered. + */ +registerNodeBuilder({ + match: (value): boolean => + !isBaseTool(value) && (isBaseAgent(value) || isAgentLike(value)), + build: (value, options) => new LLMAgentWrapper(value as BaseAgent, options), +}); diff --git a/core/src/workflow/nodes/node_tool.ts b/core/src/workflow/nodes/node_tool.ts new file mode 100644 index 000000000..a8f45c9dc --- /dev/null +++ b/core/src/workflow/nodes/node_tool.ts @@ -0,0 +1,143 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {FunctionDeclaration, Schema, Type} from '@google/genai'; + +import {Context} from '../../agents/context.js'; +import {Event} from '../../events/event.js'; +import {BaseTool, RunAsyncToolRequest} from '../../tools/base_tool.js'; +import {AsyncQueue} from '../../utils/async_queue.js'; +import { + isZodObject, + zodObjectToSchema, +} from '../../utils/simple_zod_to_json.js'; +import {BaseNode} from '../base_node.js'; +import {NodeContext} from '../node_context.js'; +import {executeChildNode} from '../node_runner.js'; + +/** + * A tool that executes a {@link BaseNode} (e.g. a `Workflow` or a function node) + * on behalf of an `LlmAgent`. This is the inverse of {@link ToolNode} (which + * exposes a tool as a workflow node): here a node/workflow is exposed to a model + * as a callable tool. + * + * The wrapped node MUST declare an `inputSchema` (the tool's parameter schema is + * derived from it). When the model calls the tool, the node runs with a + * {@link NodeContext} bridged from the tool's agent context (sharing the + * invocation, session, and state); the node's structured output becomes the + * tool result. + * + * Ported from `google/adk-python` `tools/_node_tool.py::NodeTool`. + * + * The tool is marked long-running so a node that pauses for input + * (`RequestInput`) does not force a synthetic empty response. + */ +export class NodeTool extends BaseTool { + readonly node: BaseNode; + + constructor(node: BaseNode, name?: string, description?: string) { + if (!node.inputSchema) { + throw new Error( + `Node '${node.name}' does not have an inputSchema defined. NodeTool ` + + 'requires an explicit input schema on the wrapped node.', + ); + } + super({ + name: name ?? node.name, + description: + description || node.description || `Executes the node: ${node.name}`, + isLongRunning: true, + }); + this.node = node; + } + + /** Whether the node's input schema is a (Zod) object rather than a scalar. */ + private get inputIsObject(): boolean { + return isZodObject(this.node.inputSchema); + } + + override _getDeclaration(): FunctionDeclaration { + let parameters: Schema; + if (this.inputIsObject) { + parameters = zodObjectToSchema(this.node.inputSchema as never); + } else { + // The GenAI API requires object-typed parameters; wrap a scalar schema + // under a single `request` property. + parameters = { + type: Type.OBJECT, + properties: {request: {type: Type.STRING}}, + required: ['request'], + }; + } + return {name: this.name, description: this.description, parameters}; + } + + override async runAsync({ + args, + toolContext, + }: RunAsyncToolRequest): Promise { + const nodeInput = this.inputIsObject ? args : args['request']; + + const child = await this.runNode(toolContext, nodeInput); + + if (child.interruptIds.length > 0) { + // The node paused for input. Returning undefined leaves the (long-running) + // tool call pending; the interrupt event has been surfaced separately so + // the invocation can pause and resume. (Resume wiring is layered on top.) + return undefined; + } + + return child.output === undefined ? {result: null} : child.output; + } + + /** + * Runs the wrapped node with a {@link NodeContext} bridged from the agent's + * tool context. Node events are streamed into the invocation's event queue + * when one is present (so intermediate/interrupt events surface to the agent); + * otherwise they are buffered and dropped (completion-only path). + */ + private async runNode( + toolContext: Context, + nodeInput: unknown, + ): Promise { + const ic = toolContext.invocationContext; + const runId = toolContext.functionCallId ?? this.node.name; + const channel = + (ic as {eventQueue?: AsyncQueue}).eventQueue ?? + new AsyncQueue(); + + const nodeCtx = new NodeContext({ + invocationContext: ic, + channel, + nodePath: this.node.name, + runId, + resumeInputs: collectResumeInputs(toolContext), + }); + + const base = ic.branch; + const segment = `${this.name}@${runId}`; + const overrideBranch = base ? `${base}.${segment}` : segment; + + return executeChildNode(nodeCtx, this.node, nodeInput, { + runId, + overrideBranch, + }); + } +} + +/** + * Collects resume inputs for the node from the tool context. When the tool call + * is being resumed after a `RequestInput`, the user's response is threaded + * through `toolConfirmation.payload` keyed by interrupt id (see the request-input + * resume processor). + */ +function collectResumeInputs(toolContext: Context): Record { + const payload = toolContext.toolConfirmation?.payload; + if (payload && typeof payload === 'object') { + return payload as Record; + } + return {}; +} diff --git a/core/test/agents/instructions_test.ts b/core/test/agents/instructions_test.ts index 6b7ed862e..50c01ecf9 100644 --- a/core/test/agents/instructions_test.ts +++ b/core/test/agents/instructions_test.ts @@ -15,6 +15,7 @@ import {injectSessionState} from '../../src/agents/instructions.js'; function makeContext( state: Record = {}, artifactService?: unknown, + workflowInstructionScope?: unknown, ): ReadonlyContext { const fakeInvocationContext = { session: { @@ -24,6 +25,7 @@ function makeContext( state, }, artifactService, + workflowInstructionScope, } as unknown as InvocationContext; return new ReadonlyContext(fakeInvocationContext); @@ -304,4 +306,63 @@ describe('injectSessionState', () => { 'Data: {"inlineData":{"mimeType":"text/plain","data":"abc"}}', ); }); + + describe('workflow field placeholders', () => { + it('resolves {Class.field} from the node input', async () => { + const ctx = makeContext({}, undefined, { + input: {time_info: '10:10 AM', city: 'Paris'}, + }); + expect( + await injectSessionState( + 'It is {CityTime.time_info} in {CityTime.city} right now.', + ctx, + ), + ).toBe('It is 10:10 AM in Paris right now.'); + }); + + it('resolves from predecessor outputs', async () => { + const ctx = makeContext({}, undefined, { + outputsByNode: { + lookup_time_function: {time_info: '9:00 AM', city: 'Rome'}, + }, + }); + expect( + await injectSessionState( + 'It is in ' + + '.', + ctx, + ), + ).toBe('It is 9:00 AM in Rome.'); + }); + + it('resolves workflow fields alongside normal state keys', async () => { + const ctx = makeContext({tone: 'formal'}, undefined, { + input: {city: 'Paris'}, + }); + expect(await injectSessionState('{tone}: {City.city}', ctx)).toBe( + 'formal: Paris', + ); + }); + + it('leaves {Class.field} untouched when there is no workflow scope', async () => { + const ctx = makeContext(); + expect(await injectSessionState('It is {CityTime.time_info}.', ctx)).toBe( + 'It is {CityTime.time_info}.', + ); + }); + + it('leaves untouched when there is no workflow scope', async () => { + const ctx = makeContext(); + expect(await injectSessionState('X Y', ctx)).toBe( + 'X Y', + ); + }); + + it('leaves an unknown {Class.field} untouched when the field is absent', async () => { + const ctx = makeContext({}, undefined, {input: {city: 'Paris'}}); + expect(await injectSessionState('{CityTime.missing}', ctx)).toBe( + '{CityTime.missing}', + ); + }); + }); }); diff --git a/core/test/workflow/llm_agent_test.ts b/core/test/workflow/llm_agent_test.ts new file mode 100644 index 000000000..863146b05 --- /dev/null +++ b/core/test/workflow/llm_agent_test.ts @@ -0,0 +1,268 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import {BaseAgent} from '../../src/agents/base_agent.js'; +import {injectSessionState} from '../../src/agents/instructions.js'; +import {InvocationContext} from '../../src/agents/invocation_context.js'; +import {ReadonlyContext} from '../../src/agents/readonly_context.js'; +import {createEvent, Event} from '../../src/events/event.js'; +import {PluginManager} from '../../src/plugins/plugin_manager.js'; +import {Session} from '../../src/sessions/session.js'; +import {AsyncQueue} from '../../src/utils/async_queue.js'; +import {node} from '../../src/workflow/node.js'; +import {NodeContext} from '../../src/workflow/node_context.js'; +import {LLMAgentWrapper} from '../../src/workflow/nodes/llm_agent_wrapper.js'; +import {Workflow} from '../../src/workflow/workflow.js'; + +function createIc(): InvocationContext { + const session = { + id: 's1', + appName: 'app', + userId: 'u', + events: [], + state: {}, + lastUpdateTime: Date.now(), + } as unknown as Session; + return new InvocationContext({ + invocationId: 'inv-1', + session, + agent: { + name: 'wf', + runAsync: async function* () {}, + } as unknown as BaseAgent, + pluginManager: new PluginManager(), + }); +} + +async function driveWorkflow( + wf: Workflow, + input?: unknown, +): Promise<{output: unknown; events: Event[]}> { + const channel = new AsyncQueue(); + const root = new NodeContext({ + invocationContext: createIc(), + channel, + nodePath: '', + runId: 'root', + }); + const events: Event[] = []; + const run = root.runNode(wf, input, {useAsOutput: true}).then( + () => channel.close(), + (err) => channel.fail(err), + ); + for await (const ev of channel) { + events.push(ev); + } + await run; + return {output: root.output, events}; +} + +/** + * A fake agent that echoes the most recent user turn as a model response — a + * stand-in for a real LlmAgent so the wrapper can be tested without a model. + */ +class EchoAgent extends BaseAgent { + constructor(name = 'echo') { + super({name}); + } + protected async *runAsyncImpl( + ctx: InvocationContext, + ): AsyncGenerator { + const lastUser = [...ctx.session.events] + .reverse() + .find((e) => e.author === 'user'); + const text = (lastUser?.content?.parts ?? []) + .map((p) => p.text ?? '') + .join(''); + yield createEvent({ + author: this.name, + invocationId: ctx.invocationId, + branch: ctx.branch, + content: {role: 'model', parts: [{text: `echo:${text}`}]}, + }); + } + // eslint-disable-next-line require-yield + protected async *runLiveImpl(): AsyncGenerator { + return; + } +} + +/** + * A fake agent that resolves a given instruction template against its context + * (the way the real instruction request-processor does) and yields the result — + * so we can assert workflow `{Class.field}` / `` placeholders + * resolve from the scope the wrapper attaches to the invocation context. + */ +class TemplateProbeAgent extends BaseAgent { + constructor( + private readonly template: string, + name = 'probe', + ) { + super({name}); + } + protected async *runAsyncImpl( + ctx: InvocationContext, + ): AsyncGenerator { + const resolved = await injectSessionState( + this.template, + new ReadonlyContext(ctx), + ); + yield createEvent({ + author: this.name, + invocationId: ctx.invocationId, + branch: ctx.branch, + content: {role: 'model', parts: [{text: resolved}]}, + }); + } + // eslint-disable-next-line require-yield + protected async *runLiveImpl(): AsyncGenerator { + return; + } +} + +describe('Phase 7 — LlmAgent as a node (single_turn)', () => { + it('runs an agent as a node and extracts its text output', async () => { + const wf = new Workflow({ + name: 'agent_wf', + edges: [['START', new EchoAgent()]], + }); + const {output, events} = await driveWorkflow(wf, 'hello'); + expect(output).toBe('echo:hello'); + // The agent's model event streamed through, authored by the agent. + expect(events.some((e) => e.author === 'echo')).toBe(true); + }); + + it('lets an agent be used directly in edges, feeding a downstream node (baseline bug #3)', async () => { + const upper = node( + (_c: NodeContext, input: string) => input.toUpperCase(), + { + name: 'upper', + }, + ); + const wf = new Workflow({ + name: 'agent_then_fn', + edges: [['START', new EchoAgent(), upper]], + }); + expect((await driveWorkflow(wf, 'hi')).output).toBe('ECHO:HI'); + }); + + it('resolves {Class.field} instruction placeholders from the node input', async () => { + const probe = new TemplateProbeAgent( + 'It is {CityTime.time_info} in {CityTime.city} right now.', + ); + const wf = new Workflow({name: 'tmpl_input', edges: [['START', probe]]}); + const {output} = await driveWorkflow(wf, { + time_info: '10:10 AM', + city: 'Paris', + }); + expect(output).toBe('It is 10:10 AM in Paris right now.'); + }); + + it('resolves from a predecessor output event', async () => { + const ic = createIc(); + // Seed a predecessor output the way the Runner persists node events. + ic.session.events.push( + createEvent({ + author: 'lookup_time_function', + invocationId: ic.invocationId, + nodeInfo: {path: 'wf.lookup_time_function'}, + output: {time_info: '9:00 AM', city: 'Rome'}, + }), + ); + const probe = new TemplateProbeAgent( + 'It is in ' + + '.', + ); + const channel = new AsyncQueue(); + const root = new NodeContext({ + invocationContext: ic, + channel, + nodePath: '', + runId: 'root', + }); + const run = root.runNode(node(probe), undefined, {useAsOutput: true}).then( + () => channel.close(), + (err) => channel.fail(err), + ); + for await (const _ev of channel) { + // drain + } + await run; + expect(root.output).toBe('It is 9:00 AM in Rome.'); + }); + + it('persists the injected user turn through the session service', async () => { + const ic = createIc(); + const appended: Event[] = []; + (ic as unknown as {sessionService: unknown}).sessionService = { + appendEvent: async ({ + session, + event, + }: { + session: {events: Event[]}; + event: Event; + }) => { + appended.push(event); + session.events.push(event); // mimic the base service adding to the list + return event; + }, + }; + const channel = new AsyncQueue(); + const root = new NodeContext({ + invocationContext: ic, + channel, + nodePath: '', + runId: 'root', + }); + const run = root + .runNode(node(new EchoAgent()), 'hi', {useAsOutput: true}) + .then( + () => channel.close(), + (err) => channel.fail(err), + ); + for await (const _ev of channel) { + // drain + } + await run; + + // The user turn went through the persistence path (appendEvent), not just a + // silent in-memory push, and the agent still saw it. + const userTurn = appended.find((e) => e.author === 'user'); + expect(userTurn?.content?.parts?.[0]?.text).toBe('hi'); + expect(root.output).toBe('echo:hi'); + }); + + it('node(agent) produces an LLMAgentWrapper carrying the agent name', () => { + const wrapped = node(new EchoAgent('assistant')); + expect(wrapped).toBeInstanceOf(LLMAgentWrapper); + expect(wrapped.name).toBe('assistant'); + }); + + it('routes on an agent-produced value', async () => { + // The classifier agent echoes; a function maps it to a route. + const classify = node( + (_c: NodeContext, input: string) => + createEvent({route: input.includes('?') ? 'q' : 's', output: input}), + {name: 'route_fn'}, + ); + const answer = node((_c: NodeContext, i: string) => `A:${i}`, { + name: 'answer', + }); + const comment = node((_c: NodeContext, i: string) => `C:${i}`, { + name: 'comment', + }); + const wf = new Workflow({ + name: 'agent_route', + edges: [ + ['START', new EchoAgent(), classify], + [classify, {q: answer, s: comment}], + ], + }); + // echo:'what?' contains '?', so route 'q'. + expect((await driveWorkflow(wf, 'what?')).output).toBe('A:echo:what?'); + }); +}); diff --git a/core/test/workflow/multi_agent_test.ts b/core/test/workflow/multi_agent_test.ts new file mode 100644 index 000000000..13467b74f --- /dev/null +++ b/core/test/workflow/multi_agent_test.ts @@ -0,0 +1,150 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import {BaseAgent} from '../../src/agents/base_agent.js'; +import {InvocationContext} from '../../src/agents/invocation_context.js'; +import {createEvent, Event} from '../../src/events/event.js'; +import {PluginManager} from '../../src/plugins/plugin_manager.js'; +import {Session} from '../../src/sessions/session.js'; +import {AsyncQueue} from '../../src/utils/async_queue.js'; +import {node} from '../../src/workflow/node.js'; +import {NodeContext} from '../../src/workflow/node_context.js'; +import {Workflow} from '../../src/workflow/workflow.js'; + +function createIc(): InvocationContext { + const session = { + id: 's1', + appName: 'app', + userId: 'u', + events: [], + state: {}, + lastUpdateTime: Date.now(), + } as unknown as Session; + return new InvocationContext({ + invocationId: 'inv-1', + session, + agent: { + name: 'wf', + runAsync: async function* () {}, + } as unknown as BaseAgent, + pluginManager: new PluginManager(), + }); +} + +async function driveWorkflow( + wf: Workflow, + input?: unknown, +): Promise<{output: unknown; events: Event[]}> { + const channel = new AsyncQueue(); + const root = new NodeContext({ + invocationContext: createIc(), + channel, + nodePath: '', + runId: 'root', + }); + const events: Event[] = []; + const settle = root.runNode(wf, input, {useAsOutput: true}).then( + () => channel.close(), + (err) => channel.fail(err), + ); + for await (const e of channel) { + events.push(e); + } + await settle; + return {output: root.output, events}; +} + +/** A fake agent that emits a fixed model text and optionally transfers. */ +class ScriptedAgent extends BaseAgent { + constructor( + name: string, + private readonly text: string, + private readonly transferTo?: string, + subAgents: BaseAgent[] = [], + ) { + super({name, subAgents}); + } + protected async *runAsyncImpl( + ctx: InvocationContext, + ): AsyncGenerator { + if (this.transferTo) { + yield createEvent({ + author: this.name, + invocationId: ctx.invocationId, + branch: ctx.branch, + actions: {transferToAgent: this.transferTo}, + }); + return; + } + yield createEvent({ + author: this.name, + invocationId: ctx.invocationId, + branch: ctx.branch, + content: {role: 'model', parts: [{text: this.text}]}, + }); + } + // eslint-disable-next-line require-yield + protected async *runLiveImpl(): AsyncGenerator { + return; + } +} + +describe('Phase 7b — multi-agent hand-off (transfer_to_agent)', () => { + it('follows a transfer to a peer agent and uses its output', async () => { + const specialist = new ScriptedAgent('specialist', 'specialist-answer'); + const coordinator = new ScriptedAgent( + 'coordinator', + '(unused)', + 'specialist', + [specialist], + ); + + const wf = new Workflow({ + name: 'transfer_wf', + edges: [['START', coordinator]], + }); + const {output, events} = await driveWorkflow(wf, 'question'); + + expect(output).toBe('specialist-answer'); + // Both the coordinator's transfer event and the specialist's answer stream. + expect( + events.some((e) => e.actions?.transferToAgent === 'specialist'), + ).toBe(true); + expect(events.some((e) => e.author === 'specialist')).toBe(true); + }); + + it('follows a chain of transfers', async () => { + const c = new ScriptedAgent('c_agent', 'final'); + const b = new ScriptedAgent('b_agent', '(unused)', 'c_agent', [c]); + const a = new ScriptedAgent('a_agent', '(unused)', 'b_agent', [b]); + + const wf = new Workflow({name: 'chain_wf', edges: [['START', a]]}); + expect((await driveWorkflow(wf, 'x')).output).toBe('final'); + }); +}); + +describe('Phase 7b — multi-agent orchestration via ctx.runNode', () => { + it('coordinates specialist agents imperatively (node-as-tool)', async () => { + const researcher = new ScriptedAgent('researcher', 'facts'); + const writer = new ScriptedAgent('writer', 'report'); + + // Idiomatic TS multi-agent: a coordinator drives sub-agents via runNode. + const wf = new Workflow({ + name: 'coordinator_wf', + dynamicEntry: async (ctx, input) => { + const research = await ctx.runNode(node(researcher), input); + const draft = await ctx.runNode(node(writer), research.output); + return {research: research.output, draft: draft.output}; + }, + }); + + expect(await driveWorkflow(wf, 'topic').then((r) => r.output)).toEqual({ + research: 'facts', + draft: 'report', + }); + }); +}); diff --git a/core/test/workflow/node_api_test.ts b/core/test/workflow/node_api_test.ts new file mode 100644 index 000000000..13f95a3ab --- /dev/null +++ b/core/test/workflow/node_api_test.ts @@ -0,0 +1,231 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import {BaseAgent} from '../../src/agents/base_agent.js'; +import {InvocationContext} from '../../src/agents/invocation_context.js'; +import {Event} from '../../src/events/event.js'; +import {PluginManager} from '../../src/plugins/plugin_manager.js'; +import {Session} from '../../src/sessions/session.js'; +import {BaseTool} from '../../src/tools/base_tool.js'; +import {AsyncQueue} from '../../src/utils/async_queue.js'; +import {BaseNode} from '../../src/workflow/base_node.js'; +import {node, Node} from '../../src/workflow/node.js'; +import {NodeContext} from '../../src/workflow/node_context.js'; +import {FunctionNode} from '../../src/workflow/nodes/function_node.js'; +import {JoinNode} from '../../src/workflow/nodes/join_node.js'; +import {LLMAgentWrapper} from '../../src/workflow/nodes/llm_agent_wrapper.js'; +import {ToolNode} from '../../src/workflow/nodes/tool_node.js'; +import {Workflow} from '../../src/workflow/workflow.js'; + +function createIc(): InvocationContext { + const session = { + id: 's1', + appName: 'app', + userId: 'u', + events: [], + state: {}, + lastUpdateTime: Date.now(), + } as unknown as Session; + return new InvocationContext({ + invocationId: 'inv-1', + session, + agent: { + name: 'wf', + runAsync: async function* () {}, + } as unknown as BaseAgent, + pluginManager: new PluginManager(), + }); +} + +async function runNode( + n: BaseNode, + input?: unknown, +): Promise<{events: Event[]; output: unknown}> { + const channel = new AsyncQueue(); + const root = new NodeContext({ + invocationContext: createIc(), + channel, + nodePath: '', + runId: 'root', + }); + const events: Event[] = []; + const run = root.runNode(n, input, {useAsOutput: true}).then( + () => channel.close(), + (err) => channel.fail(err), + ); + for await (const ev of channel) { + events.push(ev); + } + await run; + return {events, output: root.output}; +} + +async function driveWorkflow(wf: Workflow, input?: unknown): Promise { + return (await runNode(wf, input)).output; +} + +describe('Phase 3 — FunctionNode', () => { + it('boxes a plain return value into an output event', async () => { + const n = new FunctionNode('greet', (_c, input) => `hi ${input}`); + const {output, events} = await runNode(n, 'x'); + expect(output).toBe('hi x'); + expect(events[0].output).toBe('hi x'); + }); + + it('awaits an async handler', async () => { + const n = new FunctionNode('a', async (_c, input) => { + await Promise.resolve(); + return `async:${input}`; + }); + expect((await runNode(n, 'v')).output).toBe('async:v'); + }); + + it('supports sync generators (multiple events, last output wins)', async () => { + const n = new FunctionNode('gen', function* (_c, input) { + yield `${input}-1`; + yield `${input}-2`; + }); + const {events, output} = await runNode(n, 'g'); + expect(events.map((e) => e.output)).toEqual(['g-1', 'g-2']); + expect(output).toBe('g-2'); + }); + + it('supports async generators', async () => { + const n = new FunctionNode('agen', async function* (_c) { + yield 'one'; + await Promise.resolve(); + yield 'two'; + }); + expect((await runNode(n)).events.map((e) => e.output)).toEqual([ + 'one', + 'two', + ]); + }); + + it('skips null returns but keeps state deltas', async () => { + const n = new FunctionNode('writer', (ctx) => { + ctx.state.set('flag', true); + return null; + }); + const {events} = await runNode(n); + expect(events).toHaveLength(1); + expect(events[0].output).toBeUndefined(); + expect(events[0].actions.stateDelta['flag']).toBe(true); + }); + + it('attaches state deltas to output events', async () => { + const n = new FunctionNode('w', (ctx) => { + ctx.state.set('count', 3); + return 'ok'; + }); + const {events} = await runNode(n); + expect(events[0].output).toBe('ok'); + expect(events[0].actions.stateDelta['count']).toBe(3); + }); + + it('validates output against an outputSchema', async () => { + const schema = z.object({n: z.number()}); + const good = new FunctionNode('g', () => ({n: 5}), {outputSchema: schema}); + expect((await runNode(good)).output).toEqual({n: 5}); + + const bad = new FunctionNode('b', () => ({n: 'not-a-number'}), { + outputSchema: schema, + }); + await expect(runNode(bad)).rejects.toThrow(); + }); +}); + +describe('Phase 3 — node() factory', () => { + it('wraps a function, deriving the name', () => { + function classify() { + return 'ok'; + } + const n = node(classify); + expect(n).toBeInstanceOf(FunctionNode); + expect(n.name).toBe('classify'); + }); + + it('wraps a function with an explicit name override', () => { + const n = node((_c: NodeContext, input: unknown) => input, { + name: 'passthru', + }); + expect(n.name).toBe('passthru'); + }); + + it('wraps a BaseTool into a ToolNode', () => { + const tool = new EchoTool(); + const n = node(tool); + expect(n).toBeInstanceOf(ToolNode); + expect(n.name).toBe('echo'); + }); + + it('returns an existing BaseNode unchanged', () => { + const existing = new FunctionNode('keep', () => 1); + expect(node(existing)).toBe(existing); + }); + + it('wraps an agent into an LLMAgentWrapper', () => { + const fakeAgent = {name: 'a', runAsync: async function* () {}}; + const n = node(fakeAgent as unknown as never); + expect(n).toBeInstanceOf(LLMAgentWrapper); + expect(n.name).toBe('a'); + }); +}); + +describe('Phase 3 — Node subclass', () => { + class DoubleNode extends Node { + protected async *runNodeImpl(_ctx: NodeContext, input: number) { + yield input * 2; + } + } + + it('runs a subclass via runNodeImpl', async () => { + expect((await runNode(new DoubleNode({name: 'double'}), 5)).output).toBe( + 10, + ); + }); +}); + +describe('Phase 3 — JoinNode & ToolNode in a workflow', () => { + it('fans in with the real JoinNode', async () => { + const a = node((_c: NodeContext, input: unknown) => `A(${input})`, { + name: 'A', + }); + const b = node((_c: NodeContext, input: unknown) => `B(${input})`, { + name: 'B', + }); + const join = new JoinNode({name: 'join'}); + const wf = new Workflow({name: 'fan', edges: [['START', [a, b], join]]}); + expect(await driveWorkflow(wf, 'x')).toEqual({A: 'A(x)', B: 'B(x)'}); + }); + + it('runs a ToolNode with object args', async () => { + const wf = new Workflow({ + name: 'tool_wf', + edges: [['START', new ToolNode(new EchoTool())]], + }); + expect(await driveWorkflow(wf, {msg: 'hi'})).toEqual({ + echoed: {msg: 'hi'}, + }); + }); + + it('coerces a JSON-string ToolNode input to args', async () => { + const {output} = await runNode(new ToolNode(new EchoTool()), '{"a":1}'); + expect(output).toEqual({echoed: {a: 1}}); + }); +}); + +// A trivial tool that echoes its args back. +class EchoTool extends BaseTool { + constructor() { + super({name: 'echo', description: 'Echoes the input args.'}); + } + async runAsync({args}: {args: Record}): Promise { + return {echoed: args}; + } +} diff --git a/tests/integration/workflows/advanced_workflows_test.ts b/tests/integration/workflows/advanced_workflows_test.ts new file mode 100644 index 000000000..64cb097f9 --- /dev/null +++ b/tests/integration/workflows/advanced_workflows_test.ts @@ -0,0 +1,170 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Additional end-to-end (Runner) integration tests for advanced workflow + * scenarios: dynamic fan-out/fan-in, mid-graph HITL resume, multi-trigger + * re-execution, and a conditional dynamic loop. + */ + +import { + Event, + FunctionNode, + JoinNode, + node, + NodeContext, + RequestInput, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import { + collect, + createWorkflowRunner, + finalOutput, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +describe('workflow integration — dynamic fan-out / fan-in', () => { + it('fans out concurrent ctx.runNode calls and aggregates results', async () => { + const worker = new FunctionNode('work', (_c, item: number) => item * 10); + const wf = new Workflow({ + name: 'dynamic_fan_out_fan_in', + dynamicEntry: async (ctx) => { + const items = [1, 2, 3]; + const results = await Promise.all( + items.map((i) => ctx.runNode(worker, i, {runId: `w${i}`})), + ); + return results.map((r) => r.output); + }, + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toEqual([10, 20, 30]); + }); +}); + +describe('workflow integration — conditional dynamic loop', () => { + it('loops an LLM-free refiner until a condition is met', async () => { + const refine = new FunctionNode('refine', (_c, n: number) => n + 1); + const wf = new Workflow({ + name: 'conditional_loop', + dynamicEntry: async (ctx) => { + let value = 0; + let iterations = 0; + while (value < 5) { + value = (await ctx.runNode(refine, value, {runId: `r${iterations}`})) + .output as number; + iterations++; + } + return {value, iterations}; + }, + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toEqual({ + value: 5, + iterations: 5, + }); + }); +}); + +describe('workflow integration — mid-graph HITL resume', () => { + it('pauses in the middle of a chain and resumes without re-running upstream', async () => { + let aRuns = 0; + let cRuns = 0; + const a = node( + (_c: NodeContext, i: string) => { + aRuns++; + return `A(${i})`; + }, + {name: 'a'}, + ); + const gate = node( + (ctx: NodeContext, input: string) => { + const answer = ctx.resumeInputs['approve']; + if (answer === undefined) { + return new RequestInput({interruptId: 'approve', message: 'ok?'}); + } + return `${input}|${answer}`; + }, + // Single-node HITL gate: re-runs on resume to read its answer. + {name: 'gate', rerunOnResume: true}, + ); + const c = node( + (_c: NodeContext, i: string) => { + cRuns++; + return `C(${i})`; + }, + {name: 'c'}, + ); + const wf = new Workflow({ + name: 'mid_graph_hitl', + edges: [['START', a, gate, c]], + }); + const {run} = await createWorkflowRunner(wf); + + // Turn 1: a runs, gate interrupts, c must not run. + const turn1 = await collect(run('start')); + expect(aRuns).toBe(1); + expect(cRuns).toBe(0); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_input', + ), + ), + ).toBe(true); + + // Turn 2: resume; a is fast-forwarded (not re-run), gate resolves, c runs. + const turn2 = await collect( + run({ + role: 'user', + parts: [ + { + functionResponse: { + id: 'approve', + name: 'adk_request_input', + response: {result: 'yes'}, + }, + }, + ], + }), + ); + expect(aRuns).toBe(1); + expect(cRuns).toBe(1); + // gate re-ran with its ORIGINAL input 'A(start)', resolved with 'yes'. + expect(finalOutput(turn2)).toBe('C(A(start)|yes)'); + }); +}); + +describe('workflow integration — multi-trigger fan-in with JoinNode', () => { + it('joins three parallel branches produced from START', async () => { + const mk = (name: string): FunctionNode => + new FunctionNode(name, (_c, i: string) => `${name}:${i}`); + const join = new JoinNode({name: 'join'}); + const wf = new Workflow({ + name: 'triple_fan_in', + edges: [['START', [mk('x'), mk('y'), mk('z')], join]], + }); + const output = finalOutput(await runWorkflowOnce(wf, 'v')) as Record< + string, + string + >; + expect(output).toEqual({x: 'x:v', y: 'y:v', z: 'z:v'}); + }); +}); + +describe('workflow integration — parallel branches emit independent events', () => { + it('streams events from all parallel branches', async () => { + const mk = (name: string): FunctionNode => + new FunctionNode(name, (_c, i: string) => `${name}(${i})`); + const join = new JoinNode({name: 'join'}); + const wf = new Workflow({ + name: 'parallel_events', + edges: [['START', [mk('p'), mk('q')], join]], + }); + const events: Event[] = await runWorkflowOnce(wf, 'x'); + expect(events.some((e) => e.author === 'p')).toBe(true); + expect(events.some((e) => e.author === 'q')).toBe(true); + expect(events.some((e) => e.author === 'join')).toBe(true); + }); +}); diff --git a/tests/integration/workflows/agent_pipeline.model_responses.json b/tests/integration/workflows/agent_pipeline.model_responses.json new file mode 100644 index 000000000..1dbe81ab5 --- /dev/null +++ b/tests/integration/workflows/agent_pipeline.model_responses.json @@ -0,0 +1,41 @@ +{ + "summarizer": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "Summary: the report is positive."}] + }, + "finishReason": "STOP" + } + ] + } + ], + "researcher": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "Findings: A, B, C."}] + }, + "finishReason": "STOP" + } + ] + } + ], + "writer": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "Report drafted from the findings."}] + }, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/agent_pipeline_test.ts b/tests/integration/workflows/agent_pipeline_test.ts new file mode 100644 index 000000000..3c0f36fe4 --- /dev/null +++ b/tests/integration/workflows/agent_pipeline_test.ts @@ -0,0 +1,81 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration tests for LLM agents embedded in workflows (model responses from + * a JSON fixture): a mixed function/agent pipeline, and multi-agent + * orchestration driven imperatively via ctx.runNode. + */ + +import {FunctionNode, node, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './agent_pipeline.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — mixed function/agent pipeline', () => { + it('runs function -> LLM agent -> function, threading outputs', async () => { + const preprocess = new FunctionNode('preprocess', (_c, input: string) => + input.toUpperCase(), + ); + const summarizer = mockLlmAgent( + {name: 'summarizer', instruction: 'Summarize the provided text.'}, + responses['summarizer'], + ); + const postprocess = new FunctionNode( + 'postprocess', + (_c, input: string) => `[${input}]`, + ); + + const wf = new Workflow({ + name: 'agent_pipeline', + edges: [['START', preprocess, summarizer, postprocess]], + }); + + const events = await runWorkflowOnce(wf, 'the quarterly report'); + expect(finalOutput(events)).toBe('[Summary: the report is positive.]'); + expect(events.some((e) => e.author === 'summarizer')).toBe(true); + }); +}); + +describe('workflow integration — multi-agent orchestration (LLM)', () => { + it('coordinates two LLM agents via ctx.runNode', async () => { + const researcher = mockLlmAgent( + {name: 'researcher', instruction: 'Research the given topic.'}, + responses['researcher'], + ); + const writer = mockLlmAgent( + {name: 'writer', instruction: 'Write a report from the research.'}, + responses['writer'], + ); + + const wf = new Workflow({ + name: 'coordinator', + dynamicEntry: async (ctx, input) => { + const research = await ctx.runNode(node(researcher), input); + const report = await ctx.runNode(node(writer), research.output); + return {research: research.output, report: report.output}; + }, + }); + + const events = await runWorkflowOnce(wf, 'ADK workflows'); + expect(finalOutput(events)).toEqual({ + research: 'Findings: A, B, C.', + report: 'Report drafted from the findings.', + }); + expect(events.some((e) => e.author === 'researcher')).toBe(true); + expect(events.some((e) => e.author === 'writer')).toBe(true); + }); +}); diff --git a/tests/integration/workflows/auth_workflow_test.ts b/tests/integration/workflows/auth_workflow_test.ts new file mode 100644 index 000000000..b01cf86c2 --- /dev/null +++ b/tests/integration/workflows/auth_workflow_test.ts @@ -0,0 +1,104 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test: a workflow node that requires an API-key credential + * interrupts on the first turn and runs once the credential is supplied on + * resume (mirrors the Python `workflows/auth_api_key` sample). + */ + +import { + AuthConfig, + AuthCredential, + AuthCredentialTypes, + AuthScheme, + FunctionNode, + NodeContext, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import { + collect, + createWorkflowRunner, + finalOutput, +} from './workflow_test_utils.js'; + +const CREDENTIAL_KEY = 'weather_api'; + +function apiKeyAuthConfig(): AuthConfig { + return { + authScheme: {type: 'apiKey', in: 'header', name: 'X-API-Key'} as AuthScheme, + rawAuthCredential: {authType: AuthCredentialTypes.API_KEY}, + credentialKey: CREDENTIAL_KEY, + }; +} + +describe('workflow integration — auth gate (API key)', () => { + it('requests credentials, then runs after they are supplied on resume', async () => { + let runs = 0; + const secured = new FunctionNode( + 'secured', + (ctx: NodeContext) => { + runs++; + const cred = ctx.state.get('temp:' + CREDENTIAL_KEY); + return `weather(key=${cred?.apiKey})`; + }, + // Auth-gated nodes re-run on resume to store the credential and run their + // body (Python's auth samples set rerun_on_resume=True). + {authConfig: apiKeyAuthConfig(), rerunOnResume: true}, + ); + const wf = new Workflow({ + name: 'auth_api_key', + edges: [['START', secured]], + }); + const {run} = await createWorkflowRunner(wf); + + // Turn 1: no credential -> auth request interrupt; handler NOT run. + const turn1 = await collect(run('what is the weather?')); + expect(runs).toBe(0); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_credential', + ), + ), + ).toBe(true); + + // Turn 2: supply the credential -> node runs with it. + const credentialResponse: AuthConfig = { + authScheme: { + type: 'apiKey', + in: 'header', + name: 'X-API-Key', + } as AuthScheme, + credentialKey: CREDENTIAL_KEY, + exchangedAuthCredential: { + authType: AuthCredentialTypes.API_KEY, + apiKey: 'sk-test-123', + }, + }; + const turn2 = await collect( + run({ + role: 'user', + parts: [ + { + functionResponse: { + id: CREDENTIAL_KEY, + name: 'adk_request_credential', + response: credentialResponse as unknown as Record< + string, + unknown + >, + }, + }, + ], + }), + ); + + expect(runs).toBe(1); + expect(finalOutput(turn2)).toBe('weather(key=sk-test-123)'); + }); +}); diff --git a/tests/integration/workflows/core_workflows_test.ts b/tests/integration/workflows/core_workflows_test.ts new file mode 100644 index 000000000..1ac202da0 --- /dev/null +++ b/tests/integration/workflows/core_workflows_test.ts @@ -0,0 +1,306 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration tests for the major (non-LLM) workflow use cases, mirroring the + * Python `contributing/samples/workflows` samples, run end-to-end through the + * real Runner. Workflows are started with a text prompt (as a user would), so + * structured inputs are produced inside nodes rather than passed as the prompt. + */ + +import { + createEvent, + DEFAULT_ROUTE, + FunctionNode, + JoinNode, + node, + NodeContext, + ParallelWorker, + RequestInput, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import { + collect, + createWorkflowRunner, + finalOutput, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +describe('workflow integration — sequence', () => { + it('threads input through a linear chain', async () => { + const a = node((_c: NodeContext, i: string) => `${i}->A`, {name: 'a'}); + const b = node((_c: NodeContext, i: string) => `${i}->B`, {name: 'b'}); + const c = node((_c: NodeContext, i: string) => `${i}->C`, {name: 'c'}); + const wf = new Workflow({name: 'sequence', edges: [['START', a, b, c]]}); + + const events = await runWorkflowOnce(wf, 'INIT'); + expect(finalOutput(events)).toBe('INIT->A->B->C'); + }); +}); + +describe('workflow integration — route', () => { + it('routes to a branch and falls back to DEFAULT_ROUTE', async () => { + const routeNode = node( + (_c: NodeContext, input: string) => + createEvent( + input === 'jane' ? {output: input} : {route: 'retry', output: input}, + ), + {name: 'route_node'}, + ); + const retry = node((_c: NodeContext, i: string) => `RETRY:${i}`, { + name: 'retry_branch', + }); + const gen = node((_c: NodeContext, i: string) => `GEN:${i}`, { + name: 'generate', + }); + const wf = new Workflow({ + name: 'route', + edges: [ + ['START', routeNode], + [routeNode, {retry, [DEFAULT_ROUTE]: gen}], + ], + }); + + expect(finalOutput(await runWorkflowOnce(wf, 'john'))).toBe('RETRY:john'); + expect(finalOutput(await runWorkflowOnce(wf, 'jane'))).toBe('GEN:jane'); + }); +}); + +describe('workflow integration — fan-out / fan-in', () => { + it('runs branches in parallel and joins them', async () => { + const a = node((_c: NodeContext, i: string) => `A(${i})`, {name: 'a'}); + const b = node((_c: NodeContext, i: string) => `B(${i})`, {name: 'b'}); + const join = new JoinNode({name: 'join'}); + const wf = new Workflow({ + name: 'fan_out_fan_in', + edges: [['START', [a, b], join]], + }); + + expect(finalOutput(await runWorkflowOnce(wf, 'x'))).toEqual({ + a: 'A(x)', + b: 'B(x)', + }); + }); +}); + +describe('workflow integration — dynamic nodes & loop', () => { + it('runs an imperative loop that terminates', async () => { + const inc = new FunctionNode('inc', (_c, n: number) => (n as number) + 1); + const wf = new Workflow({ + name: 'dynamic_loop', + dynamicEntry: async (ctx) => { + let value = 0; + while (value < 3) { + value = (await ctx.runNode(inc, value)).output as number; + } + return value; + }, + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toBe(3); + }); +}); + +describe('workflow integration — parallel worker', () => { + it('maps a list across the wrapped node with bounded concurrency', async () => { + // The list is produced inside the workflow, then fanned out. + const produce = node((): number[] => [1, 2, 3, 4], {name: 'produce'}); + const worker = new ParallelWorker( + new FunctionNode('double', (_c, n: number) => (n as number) * 2), + {maxParallelWorkers: 2}, + ); + const wf = new Workflow({ + name: 'parallel_worker', + edges: [['START', produce, worker]], + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toEqual([2, 4, 6, 8]); + }); +}); + +describe('workflow integration — state', () => { + it('shares state across nodes', async () => { + const write = node( + (ctx: NodeContext, i: string) => { + ctx.state.set('greeting', `hi ${i}`); + return i; + }, + {name: 'write'}, + ); + const read = node((ctx: NodeContext) => ctx.state.get('greeting'), { + name: 'read', + }); + const wf = new Workflow({name: 'state', edges: [['START', write, read]]}); + expect(finalOutput(await runWorkflowOnce(wf, 'bob'))).toBe('hi bob'); + }); +}); + +describe('workflow integration — nested workflow', () => { + it('runs a workflow as a node inside another workflow', async () => { + const inner = new Workflow({ + name: 'inner', + edges: [ + [ + 'START', + node((_c: NodeContext, i: string) => `inner(${i})`, {name: 'in'}), + ], + ], + }); + const outer = new Workflow({ + name: 'outer', + edges: [ + [ + 'START', + inner, + node((_c: NodeContext, i: string) => `outer[${i}]`, {name: 'out'}), + ], + ], + }); + expect(finalOutput(await runWorkflowOnce(outer, 'x'))).toBe( + 'outer[inner(x)]', + ); + }); +}); + +describe('workflow integration — node as tool', () => { + it('lets a node call sub-nodes imperatively', async () => { + const add = new FunctionNode( + 'add', + (_c, args: {a: number; b: number}) => args.a + args.b, + ); + const orchestrator = node( + async (ctx: NodeContext) => { + const r1 = await ctx.runNode(add, {a: 2, b: 3}); + const r2 = await ctx.runNode(add, {a: 10, b: r1.output as number}); + return r2.output; + }, + {name: 'orchestrator'}, + ); + const wf = new Workflow({ + name: 'node_as_tool', + edges: [['START', orchestrator]], + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toBe(15); + }); +}); + +describe('workflow integration — retry', () => { + it('retries a flaky node until it succeeds', async () => { + let attempts = 0; + const flaky = new FunctionNode( + 'flaky', + () => { + attempts++; + if (attempts < 3) { + throw new Error('transient'); + } + return 'ok'; + }, + {retryConfig: {maxAttempts: 3, initialDelay: 0.001, jitter: 0}}, + ); + const wf = new Workflow({name: 'retry', edges: [['START', flaky]]}); + expect(finalOutput(await runWorkflowOnce(wf, 'x'))).toBe('ok'); + expect(attempts).toBe(3); + }); +}); + +describe('workflow integration — request_input (HITL)', () => { + it('pauses for input and resumes on a function response', async () => { + const gate = node( + (ctx: NodeContext, input: string) => { + const answer = ctx.resumeInputs['confirm']; + if (answer === undefined) { + return new RequestInput({interruptId: 'confirm', message: 'ok?'}); + } + // On resume, `input` must still be the original 'start', not the + // function-response message. + return `${input}:${answer}`; + }, + // Single-node HITL gate: re-runs on resume to read its answer (Python's + // rerun_on_resume=True). The default (two-node) semantics are covered by + // the request_input two-node test below. + {name: 'gate', rerunOnResume: true}, + ); + const wf = new Workflow({name: 'request_input', edges: [['START', gate]]}); + const {run} = await createWorkflowRunner(wf); + + const turn1 = await collect(run('start')); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_input', + ), + ), + ).toBe(true); + + const turn2 = await collect( + run({ + role: 'user', + parts: [ + { + functionResponse: { + id: 'confirm', + name: 'adk_request_input', + response: {result: 'yes'}, + }, + }, + ], + }), + ); + expect(finalOutput(turn2)).toBe('start:yes'); + }); + + it('two-node pattern: a rerun_on_resume=false node feeds its reply to the next node', async () => { + // Faithful port of Python's `request_input` two-node pattern: one node + // raises the interrupt and, on resume (with the default rerun_on_resume= + // false), does NOT re-run — its output becomes the resume value, which is + // passed as input to its successor. + let askRuns = 0; + const ask = node( + (_c: NodeContext) => { + askRuns++; + return new RequestInput({interruptId: 'review', message: 'reply?'}); + }, + {name: 'ask'}, + ); + const handle = node( + (_c: NodeContext, reply: string) => `handled(${reply})`, + {name: 'handle'}, + ); + const wf = new Workflow({ + name: 'request_input_two_node', + edges: [['START', ask, handle]], + }); + const {run} = await createWorkflowRunner(wf); + + const turn1 = await collect(run('start')); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_input', + ), + ), + ).toBe(true); + expect(askRuns).toBe(1); + + const turn2 = await collect( + run({ + role: 'user', + parts: [ + { + functionResponse: { + id: 'review', + name: 'adk_request_input', + response: {result: 'approve'}, + }, + }, + ], + }), + ); + // `ask` did NOT re-run; its reply flowed to `handle` as input. + expect(askRuns).toBe(1); + expect(finalOutput(turn2)).toBe('handled(approve)'); + }); +}); diff --git a/tests/integration/workflows/llm_loop.model_responses.json b/tests/integration/workflows/llm_loop.model_responses.json new file mode 100644 index 000000000..2d40bf7a3 --- /dev/null +++ b/tests/integration/workflows/llm_loop.model_responses.json @@ -0,0 +1,28 @@ +{ + "decider": [ + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "continue"}]}, + "finishReason": "STOP" + } + ] + }, + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "continue"}]}, + "finishReason": "STOP" + } + ] + }, + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "done"}]}, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/llm_loop_test.ts b/tests/integration/workflows/llm_loop_test.ts new file mode 100644 index 000000000..b3773693f --- /dev/null +++ b/tests/integration/workflows/llm_loop_test.ts @@ -0,0 +1,62 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test: an imperative loop driven by an LLM agent's decision + * ("continue"/"done"), with model responses from a JSON fixture. + */ + +import {node, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './llm_loop.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — LLM-driven loop', () => { + it('loops until the LLM agent decides to stop', async () => { + const decider = mockLlmAgent( + { + name: 'decider', + instruction: 'Reply "continue" to keep going or "done" to stop.', + }, + responses['decider'], + ); + + const wf = new Workflow({ + name: 'llm_loop', + dynamicEntry: async (ctx) => { + let rounds = 0; + for (;;) { + const decision = await ctx.runNode(node(decider), `round ${rounds}`, { + runId: `d${rounds}`, + }); + rounds++; + if (String(decision.output).includes('done')) { + break; + } + if (rounds > 5) { + break; // safety valve + } + } + return {rounds}; + }, + }); + + // "continue", "continue", "done" -> 3 rounds. + expect(finalOutput(await runWorkflowOnce(wf, 'start'))).toEqual({ + rounds: 3, + }); + }); +}); diff --git a/tests/integration/workflows/llm_tool_agent.model_responses.json b/tests/integration/workflows/llm_tool_agent.model_responses.json new file mode 100644 index 000000000..54a00e68a --- /dev/null +++ b/tests/integration/workflows/llm_tool_agent.model_responses.json @@ -0,0 +1,34 @@ +{ + "assistant": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [ + { + "functionCall": { + "name": "lookup", + "args": {"key": "answer"}, + "id": "call-1" + } + } + ] + }, + "finishReason": "STOP" + } + ] + }, + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "The looked-up value is 42."}] + }, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/llm_tool_agent_test.ts b/tests/integration/workflows/llm_tool_agent_test.ts new file mode 100644 index 000000000..f38527740 --- /dev/null +++ b/tests/integration/workflows/llm_tool_agent_test.ts @@ -0,0 +1,74 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test: an LLM agent that calls a tool, embedded as a workflow + * node. The mocked model first returns a function call, then a final answer + * after the tool runs. + */ + +import {FunctionTool, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './llm_tool_agent.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — LLM agent with tool calling', () => { + it('runs a tool-calling agent as a workflow node', async () => { + let toolCalled = false; + const lookup = new FunctionTool({ + name: 'lookup', + description: 'Looks up a value by key.', + parameters: z.object({key: z.string()}), + execute: async ({key}: {key: string}) => { + toolCalled = true; + return {key, value: 42}; + }, + }); + + const assistant = mockLlmAgent( + { + name: 'assistant', + instruction: 'Use the lookup tool to answer.', + tools: [lookup], + }, + responses['assistant'], + ); + + const wf = new Workflow({ + name: 'llm_tool_agent', + edges: [['START', assistant]], + }); + + const events = await runWorkflowOnce(wf, 'What is the answer?'); + + expect(toolCalled).toBe(true); + expect(finalOutput(events)).toBe('The looked-up value is 42.'); + // The tool call and its response both appear in the event stream. + expect( + events.some((e) => + (e.content?.parts ?? []).some((p) => p.functionCall?.name === 'lookup'), + ), + ).toBe(true); + expect( + events.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionResponse?.name === 'lookup', + ), + ), + ).toBe(true); + }); +}); diff --git a/tests/integration/workflows/loop_and_trigger_test.ts b/tests/integration/workflows/loop_and_trigger_test.ts new file mode 100644 index 000000000..cad0d3ab5 --- /dev/null +++ b/tests/integration/workflows/loop_and_trigger_test.ts @@ -0,0 +1,113 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration tests for routed (conditional) self-loops, error propagation from + * a parallel branch, and multi-trigger re-execution. + */ + +import { + createEvent, + DEFAULT_ROUTE, + FunctionNode, + JoinNode, + node, + NodeContext, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {finalOutput, runWorkflowOnce} from './workflow_test_utils.js'; + +describe('workflow integration — routed self-loop', () => { + it('loops a node back to itself until a route condition ends it', async () => { + const init = new FunctionNode('init', () => 0); + // Counter threads via the node output; routes back to itself until >= 3. + const worker = node( + (_c: NodeContext, n: number) => { + const next = (n as number) + 1; + return createEvent({route: next < 3 ? 'again' : 'done', output: next}); + }, + {name: 'worker'}, + ); + const report = node((_c: NodeContext, n: number) => `final:${n}`, { + name: 'report', + }); + + const wf = new Workflow({ + name: 'loop_self', + edges: [ + ['START', init, worker], + [worker, {again: worker, done: report}], + ], + }); + + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toBe('final:3'); + }); + + it('supports a routed loop with a DEFAULT_ROUTE exit', async () => { + const init = new FunctionNode('init', () => 0); + const worker = node( + (_c: NodeContext, n: number) => { + const next = (n as number) + 1; + // Emit 'again' while looping; no route (=> DEFAULT) when done. + return next < 2 + ? createEvent({route: 'again', output: next}) + : createEvent({output: next}); + }, + {name: 'worker'}, + ); + const done = node((_c: NodeContext, n: number) => `done:${n}`, { + name: 'done', + }); + const wf = new Workflow({ + name: 'loop_default_exit', + edges: [ + ['START', init, worker], + [worker, {again: worker, [DEFAULT_ROUTE]: done}], + ], + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toBe('done:2'); + }); +}); + +describe('workflow integration — parallel branch failure', () => { + it('fails the workflow when a parallel branch throws', async () => { + const good = new FunctionNode('good', (_c, i: string) => `good(${i})`); + const bad = new FunctionNode('bad', () => { + throw new Error('branch exploded'); + }); + const join = new JoinNode({name: 'join'}); + const wf = new Workflow({ + name: 'parallel_error', + edges: [['START', [good, bad], join]], + }); + await expect(runWorkflowOnce(wf, 'x')).rejects.toThrow('branch exploded'); + }); +}); + +describe('workflow integration — multi-trigger', () => { + it('re-executes a non-join node once per predecessor trigger', async () => { + let cRuns = 0; + const a = new FunctionNode('a', (_c, i: string) => `a(${i})`); + const b = new FunctionNode('b', (_c, i: string) => `b(${i})`); + const c = new FunctionNode('c', (_c, input: string) => { + cRuns++; + return `c(${input})`; + }); + // c has two predecessors and is NOT a JoinNode -> triggered twice. + const wf = new Workflow({ + name: 'multi_triggers', + edges: [ + ['START', [a, b]], + [a, c], + [b, c], + ], + }); + + await runWorkflowOnce(wf, 'x'); + expect(cRuns).toBe(2); + }); +}); diff --git a/tests/integration/workflows/node_as_tool_hitl_test.ts b/tests/integration/workflows/node_as_tool_hitl_test.ts new file mode 100644 index 000000000..a17f641da --- /dev/null +++ b/tests/integration/workflows/node_as_tool_hitl_test.ts @@ -0,0 +1,121 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test for HITL through a node-tool: an `LlmAgent` calls a node + * (passed as a tool) that raises a `RequestInput` while running. The invocation + * pauses; on the next turn the user answers the interrupt, the node-tool is + * re-run with the answer threaded as `resumeInputs`, and the tool result flows + * back to the model. Mirrors the `node_as_tool` `calculate_discount` pattern. + */ + +import { + getFunctionCalls, + getFunctionResponses, + InMemoryRunner, + node, + NodeContext, + RequestInput, +} from '@google/adk'; +import {Content} from '@google/genai'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import { + collect, + functionCallResponse, + mockLlmAgent, + textResponse, +} from './workflow_test_utils.js'; + +describe('workflow integration — HITL through a node-tool', () => { + it('pauses on RequestInput raised inside a node-tool and resumes', async () => { + const calculateDiscount = node( + (ctx: NodeContext, args: {tier: string}) => { + const resume = ctx.resumeInputs['confirm_vip_discount']; + if (!args.tier.includes('VIP')) { + return '5% off'; + } + if (resume === undefined) { + return new RequestInput({ + interruptId: 'confirm_vip_discount', + message: `Apply VIP discount for tier '${args.tier}'?`, + }); + } + const answer = + typeof resume === 'object' && resume !== null + ? (resume as {text?: string}).text + : resume; + return String(answer).toLowerCase() === 'yes' + ? '20% off' + : '5% off (VIP declined)'; + }, + { + name: 'calculate_discount', + inputSchema: z.object({tier: z.string()}), + rerunOnResume: true, + }, + ); + + const agent = mockLlmAgent( + { + name: 'discount_agent', + instruction: 'Compute the discount.', + tools: [calculateDiscount], + }, + [ + functionCallResponse('calculate_discount', { + tier: 'Verified VIP Member', + }), + textResponse('You get 20% off.'), + ], + ); + + const runner = new InMemoryRunner({agent, appName: agent.name}); + const session = await runner.sessionService.createSession({ + appName: agent.name, + userId: 'u1', + }); + + // Turn 1: the model calls the node-tool; the node interrupts for input. + const turn1 = await collect( + runner.runAsync({ + userId: 'u1', + sessionId: session.id, + newMessage: {role: 'user', parts: [{text: 'What discount do I get?'}]}, + }), + ); + const raisedInterrupt = turn1 + .flatMap((e) => getFunctionCalls(e)) + .some((fc) => fc.name === 'adk_request_input'); + expect(raisedInterrupt).toBe(true); + + // Turn 2: the user answers the interrupt; the node-tool re-runs and resolves. + const resume: Content = { + role: 'user', + parts: [ + { + functionResponse: { + id: 'confirm_vip_discount', + name: 'adk_request_input', + response: {result: 'yes'}, + }, + }, + ], + }; + const turn2 = await collect( + runner.runAsync({ + userId: 'u1', + sessionId: session.id, + newMessage: resume, + }), + ); + + const discountResult = turn2 + .flatMap((e) => getFunctionResponses(e)) + .find((fr) => fr.name === 'calculate_discount'); + expect(discountResult?.response).toMatchObject({result: '20% off'}); + }); +}); diff --git a/tests/integration/workflows/node_as_tool_test.ts b/tests/integration/workflows/node_as_tool_test.ts new file mode 100644 index 000000000..bcb0cc179 --- /dev/null +++ b/tests/integration/workflows/node_as_tool_test.ts @@ -0,0 +1,80 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test for node/workflow-as-tool: an `LlmAgent` is given a + * `Workflow` (and a function node) in its `tools`; the framework auto-wraps them + * as `NodeTool`s so the model can call them, and the node's structured output + * becomes the tool result. Mirrors the `node_as_tool` sample. + */ + +import { + getFunctionResponses, + InMemoryRunner, + node, + NodeContext, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import { + collect, + functionCallResponse, + mockLlmAgent, + textResponse, +} from './workflow_test_utils.js'; + +describe('workflow integration — node/workflow as an agent tool', () => { + it('lets an LlmAgent call a Workflow passed as a tool', async () => { + const lookup = node( + (_c: NodeContext, args: {userId: string}) => ({ + userId: args.userId, + tier: 'Verified VIP Member', + }), + {name: 'lookup_customer', inputSchema: z.object({userId: z.string()})}, + ); + const lookupWorkflow = new Workflow({ + name: 'customer_lookup_workflow', + description: 'Looks up customer status and tier by user_id.', + inputSchema: z.object({userId: z.string()}), + edges: [['START', lookup]], + }); + + const agent = mockLlmAgent( + { + name: 'customer_service_agent', + instruction: 'Help the customer.', + tools: [lookupWorkflow], + }, + [ + functionCallResponse('customer_lookup_workflow', {userId: 'u123'}), + textResponse('The customer is a Verified VIP Member.'), + ], + ); + + const runner = new InMemoryRunner({agent, appName: agent.name}); + const session = await runner.sessionService.createSession({ + appName: agent.name, + userId: 'u1', + }); + const events = await collect( + runner.runAsync({ + userId: 'u1', + sessionId: session.id, + newMessage: {role: 'user', parts: [{text: 'look up user u123'}]}, + }), + ); + + // The workflow tool ran and returned the tier as the function response. + const toolResult = events + .flatMap((e) => getFunctionResponses(e)) + .find((fr) => fr.name === 'customer_lookup_workflow'); + expect(toolResult?.response).toMatchObject({ + tier: 'Verified VIP Member', + userId: 'u123', + }); + }); +}); diff --git a/tests/integration/workflows/parallel_llm.model_responses.json b/tests/integration/workflows/parallel_llm.model_responses.json new file mode 100644 index 000000000..fa051a50e --- /dev/null +++ b/tests/integration/workflows/parallel_llm.model_responses.json @@ -0,0 +1,28 @@ +{ + "classifier": [ + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "processed"}]}, + "finishReason": "STOP" + } + ] + }, + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "processed"}]}, + "finishReason": "STOP" + } + ] + }, + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "processed"}]}, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/parallel_llm_test.ts b/tests/integration/workflows/parallel_llm_test.ts new file mode 100644 index 000000000..83e65f0d2 --- /dev/null +++ b/tests/integration/workflows/parallel_llm_test.ts @@ -0,0 +1,51 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test: a ParallelWorker mapping an LLM agent over a list of items, + * with model responses from a JSON fixture. + */ + +import {node, ParallelWorker, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './parallel_llm.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — ParallelWorker over an LLM agent', () => { + it('maps an LLM agent across a list of items', async () => { + const classifier = mockLlmAgent( + {name: 'classifier', instruction: 'Classify the item.'}, + responses['classifier'], + ); + + // Produce the list inside the workflow, then map the agent across it. + const produce = node((): string[] => ['alpha', 'beta', 'gamma'], { + name: 'produce', + }); + const worker = new ParallelWorker(node(classifier) as never, { + maxParallelWorkers: 1, + }); + + const wf = new Workflow({ + name: 'parallel_llm', + edges: [['START', produce, worker]], + }); + + const output = finalOutput(await runWorkflowOnce(wf, 'go')) as string[]; + expect(output).toHaveLength(3); + expect(output).toEqual(['processed', 'processed', 'processed']); + }); +}); diff --git a/tests/integration/workflows/plain_text_resume_test.ts b/tests/integration/workflows/plain_text_resume_test.ts new file mode 100644 index 000000000..34942e5be --- /dev/null +++ b/tests/integration/workflows/plain_text_resume_test.ts @@ -0,0 +1,60 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Interactive resume: a HITL/auth pause can be resumed by a plain-text reply + * (not just a structured function response), which is what enables `adk run` to + * drive HITL workflows by typing a message. + */ + +import { + createEvent, + node, + NodeContext, + RequestInput, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import { + collect, + createWorkflowRunner, + finalOutput, +} from './workflow_test_utils.js'; + +describe('workflow integration — plain-text interactive resume', () => { + it('resumes a HITL node from a plain-text reply (preserving original input)', async () => { + const gate = node( + (ctx: NodeContext, input: string) => { + const reply = ctx.resumeInputs['review']; + if (reply === undefined) { + return new RequestInput({interruptId: 'review', message: 'ok?'}); + } + return createEvent({output: `input=${input} reply=${reply}`}); + }, + // Single-node HITL gate: re-runs on resume to read its reply. + {name: 'gate', rerunOnResume: true}, + ); + const wf = new Workflow({ + name: 'plain_text_resume', + edges: [['START', gate]], + }); + const {run} = await createWorkflowRunner(wf); + + // Turn 1: interrupts. + const turn1 = await collect(run('hello')); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_input', + ), + ), + ).toBe(true); + + // Turn 2: a plain-text reply resumes the pending interrupt. + const turn2 = await collect(run('approve')); + expect(finalOutput(turn2)).toBe('input=hello reply=approve'); + }); +}); diff --git a/tests/integration/workflows/route.model_responses.json b/tests/integration/workflows/route.model_responses.json new file mode 100644 index 000000000..7b721d661 --- /dev/null +++ b/tests/integration/workflows/route.model_responses.json @@ -0,0 +1,28 @@ +{ + "classify_input": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "{\"category\": \"question\"}"}] + }, + "finishReason": "STOP" + } + ] + } + ], + "answer_question": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "The answer is 42."}] + }, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/route_llm_test.ts b/tests/integration/workflows/route_llm_test.ts new file mode 100644 index 000000000..d2c7a9215 --- /dev/null +++ b/tests/integration/workflows/route_llm_test.ts @@ -0,0 +1,106 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test mirroring the Python `workflows/route` sample: an LLM + * classifier with an output schema drives conditional routing to branch agents. + */ + +import {createEvent, node, NodeContext, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './route.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — route via LLM classifier', () => { + it('classifies the input and routes to the matching branch agent', async () => { + const processInput = node( + (ctx: NodeContext, input: string) => { + ctx.state.set('input', input); + return input; + }, + {name: 'process_input'}, + ); + + const classifyInput = mockLlmAgent( + { + name: 'classify_input', + instruction: + 'Based on this input, decide which category it belongs to: {input}', + outputSchema: z.object({ + category: z.enum(['question', 'statement', 'other']), + }), + outputKey: 'category', + }, + responses['classify_input'], + ); + + const routeOnCategory = node( + (_c: NodeContext, input: unknown) => { + const category = + typeof input === 'string' + ? (JSON.parse(input) as {category: string}).category + : (input as {category: string}).category; + return createEvent({route: category}); + }, + {name: 'route_on_category'}, + ); + + const answerQuestion = mockLlmAgent( + {name: 'answer_question', instruction: 'Answer the question: {input}'}, + responses['answer_question'], + ); + const commentOnStatement = mockLlmAgent( + { + name: 'comment_on_statement', + instruction: 'Comment on the statement: {input}', + }, + [], + ); + const handleOther = node( + () => + createEvent({ + content: { + role: 'model', + parts: [{text: 'I can only answer questions or comment.'}], + }, + }), + {name: 'handle_other'}, + ); + + const wf = new Workflow({ + name: 'route_llm', + edges: [ + ['START', processInput, classifyInput, routeOnCategory], + [ + routeOnCategory, + { + question: answerQuestion, + statement: commentOnStatement, + other: handleOther, + }, + ], + ], + }); + + const events = await runWorkflowOnce(wf, 'What is the meaning of life?'); + + // Classified as "question" -> routed to answer_question. + expect(finalOutput(events)).toBe('The answer is 42.'); + expect(events.some((e) => e.author === 'answer_question')).toBe(true); + expect(events.some((e) => e.author === 'comment_on_statement')).toBe(false); + }); +}); diff --git a/tests/integration/workflows/sequence.model_responses.json b/tests/integration/workflows/sequence.model_responses.json new file mode 100644 index 000000000..4ef1af622 --- /dev/null +++ b/tests/integration/workflows/sequence.model_responses.json @@ -0,0 +1,25 @@ +{ + "generate_fruit_agent": [ + { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "apple"}]}, + "finishReason": "STOP" + } + ] + } + ], + "generate_benefit_agent": [ + { + "candidates": [ + { + "content": { + "role": "model", + "parts": [{"text": "Apples are rich in fiber."}] + }, + "finishReason": "STOP" + } + ] + } + ] +} diff --git a/tests/integration/workflows/sequence_llm_test.ts b/tests/integration/workflows/sequence_llm_test.ts new file mode 100644 index 000000000..2482c61db --- /dev/null +++ b/tests/integration/workflows/sequence_llm_test.ts @@ -0,0 +1,60 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test mirroring the Python `workflows/sequence` sample: two LLM + * agents chained in a workflow, with model responses loaded from a JSON fixture. + */ + +import {Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {RawGenerateContentResponse} from '../test_case_utils.js'; +import modelResponses from './sequence.model_responses.json' with {type: 'json'}; +import { + finalOutput, + mockLlmAgent, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +const responses = modelResponses as Record< + string, + RawGenerateContentResponse[] +>; + +describe('workflow integration — sequence of LLM agents', () => { + it('chains two LLM agents, feeding the first output into the second', async () => { + const generateFruit = mockLlmAgent( + { + name: 'generate_fruit_agent', + instruction: + 'Return the name of a random fruit. Return only the name, nothing else.', + }, + responses['generate_fruit_agent'], + ); + const generateBenefit = mockLlmAgent( + { + name: 'generate_benefit_agent', + instruction: 'Tell me a health benefit about the specified fruit.', + }, + responses['generate_benefit_agent'], + ); + + const wf = new Workflow({ + name: 'sequence_llm', + edges: [['START', generateFruit, generateBenefit]], + }); + + const events = await runWorkflowOnce(wf, 'Give me a fruit fact'); + + // The final workflow output is the second agent's response. + expect(finalOutput(events)).toBe('Apples are rich in fiber.'); + // Both agents contributed events. + expect(events.some((e) => e.author === 'generate_fruit_agent')).toBe(true); + expect(events.some((e) => e.author === 'generate_benefit_agent')).toBe( + true, + ); + }); +}); diff --git a/tests/integration/workflows/task_mode_test.ts b/tests/integration/workflows/task_mode_test.ts new file mode 100644 index 000000000..0c1e02022 --- /dev/null +++ b/tests/integration/workflows/task_mode_test.ts @@ -0,0 +1,69 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test for `LlmAgent` task mode as a workflow node: the agent is + * given a `finish_task` tool and runs until it calls it; the call's arguments + * (conforming to the agent's output schema) become the node output and feed the + * next node. Mirrors the `agent_in_workflow` intake pattern. + */ + +import {node, NodeContext, Workflow} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import { + collect, + createWorkflowRunner, + finalOutput, + functionCallResponse, + mockLlmAgent, +} from './workflow_test_utils.js'; + +describe('workflow integration — LlmAgent task mode', () => { + it('runs until finish_task and promotes its args to the node output', async () => { + const intake = mockLlmAgent( + { + name: 'intake_agent', + mode: 'task', + instruction: 'Collect the patient name and phone number.', + outputSchema: z.object({name: z.string(), phoneNumber: z.string()}), + outputKey: 'identity', + }, + [ + functionCallResponse('finish_task', { + name: 'Jane Doe', + phoneNumber: '555-1234', + }), + ], + ); + + const check = node( + (_ctx: NodeContext, identity: {name: string; phoneNumber: string}) => + `checked:${identity.name} (${identity.phoneNumber})`, + {name: 'check'}, + ); + + const wf = new Workflow({ + name: 'task_wf', + edges: [['START', intake, check]], + }); + + const {run} = await createWorkflowRunner(wf); + const events = await collect(run('Hi, I am Jane Doe, 555-1234.')); + + // finish_task args flowed to `check` as its input. + expect(finalOutput(events)).toBe('checked:Jane Doe (555-1234)'); + // The finish_task args were also promoted to the node output. + expect( + events.some( + (e) => + typeof e.output === 'object' && + e.output !== null && + (e.output as {name?: string}).name === 'Jane Doe', + ), + ).toBe(true); + }); +}); diff --git a/tests/integration/workflows/tool_and_resilience_test.ts b/tests/integration/workflows/tool_and_resilience_test.ts new file mode 100644 index 000000000..983c0b32a --- /dev/null +++ b/tests/integration/workflows/tool_and_resilience_test.ts @@ -0,0 +1,119 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration tests for tools in workflows, retry exhaustion, and resuming a + * HITL interrupt raised inside a nested workflow. + */ + +import { + FunctionNode, + FunctionTool, + node, + NodeContext, + RequestInput, + ToolNode, + Workflow, +} from '@google/adk'; +import {describe, expect, it} from 'vitest'; +import {z} from 'zod'; +import { + collect, + createWorkflowRunner, + finalOutput, + runWorkflowOnce, +} from './workflow_test_utils.js'; + +describe('workflow integration — ToolNode', () => { + it('runs a BaseTool as a node with args from an upstream node', async () => { + const addTool = new FunctionTool({ + name: 'add', + description: 'Adds two numbers.', + parameters: z.object({a: z.number(), b: z.number()}), + execute: async ({a, b}: {a: number; b: number}) => ({sum: a + b}), + }); + const produceArgs = new FunctionNode('produce_args', () => ({a: 2, b: 3})); + const wf = new Workflow({ + name: 'tool_workflow', + edges: [['START', produceArgs, new ToolNode(addTool)]], + }); + expect(finalOutput(await runWorkflowOnce(wf, 'go'))).toEqual({sum: 5}); + }); +}); + +describe('workflow integration — retry exhaustion', () => { + it('fails the workflow when a node exhausts its retries', async () => { + let attempts = 0; + const flaky = new FunctionNode( + 'always_fails', + () => { + attempts++; + throw new Error('permanent failure'); + }, + {retryConfig: {maxAttempts: 3, initialDelay: 0.001, jitter: 0}}, + ); + const wf = new Workflow({name: 'retry_exhaust', edges: [['START', flaky]]}); + await expect(runWorkflowOnce(wf, 'go')).rejects.toThrow( + 'permanent failure', + ); + expect(attempts).toBe(3); + }); +}); + +describe('workflow integration — nested workflow HITL resume', () => { + it('resumes an interrupt raised inside a nested workflow', async () => { + const gate = node( + (ctx: NodeContext) => { + const answer = ctx.resumeInputs['approve']; + if (answer === undefined) { + return new RequestInput({interruptId: 'approve', message: 'ok?'}); + } + return `approved:${answer}`; + }, + // Single-node HITL gate: re-runs on resume to read its answer. + {name: 'gate', rerunOnResume: true}, + ); + const inner = new Workflow({name: 'inner', edges: [['START', gate]]}); + const outer = new Workflow({ + name: 'outer', + edges: [ + [ + 'START', + inner, + node((_c: NodeContext, i: string) => `wrapped(${i})`, {name: 'wrap'}), + ], + ], + }); + const {run} = await createWorkflowRunner(outer); + + // Turn 1: the nested gate interrupts; the interrupt bubbles up. + const turn1 = await collect(run('start')); + expect( + turn1.some((e) => + (e.content?.parts ?? []).some( + (p) => p.functionCall?.name === 'adk_request_input', + ), + ), + ).toBe(true); + + // Turn 2: resume; the nested gate resolves and the outer workflow finishes. + const turn2 = await collect( + run({ + role: 'user', + parts: [ + { + functionResponse: { + id: 'approve', + name: 'adk_request_input', + response: {result: 'yes'}, + }, + }, + ], + }), + ); + expect(finalOutput(turn2)).toBe('wrapped(approved:yes)'); + }); +}); diff --git a/tests/integration/workflows/workflow_test_utils.ts b/tests/integration/workflows/workflow_test_utils.ts new file mode 100644 index 000000000..5dc61365f --- /dev/null +++ b/tests/integration/workflows/workflow_test_utils.ts @@ -0,0 +1,125 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + Event, + InMemoryRunner, + LlmAgent, + LlmAgentConfig, + Workflow, + WorkflowAgent, +} from '@google/adk'; +import {Content, FinishReason} from '@google/genai'; +import { + GeminiWithMockResponses, + RawGenerateContentResponse, +} from '../test_case_utils.js'; + +/** + * Builds a raw generate-content response that returns plain model text. + */ +export function textResponse(text: string): RawGenerateContentResponse { + return { + candidates: [ + { + content: {role: 'model', parts: [{text}]}, + finishReason: FinishReason.STOP, + }, + ], + }; +} + +/** + * Builds a raw generate-content response that returns a single function call. + */ +export function functionCallResponse( + name: string, + args: Record, + id?: string, +): RawGenerateContentResponse { + return { + candidates: [ + { + content: {role: 'model', parts: [{functionCall: {name, args, id}}]}, + finishReason: FinishReason.STOP, + }, + ], + }; +} + +/** + * Constructs an {@link LlmAgent} whose model returns the given canned responses + * (loaded from a JSON fixture), so workflow integration tests are deterministic + * without a live model. + */ +export function mockLlmAgent( + config: Omit, + responses: RawGenerateContentResponse[], +): LlmAgent { + return new LlmAgent({ + ...config, + model: new GeminiWithMockResponses(responses), + }); +} + +/** + * Creates a runner for a {@link Workflow} (wrapped as a {@link WorkflowAgent}), + * bound to a single session so successive `run(...)` calls are additional turns + * (needed for HITL resume). Accepts a text prompt or a full `Content` (e.g. a + * function-response resume message). + */ +export async function createWorkflowRunner( + workflow: Workflow, +): Promise<{run: (message: string | Content) => AsyncGenerator}> { + const agent = new WorkflowAgent(workflow); + const runner = new InMemoryRunner({agent, appName: agent.name}); + const session = await runner.sessionService.createSession({ + appName: agent.name, + userId: 'u1', + }); + return { + run(message: string | Content): AsyncGenerator { + const newMessage: Content = + typeof message === 'string' + ? {role: 'user', parts: [{text: message}]} + : message; + return runner.runAsync({ + userId: 'u1', + sessionId: session.id, + newMessage, + }); + }, + }; +} + +/** Drains an event generator into an array. */ +export async function collect(gen: AsyncGenerator): Promise { + const events: Event[] = []; + for await (const event of gen) { + events.push(event); + } + return events; +} + +/** Runs a workflow for a single prompt and returns the emitted events. */ +export async function runWorkflowOnce( + workflow: Workflow, + prompt: string, +): Promise { + const {run} = await createWorkflowRunner(workflow); + return collect(run(prompt)); +} + +/** Returns the last non-undefined `output` across a list of events. */ +export function finalOutput(events: Event[]): unknown { + let output: unknown; + for (const event of events) { + if (event.output !== undefined) { + output = event.output; + } + } + return output; +} diff --git a/typedoc.json b/typedoc.json index 2ec80baa8..fe0cff4e8 100644 --- a/typedoc.json +++ b/typedoc.json @@ -9,5 +9,6 @@ "tsconfig": "./core/tsconfig.json", "plugin": ["typedoc-theme-fresh"], "theme": "fresh", - "excludeExternals": true + "excludeExternals": true, + "intentionallyNotExported": ["WorkflowInstructionScope"] } From 77b2d1a19add15f1319afa9ffb196df97b3edb70 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 15:03:16 -0700 Subject: [PATCH 2/6] fix(workflow): reconcile Part 6 with the updated engine Rebased Part 6 onto the current Part 5 and reconciled with the Parts 2-5 conventions: - executeChildNode now takes a single params object (node_tool.ts). - Move the LLM-agent-wrapper builder into the static node_builders.ts const list; drop its registerNodeBuilder self-registration and the now-obsolete register_builtin_nodes.ts (the const list replaces side-effect registration). --- core/src/workflow/node_builders.ts | 14 +++++++++++++ core/src/workflow/nodes/llm_agent_wrapper.ts | 21 ++++---------------- core/src/workflow/nodes/node_tool.ts | 8 +++++--- 3 files changed, 23 insertions(+), 20 deletions(-) diff --git a/core/src/workflow/node_builders.ts b/core/src/workflow/node_builders.ts index 99e09981e..7c5951ff7 100644 --- a/core/src/workflow/node_builders.ts +++ b/core/src/workflow/node_builders.ts @@ -4,8 +4,10 @@ * SPDX-License-Identifier: Apache-2.0 */ +import {BaseAgent, isBaseAgent} from '../agents/base_agent.js'; import {BaseTool, isBaseTool} from '../tools/base_tool.js'; import {FunctionNode, FunctionNodeHandler} from './nodes/function_node.js'; +import {isAgentLike, LLMAgentWrapper} from './nodes/llm_agent_wrapper.js'; import {ParallelWorker} from './nodes/parallel_worker.js'; import {ToolNode} from './nodes/tool_node.js'; import type { @@ -34,6 +36,17 @@ const TOOL_BUILDER: NodeBuilder = { build: (value, options) => new ToolNode(value as BaseTool, options), }; +/** + * Builds an {@link LLMAgentWrapper} from a {@link BaseAgent} (or agent-like + * value). Tools are excluded explicitly (a `BaseTool` also exposes `runAsync`), + * so the tool builder wins for those. + */ +const AGENT_BUILDER: NodeBuilder = { + match: (value) => + !isBaseTool(value) && (isBaseAgent(value) || isAgentLike(value)), + build: (value, options) => new LLMAgentWrapper(value as BaseAgent, options), +}; + /** * The built-in node builders, consulted in order by `buildNode` / `isNodeLike` * to turn a bare function / tool / agent into the right `BaseNode`. @@ -46,6 +59,7 @@ const TOOL_BUILDER: NodeBuilder = { export const NODE_BUILDERS: readonly NodeBuilder[] = [ FUNCTION_BUILDER, TOOL_BUILDER, + AGENT_BUILDER, ]; /** diff --git a/core/src/workflow/nodes/llm_agent_wrapper.ts b/core/src/workflow/nodes/llm_agent_wrapper.ts index a8c64fcca..c35b94a44 100644 --- a/core/src/workflow/nodes/llm_agent_wrapper.ts +++ b/core/src/workflow/nodes/llm_agent_wrapper.ts @@ -5,7 +5,7 @@ */ import {Content} from '@google/genai'; -import {BaseAgent, isBaseAgent} from '../../agents/base_agent.js'; +import {BaseAgent} from '../../agents/base_agent.js'; import { InvocationContext, InvocationContextParams, @@ -18,14 +18,12 @@ import { getFunctionCalls, getFunctionResponses, } from '../../events/event.js'; -import {isBaseTool} from '../../tools/base_tool.js'; import { FINISH_TASK_SUCCESS_RESULT, FINISH_TASK_TOOL_NAME, } from '../../tools/finish_task_tool.js'; import {BaseNode, BaseNodeConfig, isContent} from '../base_node.js'; import {NodeContext} from '../node_context.js'; -import {registerNodeBuilder} from '../utils/workflow_graph_utils.js'; /** Safety cap on chained `transfer_to_agent` hand-offs. */ const MAX_TRANSFER_DEPTH = 10; @@ -297,7 +295,7 @@ function toUserContent(input: unknown): Content { } /** Heuristic: an agent-like value exposes a `runAsync` method. */ -function isAgentLike(value: unknown): boolean { +export function isAgentLike(value: unknown): boolean { return ( typeof value === 'object' && value !== null && @@ -306,16 +304,5 @@ function isAgentLike(value: unknown): boolean { ); } -/** - * Registers the builder that wraps a {@link BaseAgent} (or agent-like object) in - * an {@link LLMAgentWrapper}. - * - * Tools are excluded explicitly: a {@link BaseTool} also exposes `runAsync`, so - * this preserves the original tool-before-agent precedence regardless of the - * order in which node builders happen to be registered. - */ -registerNodeBuilder({ - match: (value): boolean => - !isBaseTool(value) && (isBaseAgent(value) || isAgentLike(value)), - build: (value, options) => new LLMAgentWrapper(value as BaseAgent, options), -}); +// The builder that wraps a BaseAgent in an LLMAgentWrapper is wired into the +// static NODE_BUILDERS list in ../node_builders.ts. diff --git a/core/src/workflow/nodes/node_tool.ts b/core/src/workflow/nodes/node_tool.ts index a8f45c9dc..0872fadd9 100644 --- a/core/src/workflow/nodes/node_tool.ts +++ b/core/src/workflow/nodes/node_tool.ts @@ -121,9 +121,11 @@ export class NodeTool extends BaseTool { const segment = `${this.name}@${runId}`; const overrideBranch = base ? `${base}.${segment}` : segment; - return executeChildNode(nodeCtx, this.node, nodeInput, { - runId, - overrideBranch, + return executeChildNode({ + parent: nodeCtx, + node: this.node, + input: nodeInput, + options: {runId, overrideBranch}, }); } } From 90c7f3e2c0058ddcaa7303be7635f0c3643eb187 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 15:15:45 -0700 Subject: [PATCH 3/6] fix(workflow): address Part 6 review comments node-as-tool (node_tool.ts): - Bound node -> tool -> node recursion with a MAX_NODE_TOOL_DEPTH cap, tracked via a new immutable InvocationContext.nodeToolDepth carried through a depth+1 clone (so it survives agent-run ic clones). - Require the invocation event queue and a function-call id (throw otherwise) instead of falling back to a dead queue / a collapsing runId; drop the structural cast on eventQueue. - Pass an empty parent nodePath so the child path is a single segment, not the node name doubled. - Derive the tool parameter schema by narrowing isZodObject inline (drops the `as never`). instanceof -> brand guards: - llm_agent.ts uses isBaseNode; the request-input processor uses isNodeTool (new brand + guard on NodeTool). Both imports become type-only, removing the agents -> workflow value cycle. task mode (llm_agent_wrapper.ts): throw when a task-mode agent ends without a successful finish_task, instead of reporting the node COMPLETE with no output (its turn loop stays bounded by the invocation maxLlmCalls). InvocationContext: add clone(overrides) and use it for both the workflow-instruction-scope and node-tool-depth children (removes the `as unknown as InvocationContextParams` cast); remove the unused agentStates / endOfAgents fields; export WorkflowInstructionScope from common.ts / index.ts (public field type) instead of suppressing the TypeDoc warning. --- core/src/agents/invocation_context.ts | 38 +++++--- core/src/agents/llm_agent.ts | 4 +- .../request_input_llm_request_processor.ts | 4 +- core/src/common.ts | 5 +- core/src/index.ts | 1 + core/src/workflow/nodes/llm_agent_wrapper.ts | 16 +++- core/src/workflow/nodes/node_tool.ts | 95 +++++++++++++++---- typedoc.json | 3 +- 8 files changed, 120 insertions(+), 46 deletions(-) diff --git a/core/src/agents/invocation_context.ts b/core/src/agents/invocation_context.ts index d09c5f7aa..34174190c 100644 --- a/core/src/agents/invocation_context.ts +++ b/core/src/agents/invocation_context.ts @@ -53,9 +53,9 @@ export interface InvocationContextParams { activeStreamingTools?: Record; pluginManager: PluginManager; abortSignal?: AbortSignal; - agentStates?: Record; - endOfAgents?: Record; workflowInstructionScope?: WorkflowInstructionScope; + /** Nesting depth of node-as-tool executions; used to bound recursion. */ + nodeToolDepth?: number; } /** @@ -211,17 +211,6 @@ export class InvocationContext { */ eventQueue?: AsyncQueue; - /** - * Checkpointed states for workflow nodes under this invocation. - */ - agentStates: Record; - - /** - * Tracks whether specific agents or workflows have reached the end of their execution. - */ - - endOfAgents: Record; - /** * Workflow: field-resolution scope for `{Class.field}` / * `` instruction placeholders (set by @@ -229,6 +218,13 @@ export class InvocationContext { */ workflowInstructionScope?: WorkflowInstructionScope; + /** + * Nesting depth of node-as-tool ({@link NodeTool}) executions in this + * invocation. Incremented each time a node runs as a tool (via a depth+1 + * clone), so `NodeTool` can bound `node -> tool -> node` recursion. + */ + readonly nodeToolDepth: number; + /** * @param params The parameters for creating an invocation context. */ @@ -247,9 +243,8 @@ export class InvocationContext { this.activeStreamingTools = params.activeStreamingTools; this.pluginManager = params.pluginManager; this.abortSignal = params.abortSignal; - this.agentStates = params.agentStates ?? {}; - this.endOfAgents = params.endOfAgents ?? {}; this.workflowInstructionScope = params.workflowInstructionScope; + this.nodeToolDepth = params.nodeToolDepth ?? 0; // Inherit the parent invocation's cost manager when one is available. // Child contexts created for sub-agents, agent transfers and loop @@ -284,6 +279,19 @@ export class InvocationContext { incrementLlmCallCount() { this.invocationCostManager.incrementAndEnforceLlmCallsLimit(this.runConfig); } + + /** + * Returns a copy of this context with `overrides` applied. The spread carries + * every own field over (including the shared cost manager), so the copy keeps + * a single LLM-call counter for the invocation. + * + * Note: this copies own enumerable fields by value — scalar mutable fields + * (e.g. `endInvocation`) are decoupled from the original, while object-valued + * fields (`session`, …) stay shared by reference. + */ + clone(overrides: Partial = {}): InvocationContext { + return new InvocationContext({...this, ...overrides}); + } } export function newInvocationContextId(): string { diff --git a/core/src/agents/llm_agent.ts b/core/src/agents/llm_agent.ts index ab1828caf..c2e246fd3 100644 --- a/core/src/agents/llm_agent.ts +++ b/core/src/agents/llm_agent.ts @@ -9,7 +9,7 @@ import {context, trace} from '@opentelemetry/api'; import {FinishTaskTool} from '../tools/finish_task_tool.js'; import {FunctionTool} from '../tools/function_tool.js'; import {AsyncQueue} from '../utils/async_queue.js'; -import {BaseNode} from '../workflow/base_node.js'; +import {isBaseNode, type BaseNode} from '../workflow/base_node.js'; import {NodeTool} from '../workflow/nodes/node_tool.js'; import {z as z3} from 'zod/v3'; @@ -338,7 +338,7 @@ async function convertToolUnionToTools( if (isBaseTool(toolUnion)) { return [toolUnion]; } - if (toolUnion instanceof BaseNode) { + if (isBaseNode(toolUnion)) { // A node/Workflow passed as a tool is auto-wrapped as a NodeTool so the // model can call it (mirrors Python's Agent(tools=[node/workflow])). return [new NodeTool(toolUnion)]; diff --git a/core/src/agents/processors/request_input_llm_request_processor.ts b/core/src/agents/processors/request_input_llm_request_processor.ts index aca84e899..5073b6966 100644 --- a/core/src/agents/processors/request_input_llm_request_processor.ts +++ b/core/src/agents/processors/request_input_llm_request_processor.ts @@ -13,7 +13,7 @@ import { } from '../../events/event.js'; import {ToolConfirmation} from '../../tools/tool_confirmation.js'; import {AsyncQueue} from '../../utils/async_queue.js'; -import {NodeTool} from '../../workflow/nodes/node_tool.js'; +import {isNodeTool} from '../../workflow/nodes/node_tool.js'; import {REQUEST_INPUT_FUNCTION_CALL_NAME} from '../../workflow/utils/hitl_utils.js'; import {unwrapResponse} from '../../workflow/utils/rehydration_utils.js'; import {handleFunctionCallList} from '../functions.js'; @@ -58,7 +58,7 @@ export class RequestInputLlmRequestProcessor extends BaseLlmRequestProcessor { ); const toolsDict = Object.fromEntries(toolsList.map((t) => [t.name, t])); const nodeToolNames = new Set( - toolsList.filter((t) => t instanceof NodeTool).map((t) => t.name), + toolsList.filter((t) => isNodeTool(t)).map((t) => t.name), ); if (nodeToolNames.size === 0) { return; diff --git a/core/src/common.ts b/core/src/common.ts index a879e1f54..8306de705 100644 --- a/core/src/common.ts +++ b/core/src/common.ts @@ -20,7 +20,10 @@ export { functionsExportedForTestingOnly, } from './agents/functions.js'; export {InvocationContext} from './agents/invocation_context.js'; -export type {InvocationContextParams} from './agents/invocation_context.js'; +export type { + InvocationContextParams, + WorkflowInstructionScope, +} from './agents/invocation_context.js'; export {LiveRequestQueue} from './agents/live_request_queue.js'; export type {LiveRequest} from './agents/live_request_queue.js'; export {LlmAgent as Agent, LlmAgent, isLlmAgent} from './agents/llm_agent.js'; diff --git a/core/src/index.ts b/core/src/index.ts index 242f18fca..39cae5d89 100644 --- a/core/src/index.ts +++ b/core/src/index.ts @@ -25,6 +25,7 @@ export type {A2aUserBuilder, ToA2aOptions} from './a2a/agent_to_a2a.js'; export {bearerTokenUserBuilder} from './a2a/auth.js'; export type {ExecutorContext} from './a2a/executor_context.js'; export {InvocationContext} from './agents/invocation_context.js'; +export type {WorkflowInstructionScope} from './agents/invocation_context.js'; export {FileArtifactService} from './artifacts/file_artifact_service.js'; export {GcsArtifactService} from './artifacts/gcs_artifact_service.js'; export {getArtifactServiceFromUri} from './artifacts/registry.js'; diff --git a/core/src/workflow/nodes/llm_agent_wrapper.ts b/core/src/workflow/nodes/llm_agent_wrapper.ts index c35b94a44..fadf5c7b4 100644 --- a/core/src/workflow/nodes/llm_agent_wrapper.ts +++ b/core/src/workflow/nodes/llm_agent_wrapper.ts @@ -8,7 +8,6 @@ import {Content} from '@google/genai'; import {BaseAgent} from '../../agents/base_agent.js'; import { InvocationContext, - InvocationContextParams, WorkflowInstructionScope, } from '../../agents/invocation_context.js'; import {isLlmAgent, LlmAgent} from '../../agents/llm_agent.js'; @@ -147,6 +146,16 @@ export class LLMAgentWrapper extends BaseNode { yield event; } + + // The agent finished without a successful finish_task call (e.g. it answered + // in plain text). A task-mode agent must terminate via finish_task; failing + // loudly here avoids reporting the node COMPLETE with no output (which would + // make downstream `{Class.field}` placeholders resolve to nothing). The + // agent's own turn loop is bounded by the invocation's maxLlmCalls. + throw new Error( + `LLMAgentWrapper: task-mode agent '${agent.name}' ended without ` + + 'calling finish_task; no output was produced.', + ); } /** @@ -235,10 +244,7 @@ function withWorkflowInstructionScope( ic: InvocationContext, scope: WorkflowInstructionScope, ): InvocationContext { - return new InvocationContext({ - ...(ic as unknown as InvocationContextParams), - workflowInstructionScope: scope, - }); + return ic.clone({workflowInstructionScope: scope}); } /** diff --git a/core/src/workflow/nodes/node_tool.ts b/core/src/workflow/nodes/node_tool.ts index 0872fadd9..290a22777 100644 --- a/core/src/workflow/nodes/node_tool.ts +++ b/core/src/workflow/nodes/node_tool.ts @@ -7,9 +7,7 @@ import {FunctionDeclaration, Schema, Type} from '@google/genai'; import {Context} from '../../agents/context.js'; -import {Event} from '../../events/event.js'; import {BaseTool, RunAsyncToolRequest} from '../../tools/base_tool.js'; -import {AsyncQueue} from '../../utils/async_queue.js'; import { isZodObject, zodObjectToSchema, @@ -18,6 +16,18 @@ import {BaseNode} from '../base_node.js'; import {NodeContext} from '../node_context.js'; import {executeChildNode} from '../node_runner.js'; +/** + * A unique symbol branding {@link NodeTool} instances (see {@link isNodeTool}). + */ +const NODE_TOOL_SIGNATURE_SYMBOL = Symbol.for('google.adk.workflow.nodeTool'); + +/** + * Maximum nesting depth for node-as-tool executions, guarding against + * `node -> tool -> node` recursion (a node exposed as a tool whose agent can + * call that same tool again — unbounded model + tool spend otherwise). + */ +const MAX_NODE_TOOL_DEPTH = 8; + /** * A tool that executes a {@link BaseNode} (e.g. a `Workflow` or a function node) * on behalf of an `LlmAgent`. This is the inverse of {@link ToolNode} (which @@ -36,6 +46,9 @@ import {executeChildNode} from '../node_runner.js'; * (`RequestInput`) does not force a synthetic empty response. */ export class NodeTool extends BaseTool { + /** Brand identifying this object as a {@link NodeTool} (see {@link isNodeTool}). */ + readonly [NODE_TOOL_SIGNATURE_SYMBOL] = true; + readonly node: BaseNode; constructor(node: BaseNode, name?: string, description?: string) { @@ -54,15 +67,12 @@ export class NodeTool extends BaseTool { this.node = node; } - /** Whether the node's input schema is a (Zod) object rather than a scalar. */ - private get inputIsObject(): boolean { - return isZodObject(this.node.inputSchema); - } - override _getDeclaration(): FunctionDeclaration { + const schema = this.node.inputSchema; let parameters: Schema; - if (this.inputIsObject) { - parameters = zodObjectToSchema(this.node.inputSchema as never); + // Narrow inline so `zodObjectToSchema` typechecks without a cast. + if (schema && isZodObject(schema)) { + parameters = zodObjectToSchema(schema); } else { // The GenAI API requires object-typed parameters; wrap a scalar schema // under a single `request` property. @@ -75,6 +85,11 @@ export class NodeTool extends BaseTool { return {name: this.name, description: this.description, parameters}; } + /** Whether the node's input schema is a (Zod) object rather than a scalar. */ + private get inputIsObject(): boolean { + return isZodObject(this.node.inputSchema); + } + override async runAsync({ args, toolContext, @@ -95,29 +110,58 @@ export class NodeTool extends BaseTool { /** * Runs the wrapped node with a {@link NodeContext} bridged from the agent's - * tool context. Node events are streamed into the invocation's event queue - * when one is present (so intermediate/interrupt events surface to the agent); - * otherwise they are buffered and dropped (completion-only path). + * tool context. Node events are streamed into the invocation's event queue so + * intermediate/interrupt events surface to the agent (and a paused node can be + * resumed). Requires being invoked from an `LlmAgent` tool-call step, which is + * what provides that queue and the function-call id. */ private async runNode( toolContext: Context, nodeInput: unknown, ): Promise { const ic = toolContext.invocationContext; - const runId = toolContext.functionCallId ?? this.node.name; - const channel = - (ic as {eventQueue?: AsyncQueue}).eventQueue ?? - new AsyncQueue(); + + // A paused node's interrupt event must reach the session, so an event queue + // is required; without one the pause would be a silent dead end. + const channel = ic.eventQueue; + if (!channel) { + throw new Error( + `NodeTool '${this.name}' requires an invocation event queue; ` + + 'it must be invoked from an LlmAgent tool-call step.', + ); + } + + // A stable, unique run id per tool call: reused across resume so the paused + // run can be matched. (A shared fallback would collapse distinct calls.) + const runId = toolContext.functionCallId; + if (!runId) { + throw new Error( + `NodeTool '${this.name}' requires a function-call id; ` + + 'it must be invoked from an LlmAgent tool-call step.', + ); + } + + if (ic.nodeToolDepth >= MAX_NODE_TOOL_DEPTH) { + throw new Error( + `NodeTool '${this.name}': node-tool nesting exceeded ` + + `${MAX_NODE_TOOL_DEPTH} (possible node -> tool -> node recursion).`, + ); + } + // Run the node (and anything it reaches) at depth+1 so the guard above trips + // on unbounded recursion; the clone carries the depth across agent runs. + const childIc = ic.clone({nodeToolDepth: ic.nodeToolDepth + 1}); const nodeCtx = new NodeContext({ - invocationContext: ic, + invocationContext: childIc, channel, - nodePath: this.node.name, + // Empty so executeChildNode's path is a single segment (the node name), + // not the node name doubled. + nodePath: '', runId, resumeInputs: collectResumeInputs(toolContext), }); - const base = ic.branch; + const base = childIc.branch; const segment = `${this.name}@${runId}`; const overrideBranch = base ? `${base}.${segment}` : segment; @@ -130,6 +174,19 @@ export class NodeTool extends BaseTool { } } +/** + * Type guard for {@link NodeTool}. Matches on the brand rather than `instanceof` + * so it stays correct across package copies (mirrors `isBaseTool`). + */ +export function isNodeTool(value: unknown): value is NodeTool { + return ( + typeof value === 'object' && + value !== null && + NODE_TOOL_SIGNATURE_SYMBOL in value && + value[NODE_TOOL_SIGNATURE_SYMBOL] === true + ); +} + /** * Collects resume inputs for the node from the tool context. When the tool call * is being resumed after a `RequestInput`, the user's response is threaded diff --git a/typedoc.json b/typedoc.json index fe0cff4e8..2ec80baa8 100644 --- a/typedoc.json +++ b/typedoc.json @@ -9,6 +9,5 @@ "tsconfig": "./core/tsconfig.json", "plugin": ["typedoc-theme-fresh"], "theme": "fresh", - "excludeExternals": true, - "intentionallyNotExported": ["WorkflowInstructionScope"] + "excludeExternals": true } From 82dcb59639b9d00d0320f2b0558de7be4f6dc14e Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 15:22:15 -0700 Subject: [PATCH 4/6] test(workflow): use the shared createIc fixture (drop hand-rolled casts) Replace the duplicated hand-rolled createIc (with `as unknown as Session` / `as unknown as BaseAgent`) in the Part 6 workflow tests with the shared test_helpers.createIc (createSession + a real BaseAgent), removing the repeated double-casts the review flagged. --- core/test/workflow/llm_agent_test.ts | 23 +---------------------- core/test/workflow/multi_agent_test.ts | 23 +---------------------- core/test/workflow/node_api_test.ts | 25 +------------------------ 3 files changed, 3 insertions(+), 68 deletions(-) diff --git a/core/test/workflow/llm_agent_test.ts b/core/test/workflow/llm_agent_test.ts index 863146b05..7724a4d99 100644 --- a/core/test/workflow/llm_agent_test.ts +++ b/core/test/workflow/llm_agent_test.ts @@ -10,33 +10,12 @@ import {injectSessionState} from '../../src/agents/instructions.js'; import {InvocationContext} from '../../src/agents/invocation_context.js'; import {ReadonlyContext} from '../../src/agents/readonly_context.js'; import {createEvent, Event} from '../../src/events/event.js'; -import {PluginManager} from '../../src/plugins/plugin_manager.js'; -import {Session} from '../../src/sessions/session.js'; import {AsyncQueue} from '../../src/utils/async_queue.js'; import {node} from '../../src/workflow/node.js'; import {NodeContext} from '../../src/workflow/node_context.js'; import {LLMAgentWrapper} from '../../src/workflow/nodes/llm_agent_wrapper.js'; import {Workflow} from '../../src/workflow/workflow.js'; - -function createIc(): InvocationContext { - const session = { - id: 's1', - appName: 'app', - userId: 'u', - events: [], - state: {}, - lastUpdateTime: Date.now(), - } as unknown as Session; - return new InvocationContext({ - invocationId: 'inv-1', - session, - agent: { - name: 'wf', - runAsync: async function* () {}, - } as unknown as BaseAgent, - pluginManager: new PluginManager(), - }); -} +import {createIc} from './test_helpers.js'; async function driveWorkflow( wf: Workflow, diff --git a/core/test/workflow/multi_agent_test.ts b/core/test/workflow/multi_agent_test.ts index 13467b74f..22c8edbe7 100644 --- a/core/test/workflow/multi_agent_test.ts +++ b/core/test/workflow/multi_agent_test.ts @@ -8,32 +8,11 @@ import {describe, expect, it} from 'vitest'; import {BaseAgent} from '../../src/agents/base_agent.js'; import {InvocationContext} from '../../src/agents/invocation_context.js'; import {createEvent, Event} from '../../src/events/event.js'; -import {PluginManager} from '../../src/plugins/plugin_manager.js'; -import {Session} from '../../src/sessions/session.js'; import {AsyncQueue} from '../../src/utils/async_queue.js'; import {node} from '../../src/workflow/node.js'; import {NodeContext} from '../../src/workflow/node_context.js'; import {Workflow} from '../../src/workflow/workflow.js'; - -function createIc(): InvocationContext { - const session = { - id: 's1', - appName: 'app', - userId: 'u', - events: [], - state: {}, - lastUpdateTime: Date.now(), - } as unknown as Session; - return new InvocationContext({ - invocationId: 'inv-1', - session, - agent: { - name: 'wf', - runAsync: async function* () {}, - } as unknown as BaseAgent, - pluginManager: new PluginManager(), - }); -} +import {createIc} from './test_helpers.js'; async function driveWorkflow( wf: Workflow, diff --git a/core/test/workflow/node_api_test.ts b/core/test/workflow/node_api_test.ts index 13f95a3ab..7842611d5 100644 --- a/core/test/workflow/node_api_test.ts +++ b/core/test/workflow/node_api_test.ts @@ -6,11 +6,7 @@ import {describe, expect, it} from 'vitest'; import {z} from 'zod'; -import {BaseAgent} from '../../src/agents/base_agent.js'; -import {InvocationContext} from '../../src/agents/invocation_context.js'; import {Event} from '../../src/events/event.js'; -import {PluginManager} from '../../src/plugins/plugin_manager.js'; -import {Session} from '../../src/sessions/session.js'; import {BaseTool} from '../../src/tools/base_tool.js'; import {AsyncQueue} from '../../src/utils/async_queue.js'; import {BaseNode} from '../../src/workflow/base_node.js'; @@ -21,26 +17,7 @@ import {JoinNode} from '../../src/workflow/nodes/join_node.js'; import {LLMAgentWrapper} from '../../src/workflow/nodes/llm_agent_wrapper.js'; import {ToolNode} from '../../src/workflow/nodes/tool_node.js'; import {Workflow} from '../../src/workflow/workflow.js'; - -function createIc(): InvocationContext { - const session = { - id: 's1', - appName: 'app', - userId: 'u', - events: [], - state: {}, - lastUpdateTime: Date.now(), - } as unknown as Session; - return new InvocationContext({ - invocationId: 'inv-1', - session, - agent: { - name: 'wf', - runAsync: async function* () {}, - } as unknown as BaseAgent, - pluginManager: new PluginManager(), - }); -} +import {createIc} from './test_helpers.js'; async function runNode( n: BaseNode, From b3cc94764e48e483385677b4a144097471e2ebb6 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 15:42:18 -0700 Subject: [PATCH 5/6] docs(workflow): de-link isNodeTool in NodeTool brand comment --- core/src/workflow/nodes/node_tool.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/workflow/nodes/node_tool.ts b/core/src/workflow/nodes/node_tool.ts index 290a22777..7839e1530 100644 --- a/core/src/workflow/nodes/node_tool.ts +++ b/core/src/workflow/nodes/node_tool.ts @@ -17,7 +17,7 @@ import {NodeContext} from '../node_context.js'; import {executeChildNode} from '../node_runner.js'; /** - * A unique symbol branding {@link NodeTool} instances (see {@link isNodeTool}). + * A unique symbol branding {@link NodeTool} instances (see `isNodeTool`). */ const NODE_TOOL_SIGNATURE_SYMBOL = Symbol.for('google.adk.workflow.nodeTool'); @@ -46,7 +46,7 @@ const MAX_NODE_TOOL_DEPTH = 8; * (`RequestInput`) does not force a synthetic empty response. */ export class NodeTool extends BaseTool { - /** Brand identifying this object as a {@link NodeTool} (see {@link isNodeTool}). */ + /** Brand identifying this object as a {@link NodeTool} (see `isNodeTool`). */ readonly [NODE_TOOL_SIGNATURE_SYMBOL] = true; readonly node: BaseNode; From cce9d2d8957442a95544a3ed5697def4c7fb8256 Mon Sep 17 00:00:00 2001 From: kalenkevich Date: Mon, 3 Aug 2026 16:58:41 -0700 Subject: [PATCH 6/6] fix(workflow): re-publish Part 6 nodes after the Part 5 export change Part 5 replaced the `export * from './workflow/index.js'` star in index.ts with explicit named re-exports in common.ts, so Part 6's public additions must be listed there too: - Add LLMAgentWrapper / NodeTool and the LLMAgentWrapperConfig type to the common.ts workflow block (they reach the web entry point this way, and the block now mirrors the barrel exactly). - Update node_api_test to subclass the renamed `WorkflowNode` base class. --- core/src/common.ts | 3 +++ core/test/workflow/node_api_test.ts | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/core/src/common.ts b/core/src/common.ts index 8306de705..2931d78a4 100644 --- a/core/src/common.ts +++ b/core/src/common.ts @@ -352,9 +352,11 @@ export { FunctionNode, Graph, JoinNode, + LLMAgentWrapper, NodeContext, NodeStatus, NodeTimeoutError, + NodeTool, ParallelWorker, RequestInput, START, @@ -381,6 +383,7 @@ export type { FunctionNodeConfig, FunctionNodeHandler, FunctionNodeResult, + LLMAgentWrapperConfig, NodeContextOptions, NodeLike, NodeOptions, diff --git a/core/test/workflow/node_api_test.ts b/core/test/workflow/node_api_test.ts index 7842611d5..520268d16 100644 --- a/core/test/workflow/node_api_test.ts +++ b/core/test/workflow/node_api_test.ts @@ -10,7 +10,7 @@ import {Event} from '../../src/events/event.js'; import {BaseTool} from '../../src/tools/base_tool.js'; import {AsyncQueue} from '../../src/utils/async_queue.js'; import {BaseNode} from '../../src/workflow/base_node.js'; -import {node, Node} from '../../src/workflow/node.js'; +import {node, WorkflowNode} from '../../src/workflow/node.js'; import {NodeContext} from '../../src/workflow/node_context.js'; import {FunctionNode} from '../../src/workflow/nodes/function_node.js'; import {JoinNode} from '../../src/workflow/nodes/join_node.js'; @@ -155,7 +155,7 @@ describe('Phase 3 — node() factory', () => { }); describe('Phase 3 — Node subclass', () => { - class DoubleNode extends Node { + class DoubleNode extends WorkflowNode { protected async *runNodeImpl(_ctx: NodeContext, input: number) { yield input * 2; }