feat(agents): prototype v2 event stream + continuationId + React hooks

[chrisraygill]

Summary

End-to-end prototype of a v2 Agents protocol plus a framework-agnostic client core (AgentSession) and thin per-framework adapters that absorb all client glue. Server emits a discriminated event union with a single opaque continuation token; the AgentSession class owns the lifecycle so pages never touch streamFlow / runFlow directly. Validated across two frameworks: all 11 React sample pages plus 3 Angular pages running against the same backend, all verified end-to-end against gemini-flash-latest.

This is exploratory work, not a merge candidate. v2 is the only API surface (no backwards-compat layer); the goal is to validate the design with running code before committing to a wire protocol.

High-level API design

Server protocol

• AgentEvent: one discriminated union for the wire. Tagged variants on type: model-chunk, status, artifact-emitted, snapshot, interrupt, tool-error, detached, turn-end, error. Replaces the structural union of optional fields on AgentStreamChunk. Clients dispatch by type; old event.kind === 'X' pattern matching is gone.
• continuation: structured discriminated object replaces the state vs snapshotId fork. Shape is { kind: 'snapshot', snapshotId } (server-stored agents) or { kind: 'state', state } (stateless — state is carried inline as JSON, no base64). Round-trips on every turn through AgentInit.continuation and AgentOutput.continuation, and on the snapshot / detached / turn-end events. The previous FAILED_PRECONDITION validation (which would throw when callers passed the wrong kind of init for the wrong storage mode) is gone — the server reads whichever kind is in continuation. The kind discriminator keeps the storage-mode distinction visible (snapshot tokens are URL-fit, state objects can be megabytes) without forcing client code to branch.
• status event is application-defined. Single { type: 'status', status: <any> } event variant. The shape of status is intentionally not standardized — agents and their clients agree on their own structure (typed via the client's <TStatus> generic). Earlier prototype iterations had typed status / progress / phase variants; reverted to keep app-side flexibility.
• In-stream interrupt events with addressable toolCallId / toolName / input / kind: 'respond' | 'restart'. Replaces post-hoc scanning of result.message.content for unresolved tool requests. Resolves on the same turn the interrupt fires; UIs don't need to re-fetch the snapshot to discover an interrupt happened.
• In-stream tool-error events as the symmetric counterpart. Fires alongside the model-chunk that carries the failed toolResponse, carrying { toolCallId, toolName, errorText, errorCode?, details? }. Detection signal is metadata.toolError set by middleware — the stock filesystem middleware sets it. The toolResponse still flows back to the model so the conversation can continue; this event lets clients mark the matching in-flight tool call as failed.
• detached event marks the foreground → background transition explicitly, with continuation: { kind: 'snapshot', snapshotId } so the client can start polling without a second query. Was previously inferred from the absence of further chunks.
• snapshotId is re-exposed on AgentOutput and on the snapshot / detached / turn-end events as a server-side convenience field alongside continuation. Useful for direct /state / /abort calls and URL bookmarks without needing to pattern-match on continuation.kind.

Client surface (genkit/beta/client)

• walkAgentEvent(event, handlers) dispatches over the union. Handlers are all optional: onText, onReasoning, onToolRequest, onToolResponse, onToolError, onStatus (payload unknown), onArtifact, onInterrupt, onDetached, onTurnEnd. Strongly typed per-handler with no any escape hatches except where the protocol explicitly leaves a payload open (status).
• AgentSession<S, TStatus> is the framework-agnostic session core. Owns the entire client-side state machine: streaming chunk dispatch, structured-continuation round-trip, in-stream interrupt detection, foreground → background phase transitions with polling, snapshot rehydration, and lifecycle cleanup. Exposes a subscribe(listener) => unsubscribe observable + getState() snapshot getter + action methods (submit, abort, reset, respondToInterrupt, restartInterrupt, runVariants, continueFrom). Zero framework imports. The <TStatus> generic types the application-defined status payload per agent.
• No continuation codec helpers. Earlier iterations had an opaque-string token (snap:<id> / state:<base64>) with browser-safe encode/decode helpers; the structured continuation makes those unnecessary — the JSON itself is the wire format.

Framework adapters

AgentSession is the heavy lifter; each framework's adapter is a thin shim that wires the session's state observable into the framework's native reactivity model. Two adapters in this PR, validating the cross-framework design:

React (js/testapps/agents/web/src/genkit-react/useGenkitAgent.ts):

export function useGenkitAgent<S, TStatus>(
  options
): UseGenkitAgentResult<S, TStatus> {
  const sessionRef = useRef<AgentSession<S, TStatus> | null>(null);
  if (sessionRef.current === null)
    sessionRef.current = new AgentSession<S, TStatus>(options);
  const session = sessionRef.current;
  session.setOptions(options);

  const state = useSyncExternalStore(
    useMemo(() => session.subscribe.bind(session), [session]),
    useMemo(() => session.getState.bind(session), [session]),
  );
  // …bind action methods…
}

Angular (js/testapps/agents-angular/src/app/genkit-angular/inject-genkit-agent.ts):

export function injectGenkitAgent<S, TStatus>(
  options
): GenkitAgentHandle<S, TStatus> {
  const session = new AgentSession<S, TStatus>(options);
  const state = signal(session.getState());
  const unsubscribe = session.subscribe(() => state.set(session.getState()));
  inject(DestroyRef).onDestroy(() => {
    unsubscribe();
    session.dispose();
  });
  return { state: state.asReadonly(), ...boundMethods(session) };
}

Effective adapter code (excluding JSDoc/types/imports):

Adapter	Total LOC	Body LOC
React useGenkitAgent	128	11
Angular injectGenkitAgent	80	8
Core AgentSession	830	—

The duplication between adapters is the irreducible minimum: construct session, forward state changes to framework reactivity, bind action methods, tear down on destroy. No further extraction would shrink either side. Same pattern Vue (ref() + watch), Svelte (readable(initial, set => session.subscribe(...))), or Solid (from(session)) would use.

• useGenkitStream<I, O, S, Init> is a generic React primitive for non-agent streaming flows. Makes no assumptions about chunk shape.
• useGenkitRunFlow<I, O> wraps non-streaming runFlow for sibling endpoints (workspace listings, file reads, snapshot data). Useful when an agent page also calls a few one-shot flows — keeps everything reactive through one mental model.

Sample testapps

Two front-end testapps share the same backend (testapps/agents):

• testapps/agents/web — React, all 11 pages migrated.
• testapps/agents-angular — Angular, 3 representative pages ported (Weather, Banking, Background). Exists to prove the framework-agnostic core: same AgentSession, same backend, no server-side changes.

React: migrated pages

All 11 React testapp pages migrated. Each was previously a hand-rolled streamFlow loop with manual chunk dispatch, state vs snapshotId routing, interrupt detection, and (for some) background polling.

Page	Demonstrates
WeatherChat	basic chat + tool calls + URL bookmark
ClientState	typed customState round-trip
BankingInterrupt	in-stream interrupt with respondToInterrupt
WorkspaceBuilder	side-by-side chat + artifact panel
BackgroundAgent	auto detach → phase: 'background' polling, server-side abort
BranchingChat	runVariants for "pick your variant" UIs + continueFrom
TaskTracker	persisted custom state across turns
TripPlanner	URL-restorable multi-step planning
SubAgentChat	composed sub-agents
ResearchAgent	long-running research with progress
CodingAgent	the kitchen sink — filesystem middleware, tool approval, ask_user interrupt, snapshot resumption, sibling workspace endpoints via useGenkitRunFlow

Code shrinkage on the most representative pages:

Page	Before	After	Δ
WeatherChat	345 LOC	181 LOC	−47%
BankingInterrupt	358 LOC	195 LOC	−46%
BackgroundAgent	342 LOC	238 LOC	−30%
CodingAgent	1040 LOC	885 LOC	−15%

Most of the post-migration LOC is JSX + the explanatory sidebar that documents the v2 pattern. Actual hook usage is typically 5-15 lines.

End-to-end verification

Tested against the actual agents testapp with GEMINI_API_KEY against gemini-flash-latest, using Puppeteer-driven headless Chrome:

Wire format (curl):

• Stored agent (codingAgent): SSE stream returns turn-end events with payload { snapshotId, continuation: { kind: 'snapshot', snapshotId } }. No continuationId field anywhere.
• Stateless agent (weatherAgentStateless): final result returns continuation: { kind: 'state', state: {...} } with the full session state inline (no base64).
• Tool-error event: drove read_file against a nonexistent path → 2 tool-error events with addressable toolCallId and clean errorText (no Tool 'X' failed: prefix in the structured field).

React (testapps/agents/web):

• CodingAgent restored session: drove a write_file interrupt via curl, loaded /coding-agent/{sid} in browser → transcript rehydrated, approval dialog rendered with tool name + file path. Approve click → POST to /api/codingAgent returned 200, "✅ Approved" synth event + "✅ File written" tool row + follow-up model message rendered, file written to disk with correct content.
• WeatherChat: click suggestion → tool call + tool response + model response rendered in correct order, URL pushed to /weather/{snapshotId}, no console errors.
• ClientState (stateless agent): click suggestion → conversation works, sidebar renders the inline state JSON directly from agent.continuation.state (no decode step).
• BankingInterrupt: in-stream interrupt → approval dialog with Approve / Deny buttons rendered.

Angular (testapps/agents-angular):

• BackgroundAgent: submit topic with detach: true → phase transitioned to 'background' rendering snapshotId from the new derived state field → polling tick → phase transitioned to 'done' with full report rendered.

Earlier in the branch, all 11 React pages and 3 Angular pages were verified end-to-end.

Honest gaps

• useGenkitStream is still useful but rarely needed once useGenkitAgent covers chat-shaped flows.
• No subscribeAgent for background phase. Session polls. With server-sent events + Last-Event-ID resumption (suggestion S3 from the design discussion) it would prefer subscription.
• No streaming artifacts emission. Schema reserves the event types but no flow uses them yet.
• Adapters live in the testapps, not as published @genkit-ai/react / @genkit-ai/angular packages.
• Only React + Angular validated. Vue / Svelte / Solid would follow the same pattern but aren't proven in code yet.
• No type inference from agent definitions (suggestion S8). Hook consumers do well, but raw streamFlow callers don't benefit.

How to run

cd js
pnpm install
pnpm -C core build && pnpm -C ai build && pnpm -C genkit build
pnpm -C plugins/express build && pnpm -C plugins/middleware build

# Server (defaults to port 3400, matching both web apps' proxy targets)
cd testapps/agents
GEMINI_API_KEY=<your-key> npx tsx src/index.ts

# React client (in another terminal)
cd testapps/agents/web
npx vite
# Open http://localhost:5173/

# Angular client (in another terminal — same backend)
cd testapps/agents-angular
npx ng serve
# Open http://localhost:4200/

Files changed (high level)

• js/ai/src/agent.ts — schemas (AgentEvent discriminated union, structured ContinuationSchema, continuation field on AgentInit / AgentOutput / snapshot / detached / turn-end), runtime event emission, in-stream interrupt + tool-error detection with addressable refs
• js/genkit/src/client/agent-events.ts — walkAgentEvent visitor + AgentEvent / AgentContinuation type exports
• js/genkit/src/client/agent-session.ts — framework-agnostic AgentSession<S, TStatus> core
• js/genkit/src/client/index.ts — exports
• js/testapps/agents/src/* — server-side test flows updated to v2 (app-defined status payloads, structured continuation round-trip, { type: 'model-chunk', chunk }); workspace files endpoint no longer 500s on ENOENT; server default port is 3400
• js/testapps/agents/web/src/genkit-react/ — useGenkitAgent, useGenkitStream, useGenkitRunFlow
• js/testapps/agents-angular/ — Angular testapp + injectGenkitAgent
• js/testapps/agents/web/src/pages/ — all 11 pages rewritten to use the hooks
• js/testapps/agents/web/vite.config.ts — alias genkit/beta/client to TS sources

Why a draft PR

To spark concrete discussion on the v2 design with running code attached, not to merge as-is. Open design questions:

• Should walkAgentEvent live in genkit/beta/client or split into a separate @genkit-ai/client-utils so the AI SDK adapter can depend on it without pulling streamFlow?
• Adapters: extract to @genkit-ai/react / @genkit-ai/angular immediately (and add Vue / Svelte / Solid), or keep in examples/ for a release cycle to iterate on the API?
• Should AgentSession's resume path also sniff messages[last] for an unresolved tool request and surface it as pendingInterrupt (so restartInterrupt / respondToInterrupt work uniformly across in-stream and restored interrupts)?
• Should useGenkitRunFlow exist as a sibling primitive, or should pages just import runFlow directly for one-shots? (Angular has no equivalent today.)
• Should AgentInit accept an { events: AgentEvent[] } batching wrapper for network coalescing (proposal §1, deferred)?

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Code Review

This pull request introduces the v2 Agents API, which unifies state continuation via an opaque continuationId and adds a typed, discriminated AgentEvent stream. It also provides client-side event-walking helpers and React hooks (useGenkitAgent, useGenkitStream) to simplify agent interaction, tool calling, interrupts, and background execution, migrating several test applications to this new pattern. The review feedback highlights critical improvements for the React hooks to prevent memory leaks and redundant network requests by properly managing and clearing background polling timeouts and guarding state rehydration. Additionally, feedback recommends enhancing server-side robustness by throwing errors on malformed continuation tokens, handling potential circular references during status serialization, and adding defensive checks when parsing event content parts.

[chrisraygill]

Update: v2 is now the only API (no backwards-compat baggage)

Per design feedback that the Agents API is brand new and doesn't need backwards compat, this update removes the dual-emission and v1 schema fields entirely. v2 is the only API surface now, and all 11 testapp pages are migrated.

Schema simplification

Was	Now
AgentStreamChunkSchema = z.object({ modelChunk?, status?, artifact?, turnEnd?, event? })	AgentStreamChunkSchema = z.discriminatedUnion('type', [...])
AgentInitSchema = z.object({ snapshotId?, state? }) + FAILED_PRECONDITION if wrong one	AgentInitSchema = z.object({ continuationId? })
AgentOutputSchema = z.object({ snapshotId, state, continuationId, ... })	AgentOutputSchema = z.object({ continuationId, message, artifacts, state })
Runtime withEvent() dual-emission wrapper	Every emission site emits AgentEvent directly
walkAgentEvent(chunk, handlers) with legacy-field synthesis fallback	walkAgentEvent(event, handlers) direct dispatch

Wire format verified clean

$ curl -N -X POST .../api/weatherAgent -d '{"data":{"messages":[...]},"init":{}}'
EVENT type=model-chunk
EVENT type=model-chunk
EVENT type=model-chunk
EVENT type=turn-end

{
  "result": {
    "continuationId": "v1:57dc2276-...",
    "message": { "role": "model", "content": [...] },
    "state": { "messages": [...], "custom": {}, "artifacts": [] }
  }
}

All 11 pages migrated

Page	Pattern	Migration approach
WeatherChat	streaming chat + tool calls + URL restore	useGenkitAgent
BankingInterrupt	tool approval via respond interrupt	useGenkitAgent + respondToInterrupt
BackgroundAgent	detach + auto-poll + result render	useGenkitAgent (auto background phase)
ClientState	stateless agent, no server store	useGenkitAgent (same hook — continuationId hides the difference)
TripPlanner	definePromptAgent + .prompt file	useGenkitAgent
TaskTracker	typed custom state with mutating tools	useGenkitAgent<TaskState>
WorkspaceBuilder	artifacts	useGenkitAgent (artifacts reactive)
SubAgentChat	agents() middleware + call_agent	useGenkitAgent + delegation formatting
ResearchAgent	defineCustomAgent + status/progress events	useGenkitAgent (typed statusLabel + progress)
BranchingChat	parallel runs from same point	Kept on raw runFlow — branching is outside the hook's single-stream model — but updated to continuationId
CodingAgent	1040 LOC interrupt-heavy + custom UI	Minimal patch to continuationId; preserved custom approval/question dialog logic

Server-side adjustments (custom agents needed typed events)

// research-agent.ts before
sendChunk({ status: 'Decomposing question into sub-topics…' });
sendChunk({ status: `Researching (${i+1}/${n}): ${q}` });
sendChunk({ modelChunk: chunk });

// after — typed contract
sendChunk({ type: 'status', label: 'Decomposing question into sub-topics…' });
sendChunk({ type: 'progress', label: 'Researching', current: i+1, total: n });
sendChunk({ type: 'model-chunk', chunk });

A real defineAgent migration also surfaced a latent bug in task-agent.ts: the tools previously relied on the client pre-seeding state.custom = { tasks: [], nextId: 1 } via init.state. With v2, first-turn custom state is {} and the tools have to self-heal. One-line fix, but worth flagging as an example of how the v1 client-pre-seeds-state pattern was hiding tool bugs.

Sample-app verification (all in browser with Playwright + headless Chromium)

=== WeatherChat ===
  clicked suggestion (London)
  assistant reply rendered ✅

=== BankingInterrupt ===
  interrupt dialog appeared ✅
  clicked Approve → model confirmed "transferred $500 to your savings"

=== BackgroundAgent ===
  pending UI visible: true
  background result rendered ✅ (1690 chars)

=== TaskTracker ===
  tasks visible: 1 ✅ (agent.customState reactive update)

=== ResearchAgent ===
  status/sub-questions visible ✅ (typed status events drive indicator)

errors: 0 / 0

What's still not done

• BranchingChat is on raw runFlow not the hook. A useGenkitBranching hook (or a branchFrom option on useGenkitAgent) would be the natural extension. Scoped out for this prototype.
• CodingAgent kept most of its original 1040 LOC. Could migrate the streaming loop to the hook but the custom approval/question dialogs and middleware-specific tool rendering would need significant rework. Worth doing in a follow-up.
• TaskTracker tool fix (self-init on first turn) is a side-effect — confirms v1's client-pre-seeded state pattern was load-bearing for tool correctness. Worth surfacing as a Genkit docs note.

[chrisraygill]

Cleanup pass: backwards-compat baggage removed, all 11 pages verified

Token format renamed

Was	Now
v1:<snapshotId>	snap:<snapshotId>
v1s:<base64(state)>	state:<base64(state)>

The prefix is now a clear storage-kind discriminator. No version baggage.

Helpers exported from genkit/beta/client (browser-safe)

import {
  encodeSnapshotContinuation,
  encodeStateContinuation,
  continuationToSnapshotId,
} from 'genkit/beta/client';

Re-exported from genkit/beta for server-side test flows.

Hook centralizes the format

Page code no longer composes tokens by hand. The hook exposes a derived agent.snapshotId and accepts resumeFromSnapshotId (alias for URL-param case):

// Was
const resumeFromContinuation = urlSnapshotId ? `v1:${urlSnapshotId}` : undefined;
const sid = agent.continuationId?.startsWith('v1:')
  ? agent.continuationId.slice(3)
  : undefined;

// Now
const agent = useGenkitAgent({ url, resumeFromSnapshotId: urlSnapshotId });
const sid = agent.snapshotId;

getSnapshotDataAction accepts either form

// Server decodes if it sees the `snap:` prefix; otherwise treats input as
// a raw snapshotId for direct server-internal API callers.

Server test flows migrated

All seven server-side test flows (background-agent, banking-agent, branching-agent, coding-agent, file-store-agent, task-agent, weather-agent-stateless) now use continuationId + continuationToSnapshotId helper. Zero references to init: { snapshotId } or output.snapshotId.

Real bug found and fixed by the verification sweep

Importing helpers from genkit/beta (the Node package) in BranchingChat + CodingAgent broke all 11 pages at runtime with Module "node:perf_hooks" has been externalized. Caught by the Playwright sweep, not by typecheck. Fix: helpers must come from genkit/beta/client (the browser-safe path).

Worth noting as a Genkit packaging concern: nothing prevents genkit/beta imports from leaking Node-only code into browser bundles. A real @genkit-ai/react package would want stricter exports.

Browser verification — all 11 pages

Page	Result
WeatherChat	✅ PASS
ClientState (stateless)	✅ PASS
BankingInterrupt	✅ PASS
WorkspaceBuilder	✅ PASS
BackgroundAgent	✅ PASS
TaskTracker	✅ PASS
TripPlanner	✅ PASS
SubAgentChat	✅ PASS
BranchingChat	✅ PASS
ResearchAgent	✅ PASS
CodingAgent	⚠️ 2 console errors (pre-existing — filesystem-state endpoint not configured; unrelated to this PR)

Zero page errors across all 11. Driven with Playwright + headless Chromium against the live API key.

CI status

• commitlint: PR title updated to feat(agents): prototype v2 event stream + continuationId + React hooks (conventional format)
• format/lint: npm run format run; copyright headers updated
• build: server test flow build errors fixed (the snapshotId/state field references I missed in v1)
• cla/google: expected (first contribution; not signing on personal account)