Add stateful agent conversations via Cloudflare Agents SDK

Migrate AgentRunner from raw DurableObject to the Agents SDK Agent base
class, enabling persistent conversation history across workflow runs.
When an agent_id input is set, the same DO instance is reused so prior
messages are carried forward; without it, behavior is unchanged.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bchapuis

apps/api/package.json

@@ -54,6 +54,7 @@
     "@mapbox/martini": "^0.2.0",
     "@turf/helpers": "^7.3.2",
     "@turf/turf": "^7.3.2",
+    "agents": "^0.1.6",
     "aws4fetch": "^1.0.20",
     "cloudflare": "^5.2.0",
     "cron-parser": "^5.4.0",

apps/api/src/durable-objects/agent-runner.ts

@@ -8,13 +8,13 @@
  * Supports four LLM providers: anthropic, google, openai, workers-ai.
  */
 
-import { DurableObject } from "cloudflare:workers";
 import Anthropic from "@anthropic-ai/sdk";
 import type { ToolDefinition, ToolReference } from "@dafthunk/runtime";
 import { NodeToolProvider } from "@dafthunk/runtime";
 import type { AgentProvider } from "@dafthunk/runtime/nodes/agent/base-agent-node";
 import type {
   AgentLoopResult,
+  AgentLoopState,
   AgentMessage,
   LLMResponse,
 } from "@dafthunk/runtime/utils/agent-loop";
@@ -28,6 +28,7 @@ import { createCodeModeToolDefinition } from "@dafthunk/runtime/utils/code-mode"
 import type { TokenPricing } from "@dafthunk/runtime/utils/usage";
 import { calculateTokenUsage } from "@dafthunk/runtime/utils/usage";
 import { GoogleGenAI } from "@google/genai";
+import { Agent } from "agents";
 import OpenAI from "openai";
 
 import type { Bindings } from "../context";
@@ -69,6 +70,8 @@ export interface AgentRunRequest {
   googleSearch?: boolean;
   /** Organization ID for credential access (integrations, secrets) */
   organizationId: string;
+  /** When set, routes to a persistent DO that maintains state across runs */
+  agentId?: string;
 }
 
 export interface AgentRunResponse {
@@ -78,11 +81,21 @@ export interface AgentRunResponse {
   totalSteps: number;
   totalInputTokens: number;
   totalOutputTokens: number;
+  /** Full message history (only present when agentId is set) */
+  agentMessages?: AgentMessage[];
+}
+
+// ── Persistent state for stateful conversations ─────────────────────────
+
+export interface AgentRunnerState {
+  messages: AgentMessage[];
+  totalInputTokens: number;
+  totalOutputTokens: number;
 }
 
 // ── Durable Object ───────────────────────────────────────────────────────
 
-export class AgentRunner extends DurableObject<Bindings> {
+export class AgentRunner extends Agent<Bindings, AgentRunnerState> {
   private initialized = false;
 
   constructor(ctx: DurableObjectState, env: Bindings) {
@@ -199,6 +212,9 @@ export class AgentRunner extends DurableObject<Bindings> {
       // Build built-in Gemini tools (only effective for google provider)
       const geminiBuiltInTools = this.buildGeminiBuiltInTools(body);
 
+      // Build resume state from persisted conversation (if stateful)
+      const resumeState = this.buildResumeState(body.agentId, userMessage);
+
       // Run the agent loop
       const result = await runAgentLoop({
         userMessage,
@@ -220,15 +236,24 @@ export class AgentRunner extends DurableObject<Bindings> {
             runId
           );
         },
+        resumeState,
       });
 
+      // Persist conversation state for stateful sessions
+      const agentMessages = this.persistConversationState(
+        body.agentId,
+        userMessage,
+        result
+      );
+
       const response: AgentRunResponse = {
         text: result.text,
         steps: result.steps,
         finishReason: result.finishReason,
         totalSteps: result.totalSteps,
         totalInputTokens: result.totalInputTokens,
         totalOutputTokens: result.totalOutputTokens,
+        ...(agentMessages && { agentMessages }),
       };
 
       // Cache the completed result
@@ -350,6 +375,9 @@ export class AgentRunner extends DurableObject<Bindings> {
 
       const geminiBuiltInTools = this.buildGeminiBuiltInTools(body);
 
+      // Build resume state from persisted conversation (if stateful)
+      const resumeState = this.buildResumeState(body.agentId, userMessage);
+
       const result = await runAgentLoop({
         userMessage,
         tools: toolDefinitions,
@@ -370,15 +398,24 @@ export class AgentRunner extends DurableObject<Bindings> {
             runId
           );
         },
+        resumeState,
       });
 
+      // Persist conversation state for stateful sessions
+      const agentMessages = this.persistConversationState(
+        body.agentId,
+        userMessage,
+        result
+      );
+
       const response: AgentRunResponse = {
         text: result.text,
         steps: result.steps,
         finishReason: result.finishReason,
         totalSteps: result.totalSteps,
         totalInputTokens: result.totalInputTokens,
         totalOutputTokens: result.totalOutputTokens,
+        ...(agentMessages && { agentMessages }),
       };
 
       // Cache the completed result
@@ -456,12 +493,70 @@ export class AgentRunner extends DurableObject<Bindings> {
             totalInputTokens: response.totalInputTokens,
             totalOutputTokens: response.totalOutputTokens,
           },
+          ...(response.agentMessages && {
+            agent_messages: response.agentMessages,
+          }),
         },
         usage,
       },
     });
   }
 
+  // ── Conversation state helpers ───────────────────────────────────────
+
+  /**
+   * If a agentId is set and previous messages exist, builds a
+   * resumeState so the agent loop continues from the prior conversation.
+   */
+  private buildResumeState(
+    agentId: string | undefined,
+    userMessage: string
+  ): AgentLoopState | undefined {
+    if (!agentId) return undefined;
+    const prev = this.state?.messages;
+    if (!prev || prev.length === 0) return undefined;
+
+    return {
+      messages: [...prev, { role: "user" as const, content: userMessage }],
+      steps: [],
+      totalInputTokens: 0,
+      totalOutputTokens: 0,
+    };
+  }
+
+  /**
+   * After an agent run completes, persists the full conversation history
+   * when agentId is set. Returns the conversation messages for
+   * inclusion in the response, or undefined for ephemeral runs.
+   */
+  private persistConversationState(
+    agentId: string | undefined,
+    userMessage: string,
+    result: AgentLoopResult
+  ): AgentMessage[] | undefined {
+    if (!agentId) return undefined;
+
+    const prevMessages = this.state?.messages ?? [];
+    const newMessages: AgentMessage[] = [
+      { role: "user", content: userMessage },
+    ];
+    for (const step of result.steps) {
+      newMessages.push(step.assistantMessage);
+      newMessages.push(...step.toolResults);
+    }
+    newMessages.push({ role: "assistant", content: result.text });
+
+    const allMessages = [...prevMessages, ...newMessages];
+    this.setState({
+      messages: allMessages,
+      totalInputTokens:
+        (this.state?.totalInputTokens ?? 0) + result.totalInputTokens,
+      totalOutputTokens:
+        (this.state?.totalOutputTokens ?? 0) + result.totalOutputTokens,
+    });
+    return allMessages;
+  }
+
   // ── Built-in Gemini tools ─────────────────────────────────────────────
 
   private buildGeminiBuiltInTools(

apps/api/src/types/agents.d.ts

@@ -0,0 +1,26 @@
+/**
+ * Type declarations for the Cloudflare Agents SDK (`agents` package).
+ * Provides the Agent base class for Durable Objects with built-in state management.
+ */
+declare module "agents" {
+  import { DurableObject } from "cloudflare:workers";
+  import type { DurableObjectState } from "cloudflare:workers";
+
+  /**
+   * Agent base class that extends DurableObject with persistent state management.
+   * Use `this.state` to read and `this.setState()` to persist state across requests.
+   *
+   * Note: Agent extends partyserver's Server which adds WebSocket/room routing
+   * in `fetch()`. Override `fetch()` directly if you only need state management
+   * and want to bypass partyserver's header requirements.
+   */
+  export class Agent<Env, State> extends DurableObject<Env> {
+    constructor(ctx: DurableObjectState, env: Env);
+
+    /** Current persistent state (undefined until first setState call) */
+    get state(): State | undefined;
+
+    /** Persist new state (replaces previous state) */
+    setState(state: State): void;
+  }
+}

packages/runtime/src/nodes/agent/base-agent-node.ts

@@ -18,6 +18,13 @@ export interface AgentNodeConfig {
 
 /** Standard inputs shared by all agent nodes */
 const AGENT_INPUTS: NodeType["inputs"] = [
+  {
+    name: "agent_id",
+    type: "string",
+    description:
+      "When set, routes to a persistent agent that maintains state and conversation history across runs",
+    required: false,
+  },
   {
     name: "instructions",
     type: "string",
@@ -106,6 +113,12 @@ const AGENT_OUTPUTS: NodeType["outputs"] = [
     description: "Token usage and cost information",
     hidden: true,
   },
+  {
+    name: "agent_messages",
+    type: "json",
+    description: "Full message history (only present when agent_id is set)",
+    hidden: true,
+  },
 ];
 
 /**
@@ -145,6 +158,7 @@ interface AgentRunResponse {
   totalSteps: number;
   totalInputTokens: number;
   totalOutputTokens: number;
+  agentMessages?: unknown[];
 }
 
 /** Event payload sent by AgentRunner DO when async execution completes */
@@ -155,6 +169,7 @@ export interface AgentCompleteEvent {
     total_steps: number;
     finish_reason: string;
     usage_metadata: { totalInputTokens: number; totalOutputTokens: number };
+    agent_messages?: unknown[];
   };
   usage: number;
   error?: string;
@@ -192,9 +207,17 @@ export abstract class BaseAgentNode extends ExecutableNode {
     config: AgentNodeConfig
   ): Promise<NodeExecution> {
     try {
-      const { instructions, input, max_steps, tools, code_mode, googleSearch } =
-        context.inputs;
+      const {
+        instructions,
+        input,
+        max_steps,
+        tools,
+        code_mode,
+        googleSearch,
+        agent_id,
+      } = context.inputs;
       const agentContext = context.inputs.context as string | undefined;
+      const agentId = agent_id as string | undefined;
 
       if (!input) {
         return this.createErrorResult("Input is required");
@@ -206,7 +229,10 @@ export abstract class BaseAgentNode extends ExecutableNode {
 
       const agentRunner = context.env.AGENT_RUNNER as DurableObjectNamespace;
       const runId = `${context.executionId}:${context.nodeId}`;
-      const doId = agentRunner.idFromName(runId);
+      // Stateful: route to a persistent DO keyed by agentId
+      // Ephemeral: route to a per-execution DO (current behavior)
+      const doName = agentId ? `agent:${agentId}` : runId;
+      const doId = agentRunner.idFromName(doName);
       const stub = agentRunner.get(doId);
 
       const response = await stub.fetch("https://agent/start", {
@@ -227,6 +253,7 @@ export abstract class BaseAgentNode extends ExecutableNode {
           codeMode: code_mode ?? false,
           googleSearch: googleSearch ?? false,
           organizationId: context.organizationId,
+          agentId,
         }),
       });
 
@@ -267,9 +294,17 @@ export abstract class BaseAgentNode extends ExecutableNode {
     config: AgentNodeConfig
   ): Promise<NodeExecution> {
     try {
-      const { instructions, input, max_steps, tools, code_mode, googleSearch } =
-        context.inputs;
+      const {
+        instructions,
+        input,
+        max_steps,
+        tools,
+        code_mode,
+        googleSearch,
+        agent_id,
+      } = context.inputs;
       const agentContext = context.inputs.context as string | undefined;
+      const agentId = agent_id as string | undefined;
 
       if (!input) {
         return this.createErrorResult("Input is required");
@@ -281,7 +316,10 @@ export abstract class BaseAgentNode extends ExecutableNode {
 
       const agentRunner = context.env.AGENT_RUNNER as DurableObjectNamespace;
       const runId = `${context.executionId ?? context.workflowId}:${context.nodeId}`;
-      const doId = agentRunner.idFromName(runId);
+      // Stateful: route to a persistent DO keyed by agentId
+      // Ephemeral: route to a per-execution DO (current behavior)
+      const doName = agentId ? `agent:${agentId}` : runId;
+      const doId = agentRunner.idFromName(doName);
       const stub = agentRunner.get(doId);
 
       const response = await stub.fetch("https://agent/run", {
@@ -299,6 +337,7 @@ export abstract class BaseAgentNode extends ExecutableNode {
           codeMode: code_mode ?? false,
           googleSearch: googleSearch ?? false,
           organizationId: context.organizationId,
+          agentId,
         }),
       });
 
@@ -330,6 +369,9 @@ export abstract class BaseAgentNode extends ExecutableNode {
             totalInputTokens: result.totalInputTokens,
             totalOutputTokens: result.totalOutputTokens,
           },
+          ...(result.agentMessages && {
+            agent_messages: result.agentMessages,
+          }),
         },
         usage
       );