cloudflare · ericclemmons · Jul 29, 2026
diff --git a/.changeset/calm-workflows-deduplicate.md b/.changeset/calm-workflows-deduplicate.md
@@ -0,0 +1,5 @@
+---
+"agents": minor
+---
+
+Add active-lifetime idempotency keys to `runWorkflow()` so concurrent calls can reuse one Workflow instance without blocking Durable Object concurrency.
diff --git a/design/rfc-workflow-idempotency.md b/design/rfc-workflow-idempotency.md
@@ -0,0 +1,49 @@
+Status: proposed
+
+# Active workflow idempotency keys
+
+## The problem
+
+`runWorkflow()` creates a Workflow instance before it writes the Agent's local
+tracking row. Two requests can therefore both observe no active workflow, call
+`runWorkflow()`, and create overlapping instances. Serializing the whole
+operation with `blockConcurrencyWhile()` holds a Durable Object lock across an
+external Workflow RPC and can reset the object when that RPC is slow.
+
+Workflow instance IDs are not the right logical singleton key. Instances and
+their platform state can be retained after completion, while callers need to
+start a new instance as soon as the active run reaches a terminal state.
+
+## The proposal
+
+Add `RunWorkflowOptions.idempotencyKey`. Before the Workflow RPC, synchronously
+insert the generated Workflow ID and idempotency key into Agent SQLite. A
+partial unique index on `(workflow_name, idempotency_key)` covers non-terminal
+rows. Durable Object input gates make that local claim atomic without holding a
+lock across external I/O.
+
+If another active row already owns the key, return its Workflow ID. When the
+row becomes `complete`, `errored`, or `terminated`, it leaves the partial index
+and a later call can claim the same logical key with a new Workflow instance
+ID. Historical tracking and platform Workflow retention remain unchanged.
+
+The key is additionally scoped by the Agent or sub-agent facet because each
+facet has its own SQLite storage.
+
+## Alternatives
+
+- Query active workflows before creation: this leaves a time-of-check to
+  time-of-use race.
+- Use the idempotency key as the Workflow instance ID: retained Workflow IDs
+  prevent starting the next generation with the same key.
+- Wrap creation in `blockConcurrencyWhile()`: this holds a global Durable
+  Object lock across external I/O and is subject to the callback deadline.
+- Add queue, cancel, or replace policies now: these require durable pending
+  payloads and lifecycle orchestration. They should build on this claim
+  primitive in a separate concurrency-policy design.
+
+## The decision
+
+Pending review. The initial conflict behavior is `use existing`: concurrent
+calls return the active Workflow ID. The API can grow a separate explicit
+concurrency policy without changing the meaning of `idempotencyKey`.
diff --git a/docs/agents/workflows.md b/docs/agents/workflows.md
@@ -246,7 +246,8 @@ const instanceId = await this.runWorkflow(
   {
     id: "custom-id", // optional - auto-generated if not provided
     metadata: { userId: "user-456", priority: "high" }, // optional - for querying
-    agentBinding: "MyAgent" // optional - auto-detected from class name if not provided
+    agentBinding: "MyAgent", // optional - auto-detected from class name if not provided
+    idempotencyKey: "process-user-456" // optional - deduplicates active instances
   }
 );
 ```
@@ -258,6 +259,7 @@ const instanceId = await this.runWorkflow(
 - `options.id` - Custom workflow ID (auto-generated if not provided)
 - `options.metadata` - Optional metadata stored for querying (not passed to workflow)
 - `options.agentBinding` - Agent binding name (auto-detected from class name if not provided). When called from a sub-agent, this is the root Agent binding name.
+- `options.idempotencyKey` - Logical key that allows at most one active workflow for the same workflow binding in this Agent or sub-agent facet. Concurrent calls return the existing workflow ID. The key becomes available again after the tracked workflow completes, errors, or is terminated.
 
 **Returns:** Workflow instance ID
 
@@ -525,21 +527,24 @@ Workflows started with `runWorkflow()` are automatically tracked in the originat
 
 ### `cf_agents_workflows` Table
 
-| Column          | Type    | Description                     |
-| --------------- | ------- | ------------------------------- |
-| `id`            | TEXT    | Internal row ID                 |
-| `workflow_id`   | TEXT    | Cloudflare workflow instance ID |
-| `workflow_name` | TEXT    | Workflow binding name           |
-| `status`        | TEXT    | Current status                  |
-| `metadata`      | TEXT    | JSON metadata (for querying)    |
-| `error_name`    | TEXT    | Error name (if failed)          |
-| `error_message` | TEXT    | Error message (if failed)       |
-| `created_at`    | INTEGER | Unix timestamp                  |
-| `updated_at`    | INTEGER | Unix timestamp                  |
-| `completed_at`  | INTEGER | Unix timestamp (when done)      |
+| Column            | Type    | Description                     |
+| ----------------- | ------- | ------------------------------- |
+| `id`              | TEXT    | Internal row ID                 |
+| `workflow_id`     | TEXT    | Cloudflare workflow instance ID |
+| `idempotency_key` | TEXT    | Optional active-instance key    |
+| `workflow_name`   | TEXT    | Workflow binding name           |
+| `status`          | TEXT    | Current status                  |
+| `metadata`        | TEXT    | JSON metadata (for querying)    |
+| `error_name`      | TEXT    | Error name (if failed)          |
+| `error_message`   | TEXT    | Error message (if failed)       |
+| `created_at`      | INTEGER | Unix timestamp                  |
+| `updated_at`      | INTEGER | Unix timestamp                  |
+| `completed_at`    | INTEGER | Unix timestamp (when done)      |
 
 Note: Workflow params and output are not stored by default. Use `metadata` to store queryable information, and store large payloads in your own tables if needed.
 
+An idempotency key is scoped to the workflow binding and the originating Agent storage. It does not replace the Workflow instance ID and does not shorten Workflow or tracking retention. Terminal history rows retain their key, but only non-terminal rows reserve it.
+
 ### Workflow Status Values
 
 - `queued` - Waiting to start

diff --git a/packages/agents/src/index.ts b/packages/agents/src/index.ts
@@ -1125,7 +1125,7 @@ type AgentToolRecoveryInspection =
  * The constructor stores this as a row in cf_agents_state and checks it
  * on wake to skip DDL on established DOs.
  */
-const CURRENT_SCHEMA_VERSION = 11;
+const CURRENT_SCHEMA_VERSION = 12;
 
 const SCHEMA_VERSION_ROW_ID = "cf_schema_version";
 const STATE_ROW_ID = "cf_state_row_id";
@@ -2126,6 +2126,7 @@ export class Agent<
         CREATE TABLE IF NOT EXISTS cf_agents_workflows (
           id TEXT PRIMARY KEY NOT NULL,
           workflow_id TEXT NOT NULL UNIQUE,
+          idempotency_key TEXT,
           workflow_name TEXT NOT NULL,
           status TEXT NOT NULL CHECK(status IN (
             'queued', 'running', 'paused', 'errored',
@@ -2149,6 +2150,17 @@ export class Agent<
         CREATE INDEX IF NOT EXISTS idx_workflows_name ON cf_agents_workflows(workflow_name)
       `;
 
+      addColumnIfNotExists(
+        "ALTER TABLE cf_agents_workflows ADD COLUMN idempotency_key TEXT"
+      );
+
+      this.sql`
+        CREATE UNIQUE INDEX IF NOT EXISTS idx_workflows_active_idempotency_key
+        ON cf_agents_workflows(workflow_name, idempotency_key)
+        WHERE idempotency_key IS NOT NULL
+          AND status NOT IN ('complete', 'errored', 'terminated')
+      `;
+
       // Clean up legacy STATE_WAS_CHANGED rows from the single-row state optimization
       this.ctx.storage.sql.exec(
         "DELETE FROM cf_agents_state WHERE id = ?",
@@ -11231,8 +11243,46 @@ export class Agent<
       );
     }
 
+    const idempotencyKey = options?.idempotencyKey;
+    if (idempotencyKey !== undefined && idempotencyKey.trim() === "") {
+      throw new Error("idempotencyKey must not be blank");
+    }
+
     // Workflows instance IDs must start with [a-zA-Z0-9_].
     const workflowId = options?.id ?? `wf_${nanoid()}`;
+    const trackingId = nanoid();
+    const metadataJson = options?.metadata
+      ? JSON.stringify(options.metadata)
+      : null;
+
+    if (idempotencyKey !== undefined) {
+      const claimed = this.sql<{ workflow_id: string }>`
+        INSERT OR IGNORE INTO cf_agents_workflows
+          (id, workflow_id, idempotency_key, workflow_name, status, metadata)
+        VALUES
+          (${trackingId}, ${workflowId}, ${idempotencyKey}, ${workflowName}, 'queued', ${metadataJson})
+        RETURNING workflow_id
+      `;
+
+      if (claimed.length === 0) {
+        const [existing] = this.sql<{ workflow_id: string }>`
+          SELECT workflow_id
+          FROM cf_agents_workflows
+          WHERE workflow_name = ${workflowName}
+            AND idempotency_key = ${idempotencyKey}
+            AND status NOT IN ('complete', 'errored', 'terminated')
+          LIMIT 1
+        `;
+
+        if (existing) {
+          return existing.workflow_id;
+        }
+
+        throw new Error(
+          `Workflow with ID "${workflowId}" is already being tracked`
+        );
+      }
+    }
 
     // Inject agent identity and workflow name into params
     const augmentedParams = {
@@ -11246,32 +11296,40 @@ export class Agent<
       __agentOrigin: agentOrigin
     };
 
-    // Create the workflow instance
-    const instance = await workflow.create({
-      id: workflowId,
-      params: augmentedParams
-    });
-
-    // Track the workflow in our database
-    const id = nanoid();
-    const metadataJson = options?.metadata
-      ? JSON.stringify(options.metadata)
-      : null;
+    let instance: { id: string };
     try {
-      this.sql`
-        INSERT INTO cf_agents_workflows (id, workflow_id, workflow_name, status, metadata)
-        VALUES (${id}, ${instance.id}, ${workflowName}, 'queued', ${metadataJson})
-      `;
-    } catch (e) {
-      if (
-        e instanceof Error &&
-        e.message.includes("UNIQUE constraint failed")
-      ) {
-        throw new Error(
-          `Workflow with ID "${workflowId}" is already being tracked`
-        );
+      instance = await workflow.create({
+        id: workflowId,
+        params: augmentedParams
+      });
+    } catch (error) {
+      if (idempotencyKey !== undefined) {
+        this.sql`
+          DELETE FROM cf_agents_workflows
+          WHERE workflow_id = ${workflowId} AND status = 'queued'
+        `;
+      }
+
+      throw error;
+    }
+
+    if (idempotencyKey === undefined) {
+      try {
+        this.sql`
+          INSERT INTO cf_agents_workflows (id, workflow_id, workflow_name, status, metadata)
+          VALUES (${trackingId}, ${instance.id}, ${workflowName}, 'queued', ${metadataJson})
+        `;
+      } catch (e) {
+        if (
+          e instanceof Error &&
+          e.message.includes("UNIQUE constraint failed")
+        ) {
+          throw new Error(
+            `Workflow with ID "${workflowId}" is already being tracked`
+          );
+        }
+        throw e;
       }
-      throw e;
     }
 
     this._emit("workflow:start", { workflowId: instance.id, workflowName });
@@ -12011,6 +12069,7 @@ export class Agent<
     return {
       id: row.id,
       workflowId: row.workflow_id,
+      idempotencyKey: row.idempotency_key,
       workflowName: row.workflow_name,
       status: row.status,
       metadata: row.metadata ? JSON.parse(row.metadata) : null,

diff --git a/packages/agents/src/tests/agents/workflow.ts b/packages/agents/src/tests/agents/workflow.ts
@@ -691,6 +691,13 @@ export class TestWorkflowAgent extends Agent {
     return this.runWorkflow("TEST_WORKFLOW", params, { id: workflowId });
   }
 
+  async runIdempotentWorkflowTest(
+    idempotencyKey: string,
+    params: { taskId: string; shouldFail?: boolean; waitForApproval?: boolean }
+  ): Promise<string> {
+    return this.runWorkflow("TEST_WORKFLOW", params, { idempotencyKey });
+  }
+
   // Start a simple workflow
   async runSimpleWorkflowTest(
     workflowId: string,

diff --git a/packages/agents/src/tests/schema-and-state-optimization.test.ts b/packages/agents/src/tests/schema-and-state-optimization.test.ts
@@ -91,6 +91,7 @@ const EXPECTED_SCHEMA_DDL = [
   `CREATE TABLE cf_agents_workflows (
           id TEXT PRIMARY KEY NOT NULL,
           workflow_id TEXT NOT NULL UNIQUE,
+          idempotency_key TEXT,
           workflow_name TEXT NOT NULL,
           status TEXT NOT NULL CHECK(status IN (
             'queued', 'running', 'paused', 'errored',
@@ -106,7 +107,7 @@ const EXPECTED_SCHEMA_DDL = [
         )`
 ];
 
-const EXPECTED_SCHEMA_VERSION = 11;
+const EXPECTED_SCHEMA_VERSION = 12;
 
 describe("schema DDL snapshot", () => {
   it("should match the expected DDL for the current schema version", async () => {

diff --git a/packages/agents/src/tests/workflow.test.ts b/packages/agents/src/tests/workflow.test.ts
@@ -972,6 +972,64 @@ describe("workflow operations", () => {
     });
   });
 
+  describe("workflow idempotency", () => {
+    it("returns one active workflow for concurrent calls with the same key", async () => {
+      const agentStub = await getTestAgent("workflow-idempotency-concurrent");
+
+      const [firstId, secondId] = await Promise.all([
+        agentStub.runIdempotentWorkflowTest("singleton", {
+          taskId: "first",
+          waitForApproval: true
+        }),
+        agentStub.runIdempotentWorkflowTest("singleton", {
+          taskId: "second",
+          waitForApproval: true
+        })
+      ]);
+
+      expect(secondId).toBe(firstId);
+
+      const workflows = (await agentStub.getWorkflowsForTest(
+        {}
+      )) as WorkflowInfo[];
+      expect(workflows).toHaveLength(1);
+      expect(workflows[0].idempotencyKey).toBe("singleton");
+    });
+
+    it("makes a key available after the active workflow terminates", async () => {
+      const agentStub = await getTestAgent("workflow-idempotency-reuse");
+      const firstId = await agentStub.runIdempotentWorkflowTest("singleton", {
+        taskId: "first",
+        waitForApproval: true
+      });
+
+      const termination = await agentStub.expectThrow(
+        "terminateWorkflow",
+        firstId
+      );
+      expect(termination.threw).toBe(false);
+
+      const secondId = await agentStub.runIdempotentWorkflowTest("singleton", {
+        taskId: "second",
+        waitForApproval: true
+      });
+
+      expect(secondId).not.toBe(firstId);
+    });
+
+    it("rejects blank keys", async () => {
+      const agentStub = await getTestAgent("workflow-idempotency-blank");
+      const result = await agentStub.expectThrow(
+        "runIdempotentWorkflowTest",
+        "   ",
+        { taskId: "blank" }
+      );
+
+      expect(result.threw).toBe(true);
+      expect(result.message).toContain("idempotencyKey must not be blank");
+    });
+  });
+
   describe("workflow lifecycle integration", () => {
     async function waitForRunning(
       agentStub: Awaited<ReturnType<typeof getTestAgent>>,

diff --git a/packages/agents/src/workflow-types.ts b/packages/agents/src/workflow-types.ts
@@ -202,6 +202,8 @@ export type WorkflowTrackingRow = {
   id: string;
   /** Cloudflare Workflow instance ID */
   workflow_id: string;
+  /** Logical key that deduplicates active workflow instances */
+  idempotency_key: string | null;
   /** Workflow binding name */
   workflow_name: string;
   /** Current workflow status */
@@ -230,6 +232,13 @@ export type RunWorkflowOptions = {
   metadata?: Record<string, unknown>;
   /** Agent binding name (auto-detected from class name if not provided) */
   agentBinding?: string;
+  /**
+   * Return the existing active workflow for this key instead of creating a
+   * duplicate. The key is scoped to the workflow binding and originating
+   * Agent (or sub-agent facet), and becomes available again when the tracked
+   * workflow reaches a terminal status.
+   */
+  idempotencyKey?: string;
 };
 
 /**
@@ -250,6 +259,8 @@ export type WorkflowInfo = {
   id: string;
   /** Cloudflare Workflow instance ID */
   workflowId: string;
+  /** Logical key used to deduplicate active workflow instances */
+  idempotencyKey: string | null;
   /** Workflow binding name */
   workflowName: string;
   /** Current workflow status */