Cloudflare Storage

The Cloudflare storage package (@helix-agents/store-cloudflare) provides state and stream storage for edge deployment. There are two sets of stores depending on which runtime approach you use:

Approach	State Store	Stream Manager	Package
Workflows	D1StateStore	DurableObjectStreamManager	@helix-agents/store-cloudflare
Durable Objects	DOStateStore	DOStreamManager	@helix-agents/runtime-cloudflare

Choose Based on Runtime

• Workflows runtime: Use D1StateStore + DurableObjectStreamManager from @helix-agents/store-cloudflare
• DO runtime: Use DOStateStore + DOStreamManager from @helix-agents/runtime-cloudflare (built into AgentServer)

When to Use

Good fit:

• Cloudflare Workers deployments
• Global edge distribution
• Serverless architecture
• Existing Cloudflare infrastructure

Not ideal for:

• Non-Cloudflare environments
• High write throughput (D1 limitations - Workflows only)
• Complex queries (limited SQL support)

Installation

npm install @helix-agents/store-cloudflare

Entry Points and Import Patterns

The package provides three entry points to support different deployment contexts:

Entry Point	Use Case	Contains
@helix-agents/store-cloudflare	Worker entry points (full package)	Everything
@helix-agents/store-cloudflare/d1	Next.js routes, any Node.js context	D1StateStore, D1UsageStore, migrations
@helix-agents/store-cloudflare/stream	Worker entry points only	StreamServer, DurableObjectStreamManager

Why Split Entry Points?

The /stream entry point imports partyserver which depends on cloudflare:workers - a Cloudflare-specific module that cannot be imported in:

• Next.js API routes (even with OpenNext for Cloudflare)
• Node.js environments
• Any code that runs during Next.js build

This is because:

1. Build-time evaluation: Next.js evaluates all imports during build, even for server routes
2. cloudflare:workers is runtime-only: The cloudflare:workers URL scheme only works inside the Cloudflare Workers runtime
3. Durable Object class definitions vs bindings: You CAN access Durable Objects via env bindings, but you CANNOT import the class definition outside a Worker

Correct Import Patterns

In Next.js API routes (OpenNext):

// ✅ CORRECT - Use /d1 entry point
import { D1StateStore, runMigration } from '@helix-agents/store-cloudflare/d1';

export async function GET(request: Request) {
  const { env } = getCloudflareContext();
  const stateStore = new D1StateStore({ database: env.DATABASE });
  // Access Durable Objects via binding (NOT import)
  const streamStub = env.STREAMS.get(env.STREAMS.idFromName('my-stream'));
}

// ❌ WRONG - Will fail during Next.js build
import { D1StateStore } from '@helix-agents/store-cloudflare';
import { StreamServer } from '@helix-agents/store-cloudflare/stream';

In Cloudflare Worker entry points:

// ✅ CORRECT - Worker entry points can import everything
import { StreamServer } from '@helix-agents/store-cloudflare/stream';
import { D1StateStore } from '@helix-agents/store-cloudflare/d1';

// Export Durable Object class for Cloudflare to instantiate
export { StreamServer };

export default {
  async fetch(request: Request, env: Env) {
    // Now you can use both D1 and stream functionality
    const stateStore = new D1StateStore({ database: env.DB });
    // Access StreamServer via binding
    const stub = env.STREAMS.get(env.STREAMS.idFromName('stream-id'));
  },
};

Architecture Pattern for Next.js + Cloudflare

When using OpenNext with Cloudflare:

graph TB
    subgraph NextJS ["Next.js App (OpenNext)"]
        subgraph API ["API Routes"]
            A1["Import from @helix-agents/store-cloudflare/d1"]
            A2["Query D1 directly for state/messages"]
            A3["Access workers via env.WORKER_NAME service binding"]
        end
    end

    NextJS -->|"Service Binding"| AgentWorker

    subgraph AgentWorker ["Agent Worker (separate worker)"]
        subgraph Entry ["Entry Point (src/index.ts)"]
            E1["Export StreamServer (Durable Object class)"]
            E2["Import from any entry point"]
            E3["Handle agent execution"]
        end
    end

Prerequisites

You need:

• Cloudflare account with Workers Paid plan
• D1 database created
• Durable Object namespace configured

D1StateStore

Setup

Configure D1 in wrangler.toml:

[[d1_databases]]
binding = "DB"
database_name = "agent-state"
database_id = "your-database-id"

Migrations

The D1StateStore uses programmatic migrations that are automatically applied. You should call runMigration() on startup:

import { D1StateStore, runMigration } from '@helix-agents/store-cloudflare';

export default {
  async fetch(request: Request, env: Env) {
    // Run migrations on startup (safe to call every request - no-op if already migrated)
    await runMigration(env.DB);

    const stateStore = new D1StateStore({ database: env.DB });
    // ...
  },
};

runMigration() is safe to call on every boot, from any number of isolates at once:

• Claim-first. Each migration is one batch whose first statement records its version row. Of several isolates racing for the same version exactly one applies it; the others apply nothing and continue from what it recorded, so a concurrent boot no longer fails with duplicate column name, and a data statement in a migration never runs twice.
• Honest errors. A migration that fails for any other reason rethrows its original error. getAgentsMigrationVersion() returns 0 only when the version table does not exist yet; a failed read throws instead of being reported as version 0.

The runner is shared with @helix-agents/memory-cloudflare (see Core: SQL migration runner).

The framework creates these tables automatically (all prefixed with __agents_). The current schema is V23; migrations apply progressively from V1. The block below shows the V1–V9 core; later migrations add columns (see Commit columns (V17 / V18) and V19).

-- Core state table (V1 baseline + V3 user context + V7 HITL columns
--                  + V8 completed_client_tool_calls + V9 suspension_context)
CREATE TABLE IF NOT EXISTS __agents_states (
  session_id TEXT PRIMARY KEY,
  agent_type TEXT NOT NULL,
  parent_session_id TEXT,
  stream_id TEXT NOT NULL,
  step_count INTEGER NOT NULL DEFAULT 0,
  status TEXT NOT NULL DEFAULT 'active',
  output TEXT,
  error TEXT,
  failure_reason TEXT,                  -- V7: cascade discriminator
  aborted INTEGER NOT NULL DEFAULT 0,
  abort_reason TEXT,
  custom_state TEXT NOT NULL DEFAULT '{}',
  user_id TEXT,                         -- V3
  tags TEXT,                            -- V3 (JSON array)
  metadata TEXT,                        -- V3 (JSON object)
  pending_client_tool_calls TEXT,       -- V7: HITL pending map
  completed_client_tool_calls TEXT,     -- V8: idempotency tombstones
  expires_at INTEGER,                   -- V7 (also packed in suspension_context)
  suspension_context TEXT DEFAULT NULL, -- V9: stateless HITL/client-tools
  created_at INTEGER NOT NULL,
  updated_at INTEGER NOT NULL
);

-- Partial index for operator cleanup of abandoned suspended sessions
CREATE INDEX IF NOT EXISTS idx___agents_states_suspension_expires_at
  ON __agents_states(json_extract(suspension_context, '$.expiresAt'))
  WHERE suspension_context IS NOT NULL;

-- Messages table (separated for O(1) append)
CREATE TABLE IF NOT EXISTS __agents_messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  session_id TEXT NOT NULL,
  sequence INTEGER NOT NULL,
  message TEXT NOT NULL,
  created_at INTEGER NOT NULL,
  UNIQUE(session_id, sequence)
);

-- Sub-session refs (V2 adds remote_json; V4 adds mode + name for persistent children)
CREATE TABLE IF NOT EXISTS __agents_sub_session_refs (
  session_id TEXT NOT NULL,
  sub_session_id TEXT NOT NULL,
  agent_type TEXT NOT NULL,
  parent_tool_call_id TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active',
  remote_json TEXT,                     -- V2
  mode TEXT,                            -- V4: 'persistent' | 'ephemeral'
  name TEXT,                            -- V4: persistent child name
  started_at INTEGER NOT NULL,
  completed_at INTEGER,
  PRIMARY KEY (session_id, sub_session_id)
);

-- Run tracking (__agents_runs is part of the V1 baseline)
CREATE TABLE IF NOT EXISTS __agents_runs (
  run_id TEXT PRIMARY KEY,
  session_id TEXT NOT NULL,
  turn INTEGER NOT NULL,
  status TEXT NOT NULL,
  step_count INTEGER NOT NULL DEFAULT 0,
  started_at INTEGER NOT NULL,
  completed_at INTEGER
);

suspension_context packs suspendedAwaitingChildren, suspendedStepId, tracingContext, and expiresAt as a single JSON-encoded blob, written and read atomically as part of saveStateAndPromoteStaging.

Additional tables are created for usage tracking, checkpoints, stop records, and message-merge staging. See d1-migrations.ts for the full schema progression V1 → V20.

Commit columns (V17 / V18)

V17 and V18 add the commit model's bookkeeping. Every column is nullable:

Migration	Adds
V17	__agents_checkpoints.kind / run_id / keep_count; __agents_runs.lease_until / start_digest / owner_token / owner_unknown_since; __agents_states.current_run_id (set only by a run-start commit) / latest_truncate_checkpoint_id
V18	__agents_runs.entry_step_count / input_row_count / parent_session_id / parent_tool_call_id; the __agents_abandoned_run_starts tombstone table (abandonRunStart)

runMigration() applies them like every other migration.

V19: run completion reason

V19 adds a nullable __agents_runs.completion_reason TEXT column (added only when missing). It holds RunMetadata.completionReason, written only for a run that ended at a stopWhen stop ('stop_when') or a maxSteps cut-off ('max_steps'); retry() reads it to continue a cut-off run with a fresh budget. Rows written before V19 read back with no reason. runMigration() applies it automatically; no backfill is needed.

Delete cascade columns (V21)

V21 adds a partial index on __agents_states(parent_session_id) (WHERE parent_session_id IS NOT NULL) for the deleteSession cascade's walk, and the __agents_sub_session_refs.location column ('local' or 'remote'). The collapsed initial schema carries the same index. runMigration() applies it automatically.

Deleting a session

deleteSession is two steps. Step 1 reads the whole id set once with a WITH RECURSIVE … SELECT over the session's local refs and the parent_session_id index, excluding remote refs (the UNION stops cycles). Step 2 is one db.batch of DELETE … WHERE session_id IN (SELECT value FROM json_each(?)) per session-keyed table (states, messages, runs, checkpoints, refs, staging, interrupt flags, abandoned run starts and __agents_usage), each binding the id set as one JSON array. Reading first matters: a CTE inside each batch statement would re-run against edges that earlier deletes in the same batch had already removed. The deletes are atomic; the read and the batch are not, so a child created in between survives the delete. That window only exists for a delete with a live run, which is undefined. D1 deletes the tree's usage rows (the memory and Redis stores cannot), so a re-spawned companion's earlier usage rows go too, as on Postgres: the re-spawn deletes the companion's old session first. A tree read that does not name the session itself throws a D1StateError instead of deleting nothing. See Deleting a session.

Atomic Writes (commits)

Every commit (appendMessages, truncateMessages, commitState, saveStateAndPromoteStaging / promoteStaging, cloneSession, and the run-start commit) is one db.batch([...]), which Cloudflare D1 executes as a single auto-commit transaction. The batch is guarded: it pins the session's version and its current_run_id (and, for a fenced commit, that the run is still live), and writes the rows, the state, the checkpoint and the latest-checkpoint pointer together. The run-start commit additionally pins the current run's status, lease_until and owner_token to the values its owner guard decided on, so a lease renewed between the store's reads makes the batch a no-op and the decision is retried on the renewed lease. For saveStateAndPromoteStaging the batch:

1. INSERTs new messages (packed into JSON-array statements below D1's 2 MB value limit)
2. UPDATEs __agents_states with the new state, version (post-bump), and suspension context
3. Promotes the staging rows
4. INSERTs the step checkpoint row and moves the pointer to it

This guarantees the cross-runtime invariant C-1: pendingClientToolCalls, clientToolCallOwnership, completed phase-1 messages, the checkpoint, and suspendedStepId all become visible together. saveState is not a commit: it writes no checkpoint and never moves the pointer.

When the guard fails because other writers keep landing between the pre-read and the batch, the store re-reads and retries. If every attempt loses, it throws D1CommitContentionError (exported by @helix-agents/store-cloudflare). It extends StaleStateError (and keeps its name), so runtime retry paths treat it as retryable; its message names what diverged on the last attempt (the state version, the current-run pointer, the checkpoint pointer, the fenced run's liveness or the run-start view).

Breaking: cloneSession(from, to, { fromCheckpointId }) now validates the checkpoint through core resolveCheckpoint and throws state_checkpoint_not_found / state_checkpoint_foreign before writing anything; D1 used to clone the whole session silently when the checkpoint was missing. The clone is one batch with its clone checkpoint. See the upgrade guide.

Usage

import { D1StateStore } from '@helix-agents/store-cloudflare';

export default {
  async fetch(request: Request, env: Env) {
    const stateStore = new D1StateStore({ database: env.DB });

    // Save state (no checkpoint; runtimes pass { fenceRunId })
    await stateStore.saveState(sessionId, sessionState);

    // Load state
    const loaded = await stateStore.loadState('session-123');

    // Delete state
    await stateStore.deleteState('session-123');
  },
};

Atomic Operations

// Append messages (a commit: fenced on commit.runId, writes an 'append' checkpoint)
await stateStore.appendMessages('session-123', [{ role: 'assistant', content: 'Hello!' }], {
  streamSequence,
  runId,
});

// Get messages with pagination
const { messages, hasMore } = await stateStore.getMessages('session-123', { limit: 100 });

// Update state atomically
await stateStore.saveState('session-123', updatedState);

Message Truncation

For crash recovery scenarios:

// Truncate to specific message count (removes orphaned messages) - a 'truncate' commit
await stateStore.truncateMessages('session-123', messageCount, { streamSequence, runId });

This removes messages beyond the specified count, used during crash recovery to sync messages with a checkpoint's messageCount field (a from_checkpoint rewind or a retry() carries its rewind in its run-start commit instead, startRun.truncateTo). On D1 the delete and the update of the denormalized states.message_count (read by listSessions) run in one batch, so listSessions().messageCount stays equal to the live row count after a truncate (RM-30).

Stop records

D1StateStore (like every store: memory, Redis, Postgres, DO) implements the run-scoped stop record on the SessionStateStore interface (WA MR2). A stop names the run it targets, and nothing clears it.

// Record a stop on the session's current live run (core `stopRun` calls this)
const result = await stateStore.requestStop('session-123', {
  kind: 'interrupt',
  reason: 'user_requested',
  target: { kind: 'current' },
  deadlineMs: null,
});
// { recorded: true, stop } | { recorded: false, reason: 'no_live_run' }

// Read one run's stop (called by the run at its boundaries)
const stop = await stateStore.readStop('session-123', runId);

Migration V20 creates the table and drops the old interrupt-flag table:

CREATE TABLE IF NOT EXISTS __agents_stop_requests (
  session_id TEXT NOT NULL,
  run_id TEXT NOT NULL,
  kind TEXT NOT NULL CHECK (kind IN ('interrupt','abort')),
  reason TEXT NOT NULL,
  requested_at INTEGER NOT NULL,
  deadline_at INTEGER,
  PRIMARY KEY (session_id, run_id)
);

See Interrupt and Resume for complete documentation on stop behavior, including sub-agent propagation, and the upgrade guide.

Programmatic Migrations

For dynamic environments:

import { D1StateStore, runMigration } from '@helix-agents/store-cloudflare';

// Run migrations on first request
let migrated = false;

export default {
  async fetch(request: Request, env: Env) {
    if (!migrated) {
      await runMigration(env.DB);
      migrated = true;
    }

    const stateStore = new D1StateStore({ database: env.DB });
    // ...
  },
};

Stream info for unknown streams

DurableObjectStreamManager.getStreamInfo(streamId) returns null for a stream that does not exist (nothing has written, initialised, claimed or ended it), as every other stream manager does. A stream a run has claimed counts as existing even before its first chunk. Anything that attaches to a stream by id (the resumable stream handler, createResumableReader) therefore answers "not found" for an unknown id.

DOStreamManager (Durable Objects)

Setup

Configure Durable Objects in wrangler.toml:

[durable_objects]
bindings = [
  { name = "STREAM_MANAGER", class_name = "StreamManagerDO" }
]

[[migrations]]
tag = "v1"
new_classes = ["StreamManagerDO"]

Durable Object Implementation

// src/stream-manager-do.ts
import { DurableObject } from 'cloudflare:workers';
import type { StreamChunk } from '@helix-agents/core';

interface StreamState {
  chunks: StreamChunk[];
  status: 'active' | 'ended' | 'failed';
  endedAt?: number;
  error?: string;
}

export class StreamManagerDO extends DurableObject {
  private state: StreamState = {
    chunks: [],
    status: 'active',
  };

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    // Load state from storage
    ctx.blockConcurrencyWhile(async () => {
      const stored = await ctx.storage.get<StreamState>('state');
      if (stored) {
        this.state = stored;
      }
    });
  }

  async write(chunk: StreamChunk): Promise<void> {
    if (this.state.status !== 'active') {
      throw new Error(`Cannot write to ${this.state.status} stream`);
    }

    this.state.chunks.push(chunk);
    await this.ctx.storage.put('state', this.state);

    // Notify WebSocket clients
    for (const ws of this.ctx.getWebSockets()) {
      ws.send(JSON.stringify(chunk));
    }
  }

  async read(fromOffset: number = 0): Promise<StreamChunk[]> {
    return this.state.chunks.slice(fromOffset);
  }

  async end(finalOutput?: unknown): Promise<void> {
    this.state.status = 'ended';
    this.state.endedAt = Date.now();
    await this.ctx.storage.put('state', this.state);

    // Close WebSocket connections
    for (const ws of this.ctx.getWebSockets()) {
      ws.close(1000, 'Stream ended');
    }
  }

  async fail(error: string): Promise<void> {
    this.state.status = 'failed';
    this.state.error = error;
    await this.ctx.storage.put('state', this.state);

    for (const ws of this.ctx.getWebSockets()) {
      ws.close(1011, error);
    }
  }

  async getInfo(): Promise<{ status: string; chunkCount: number }> {
    return {
      status: this.state.status,
      chunkCount: this.state.chunks.length,
    };
  }

  async fetch(request: Request): Promise<Response> {
    // WebSocket upgrade for real-time streaming
    if (request.headers.get('Upgrade') === 'websocket') {
      const [client, server] = Object.values(new WebSocketPair());
      this.ctx.acceptWebSocket(server);

      // Send historical chunks
      for (const chunk of this.state.chunks) {
        server.send(JSON.stringify(chunk));
      }

      return new Response(null, { status: 101, webSocket: client });
    }

    return new Response('Expected WebSocket', { status: 400 });
  }
}

Stream Manager Wrapper

// src/do-stream-manager.ts
import type { StreamManager, StreamWriter, StreamReader, StreamChunk } from '@helix-agents/core';

export class DOStreamManager implements StreamManager {
  constructor(private readonly binding: DurableObjectNamespace) {}

  async createWriter(streamId: string, agentId: string, agentType: string): Promise<StreamWriter> {
    const stub = this.getStub(streamId);

    return {
      write: async (chunk: StreamChunk) => {
        await stub.write(chunk);
      },
      close: async () => {
        // Writer close is a no-op - stream continues
      },
    };
  }

  async createReader(streamId: string): Promise<StreamReader | null> {
    const stub = this.getStub(streamId);
    const info = await stub.getInfo();

    if (info.status === 'failed') {
      return null;
    }

    let offset = 0;
    let done = false;

    return {
      [Symbol.asyncIterator]: () => ({
        next: async () => {
          if (done) {
            return { done: true, value: undefined };
          }

          const chunks = await stub.read(offset);

          if (chunks.length > 0) {
            offset += chunks.length;
            return { done: false, value: chunks[0] };
          }

          // Check if stream ended
          const currentInfo = await stub.getInfo();
          if (currentInfo.status === 'ended') {
            done = true;
            return { done: true, value: undefined };
          }

          // Poll for new chunks (in practice, use WebSocket)
          await new Promise((r) => setTimeout(r, 100));
          return this.next();
        },
      }),
      close: async () => {},
    };
  }

  async endStream(streamId: string, finalOutput?: unknown): Promise<void> {
    const stub = this.getStub(streamId);
    await stub.end(finalOutput);
  }

  async failStream(streamId: string, error: string): Promise<void> {
    const stub = this.getStub(streamId);
    await stub.fail(error);
  }

  private getStub(streamId: string) {
    const id = this.binding.idFromName(streamId);
    return this.binding.get(id);
  }
}

Export Durable Object

// src/index.ts
export { StreamManagerDO } from './stream-manager-do';

Complete Example

// src/index.ts
import { CloudflareAgentExecutor } from '@helix-agents/runtime-cloudflare';
import { D1StateStore, runMigration } from '@helix-agents/store-cloudflare';
import { DOStreamManager } from './do-stream-manager';
import { StreamManagerDO } from './stream-manager-do';
import { registry } from './agents';

export { StreamManagerDO };

export interface Env {
  DB: D1Database;
  STREAM_MANAGER: DurableObjectNamespace;
  AGENT_WORKFLOW: Workflow;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    // Run migrations on startup
    await runMigration(env.DB);

    const stateStore = new D1StateStore({ database: env.DB });
    const streamManager = new DOStreamManager(env.STREAM_MANAGER);

    const executor = new CloudflareAgentExecutor({
      workflowBinding: env.AGENT_WORKFLOW,
      stateStore,
      streamManager,
    });

    // POST /execute
    if (url.pathname === '/execute' && request.method === 'POST') {
      const { agentType, message, sessionId } = await request.json();
      const agent = registry.get(agentType);

      const handle = await executor.execute(agent, { message }, { sessionId });

      return Response.json({
        sessionId: handle.sessionId,
        streamUrl: `/stream/${handle.sessionId}`,
      });
    }

    // GET /stream/:sessionId (SSE)
    if (url.pathname.startsWith('/stream/')) {
      const sessionId = url.pathname.split('/').pop()!;
      const handle = await executor.getHandle(registry.get('default'), sessionId);

      if (!handle) {
        return new Response('Not found', { status: 404 });
      }

      const stream = await handle.stream();
      if (!stream) {
        return new Response('Stream not available', { status: 404 });
      }

      return new Response(
        new ReadableStream({
          async start(controller) {
            const encoder = new TextEncoder();
            for await (const chunk of stream) {
              controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
            }
            controller.close();
          },
        }),
        {
          headers: {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
          },
        }
      );
    }

    // GET /result/:sessionId
    if (url.pathname.startsWith('/result/')) {
      const sessionId = url.pathname.split('/').pop()!;
      const handle = await executor.getHandle(registry.get('default'), sessionId);

      if (!handle) {
        return new Response('Not found', { status: 404 });
      }

      const result = await handle.result();
      return Response.json(result);
    }

    return new Response('Not found', { status: 404 });
  },
};

D1 Considerations

Write Limits

D1 has write throughput limits. For high-volume writes:

// The framework handles message batching internally via appendMessages()
// For custom implementations, batch message inserts like this:
const messages = [msg1, msg2, msg3];
const stmt = env.DB.prepare(
  'INSERT INTO __agents_messages (session_id, sequence, message, created_at) VALUES (?, ?, ?, ?)'
);

await env.DB.batch(
  messages.map((msg, i) => stmt.bind(sessionId, baseSequence + i, JSON.stringify(msg), Date.now()))
);

Read Replicas

D1 supports read replicas for global reads:

// Reads are automatically routed to nearest replica
const state = await stateStore.loadState(sessionId);

// Writes go to primary
await stateStore.saveState(sessionId, state);

SQLite Limitations

D1 is SQLite-based:

• No stored procedures
• Limited concurrent writes
• JSON stored as TEXT (use JSON1 functions)

Durable Object Considerations

Geographic Pinning

Durable Objects are pinned to a region:

// First access pins the DO to nearest region
const stub = env.STREAM_MANAGER.get(id);

// Subsequent access goes to that region
// For global apps, consider DO per-region

Memory Limits

Durable Objects have memory limits:

// For large streams, persist to storage
async write(chunk: StreamChunk) {
  this.state.chunks.push(chunk);

  // Persist every N chunks
  if (this.state.chunks.length % 100 === 0) {
    await this.ctx.storage.put('state', this.state);
  }
}

WebSocket Limits

Limited concurrent WebSocket connections:

// Max ~32 WebSocket connections per DO
// For more, use pub/sub pattern or multiple DOs

Deployment

Create D1 Database

wrangler d1 create agent-state

Run Migrations

wrangler d1 migrations apply agent-state

Deploy

wrangler deploy

Secrets

wrangler secret put OPENAI_API_KEY

DO Runtime Stores

When using the Durable Objects runtime, stores are built into the AgentServer DO:

DOStateStore

Uses the DO's built-in SQLite for state persistence:

import { DOStateStore } from '@helix-agents/runtime-cloudflare';

// Created internally by AgentServer
const stateStore = new DOStateStore({
  sql: this.sql, // PartyServer's sql tagged template
  logger: this.logger,
});

// Same SessionStateStore interface
await stateStore.saveState(sessionId, state, { fenceRunId: runId }); // no checkpoint
const loaded = await stateStore.loadState(sessionId);
await stateStore.appendMessages(sessionId, messages, { streamSequence, runId }); // a commit

Key differences from D1StateStore:

• Uses synchronous this.sql template literals (not async D1 queries)
• Single-session-per-DO architecture (no session_id filtering needed)
• No race conditions (single-threaded DO execution)
• Tables created automatically on first use
• Every commit (rows + checkpoint + latest-checkpoint pointer) runs in one transactionSync; saveState writes no checkpoint
• The DO's own run-starts go through runStartCommitSync inside DurableObjectAgentBase's claim, in the same transactionSync as the claim check (see the DO runtime page). The synchronous helpers throw DOStoreNotInitializedError before the schema is initialised
• readSnapshotViewSync serves GET /snapshot from one checkpoint in one transactionSync

compareAndSetStatus return shape (v7)

The v7 contract returns a discriminated union (replacing the boolean of v6) so callers can branch on conflict without an extra round-trip:

const result = await stateStore.compareAndSetStatus(
  sessionId,
  ['active'], // expected statuses
  'paused', // new status
  { error: 'user paused', expectedVersion: 7 }
);

if (result.ok) {
  console.log(`Promoted; new version=${result.newVersion}`);
} else {
  console.log(`CAS conflict: stored=${result.currentStatus} v${result.currentVersion}`);
}

Both D1StateStore and DOStateStore implement this signature uniformly with the rest of the v7 stores.

DOStateStore Schema

The DO-local SQLite schema is at 20 (DO_SCHEMA_VERSION = 20). Migrations apply lazily on first DO access — no operator action required, even for upgrades from earlier versions. The runner applies every listed version above the stored one.

Version	Adds
V1	Baseline state, messages, checkpoints, sub-session refs (incl. message_count/stream_sequence on checkpoints)
V2	mode column on sub_session_refs (persistent sub-agent disambiguation)
V3	pending_client_tool_calls, root_session_id, client_tool_call_ownership on state (HITL pending map + cross-session ownership)
V4	completed_client_tool_calls TEXT on state (HITL idempotency tombstones)
V5	suspension_context TEXT DEFAULT NULL on state (stateless HITL: packs suspendedAwaitingChildren, suspendedStepId, tracingContext, expiresAt, failureReason)
6	mode on state
7	staging.writes (the StepWrites op-list) replaces patches / merge_changes
8	completion_delivered on sub_session_refs
9	forced_completion on state
10	error_detail on state
11	Commit model: checkpoints.kind / run_id / keep_count; runs.lease_until / start_digest / owner_token / owner_unknown_since; state.current_run_id / latest_truncate_checkpoint_id
12	Run-start bookkeeping: runs.entry_step_count / input_row_count / parent_session_id / parent_tool_call_id; the abandoned_run_starts tombstone table (columns added only when missing)
13	runs.completion_reason (nullable; 'stop_when' / 'max_steps' only), added only when missing
14	stop_requests (one durable stop row per (session_id, run_id); the interrupt_flag table is dropped) and sub_session_refs.parent_run_id (WA MR2)
15	sub_session_refs.location ('local' / 'remote', RM-64)
16	runs.request_key / request_digest and a unique (session_id, request_key) (D-2 request keys)
17	sub_session_refs.consult_tool_call_id / child_run_id / outcome / verify_owed_at (DO companion lifecycle MR A: the consult the ref tracks; behind the ChildConsultStore methods)
18	sub_session_refs.held_prior / outcome_generation / held_generation (DO companion lifecycle MR B: the held prior, behind holdPriorResult / releasePriorResult)
19	sub_session_refs.remote_json (SubSessionRef.remote, a remote child's stream identity, read through a schema: a malformed row is a typed InvalidStoredStateError; S-C, CP-141)
20	state.workspace_ref (SessionState.workspaceRef, RM-19)

-- V3
ALTER TABLE state ADD COLUMN pending_client_tool_calls TEXT;
ALTER TABLE state ADD COLUMN root_session_id TEXT;
ALTER TABLE state ADD COLUMN client_tool_call_ownership TEXT;
-- V4
ALTER TABLE state ADD COLUMN completed_client_tool_calls TEXT;
-- V5
ALTER TABLE state ADD COLUMN suspension_context TEXT DEFAULT NULL;

suspension_context carries the same JSON-encoded payload as the D1 column: suspendedAwaitingChildren, suspendedStepId, tracingContext, expiresAt.

Atomic Writes via transactionSync

DOStateStore.saveStateAndPromoteStaging uses transactionSync(() => { ... }) (the DO equivalent of db.batch) so that messages, state row updates, staging promotion, the step checkpoint and the pointer move become visible together. Every other commit does the same. This satisfies the cross-runtime atomicity invariant C-1 documented in the core SessionStateStore interface.

Lazy Migration

When a DO with an older schema wakes up after an upgrade, the next async store call (DOAB's onStart initialises the store) detects the older schema_version row and runs the pending migrations inline before serving the request. Idle DOs migrate on their next access — there is no SchemaVersionMismatchError and no namespace-rename workaround needed.

DOStreamManager

Streams directly to connected WebSocket/SSE clients:

import { DOStreamManager } from '@helix-agents/runtime-cloudflare';

// Created internally by AgentServer
const streamManager = new DOStreamManager({
  sql: this.sql,
  getConnections: () => this.getConnections(),
  broadcast: (msg) => this.broadcast(msg),
  sseConnectionManager: this.sseConnectionManager,
  logger: this.logger,
});

// Writes to SQLite AND broadcasts to all connected clients
const writer = await streamManager.createWriter(streamId, agentId, agentType);
await writer.write(chunk); // FREE - no subrequest!

Key differences from DurableObjectStreamManager:

• Writes directly to WebSocket/SSE connections (no subrequest cost)
• Also persists to SQLite for resumability
• Built into AgentServer (not separate DO)
• Supports hibernation (connections survive DO sleep/wake)

Stream cleanup methods:

// Clean up chunks beyond a specific step (for crash recovery)
await streamManager.cleanupToStep(streamId, stepCount);

// Reset entire stream (for fresh execution with same ID)
await streamManager.resetStream(streamId);

cleanupToStep() removes chunks with step > stepCount, used during crash recovery. resetStream() clears all chunks, called by execute() for fresh starts.

DOUsageStore

Tracks token/tool/sub-agent usage using DO's SQLite:

import { DOUsageStore } from '@helix-agents/runtime-cloudflare';

// Created internally by AgentServer
const usageStore = new DOUsageStore({
  sql: this.sql,
  sessionId: 'session-123',
  logger: this.logger,
});

// Same UsageStore interface
await usageStore.recordEntry(entry);
const rollup = await usageStore.getRollup(sessionId);
const entries = await usageStore.getEntries(sessionId, { kinds: ['tokens'] });

Key differences from D1UsageStore:

• Uses synchronous this.sql template literals (not async D1 queries)
• Single-session-per-DO architecture (sessionId passed at construction)
• Strict about its session: recordEntry for another session throws a HelixError validation_error and inserts nothing, and getEntries / getRollup read only rows whose sessionId is the store's (a row whose data is not valid JSON is excluded and reported in a warn)
• Created automatically by AgentServer when usageStore is not configured; a configured usageStore factory replaces it entirely (no fallback to it)
• Usage accessible via /usage endpoint on the DO

Automatic usage tracking:

AgentServer automatically creates DOUsageStore and passes it to the JSAgentExecutor. After execution, retrieve usage via the handle:

const handle = await executor.execute(agent, 'Do the task');
await handle.result();

// Get usage rollup
const rollup = await handle.getUsageRollup();
console.log(`Total tokens: ${rollup?.tokens.total}`);

Automatic Schema

DOStateStore creates these tables automatically:

CREATE TABLE state (
  id INTEGER PRIMARY KEY DEFAULT 1,
  session_id TEXT NOT NULL,
  agent_type TEXT NOT NULL,
  stream_id TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active',
  step_count INTEGER DEFAULT 0,
  custom_state TEXT,
  output TEXT,
  error TEXT,
  checkpoint_id TEXT,
  ...
);

CREATE TABLE messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  role TEXT NOT NULL,
  content TEXT,
  tool_calls TEXT,
  ...
);

CREATE TABLE stream_chunks (
  sequence INTEGER PRIMARY KEY,
  chunk TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

Next Steps

• In-Memory Storage - For development
• Redis Storage - For self-hosted production
• Cloudflare Runtime - Overview of both approaches
• DO Runtime - Uses DOStateStore and DOStreamManager
• Workflows Runtime - Uses D1StateStore