@helix-agents/store-cloudflare

Cloudflare storage implementations using D1 and Durable Objects.

Installation

npm install @helix-agents/store-cloudflare

createCloudflareStore

Factory function for creating stores from bindings.

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

const { stateStore, streamManager } = createCloudflareStore({
  db: env.AGENT_DB, // D1Database binding
  streams: env.STREAMS, // DurableObjectNamespace binding
});

Options

interface CreateCloudflareStoreOptions {
  // State store options
  stateOptions?: {
    logger?: Logger;
  };

  // Stream manager options
  streamOptions?: {
    bufferSize?: number;
    flushInterval?: number;
    logger?: Logger;
  };
}

D1StateStore

D1-backed state storage.

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

const stateStore = new D1StateStore({
  database: env.AGENT_DB,
  logger: console,
});

D1StateStoreOptions

interface D1StateStoreOptions {
  database: D1Database; // the D1 binding
  logger?: Logger; // default: silent
  maxStateBytes?: number; // default DEFAULT_D1_MAX_STATE_BYTES (1.8 MB)
}

DEFAULT_D1_MAX_STATE_BYTES (1.8 MB) is exported. maxStateBytes is the store's one size limit (limits = uniformStoreLimits(maxStateBytes)), in UTF-8 bytes of the JSON it binds: one tool's staged writes, the folded customState (and the state row's output and client-tool maps) of a step or state commit, the caller's tags / metadata at createSession, and each message row. Over it the store throws core's StateSizeExceededError (state_size_exceeded, non-retryable) with nothing written. saveStateAndPromoteStaging folds the staged writes in JS and checks the JSON it binds; promoteStaging and mergeCustomState fold inside SQL, so after a check of the fold they read first, their batch also refuses the SQL fold by its byte length (length(CAST(… AS BLOB)), never a character count). A D1 error after a bind over D1's 2 MB value cap (D1_MAX_VALUE_BYTES) is mapped to the same error by the measured bytes, never by its text. The value must be a positive integer or Infinity, else a typed validation_error. On CF Workflows every step result is also capped at 1 MiB, which is lower; see Step results and state size.

Methods

// Save state
await stateStore.saveState(sessionId, state);

// Load state
const state = await stateStore.loadState(sessionId);

// Check existence
const exists = await stateStore.sessionExists(sessionId);

// Update status
await stateStore.updateStatus(sessionId, 'completed');

// Get messages with pagination
const { messages, hasMore } = await stateStore.getMessages(sessionId, {
  offset: 0,
  limit: 50,
});

Checkpoint Methods

// Get a specific checkpoint
const checkpoint = await stateStore.getCheckpoint('cpv1-session-123-s5-...');

// Get most recent checkpoint
const latest = await stateStore.getLatestCheckpoint('session-123');

// List checkpoints with pagination
const result = await stateStore.listCheckpoints('session-123', {
  limit: 10,
  offset: 0,
});
// result.checkpoints, result.total, result.hasMore

D1 listCheckpoints reads total and the page in one db.batch (one implicit transaction), ordered by step_count, timestamp, id, so a checkpoint committed meanwhile can no longer make the page and total disagree (RM-53, RM-54; the finding records are updated once their conformance cells confirm it).

Commits, run start and fences

Breaking: the store implements the checkpointed-commit contract

appendMessages(sessionId, messages, commit) and truncateMessages(sessionId, keepCount, commit) take a required MessageCommit ({ streamSequence, runId, startRun? }); promote takes the commit's streamSequence and runId; commitState, renewRunLease, setRunOwnerUnknownSince and abandonRunStart are new. See the core store interface and the upgrade guide.

Every message-log mutation that carries a MessageCommit is a checkpointed commit: appendMessages (kind: 'append', or 'input' with commit.startRun), truncateMessages ('truncate'), promoteStaging / saveStateAndPromoteStaging ('step'), commitState ('state') and cloneSession ('clone', on the target). Each writes a checkpoint carrying kind, runId and (truncate) keepCount, and moves the session's latest pointer, atomically with the mutation. saveState never writes a checkpoint, createCheckpoint never moves the pointer, and getLatestCheckpoint follows the pointer.

D1 has no interactive transactions, so each commit is one db.batch: the store pre-reads the view (state version, the current-run pointer, the checkpoint pointer, the runs), decides with the core helpers (applyRunStartRules, assertRunIsCurrentLive, ...) and guards every statement on that view. A batch that lost a race is all no-ops; the store re-reads and re-decides, up to MAX_COMMIT_ATTEMPTS (25) times with jittered backoff. Persistent contention throws D1CommitContentionError. A commit by a superseded run throws RunSupersededError and writes nothing.

• Run-start commit (appendMessages with commit.startRun): idempotent by runId + start digest, a compare-and-set on startRun.expectCurrentRunId and the run's status (expectCurrentRunStatus), the owner guard, the session gate (entryStatuses, against the raw session status read once in the same step as the claim guard), supersede every other live run, the optional rewind (startRun.truncateTo deletes the rows past it and writes a truncate checkpoint by the new run), then the run, its input rows and the input checkpoint. A runId given up with abandonRunStart is refused with RunStartRejectedError('entry_abandoned').
• Fences. saveState(sessionId, state, { fenceRunId }) applies only while fenceRunId is the current live run; { fenceIdle: { runId } } only while the current run is still runId (null: none) and not live. updateRunStatus is sticky for terminal statuses (including superseded).
• Leases. renewRunLease(sessionId, runId, ttlMs) sets the run's lease on the Worker clock; setRunOwnerUnknownSince records a CF Workflows owner that reports unknown.
• getMessages throws CorruptMessageRowError for a corrupt row in the requested page (it used to skip it).
• cloneSession with a fromCheckpointId that does not exist, or belongs to another session, throws state_checkpoint_not_found / state_checkpoint_foreign (breaking: it used to clone the whole log), and the clone is one atomic batch.

Every statement in a db.batch counts as a D1 query against the Worker's per-invocation limit (1000 on the paid plan, 50 on the free plan). A commit that exhausts all its attempts costs roughly 150–200 queries, so on the free plan heavy contention on one session can fail with D1's query limit instead of D1CommitContentionError.

Staging Methods

// Stage changes for a tool call within a step
await stateStore.stageChanges('session-123', 'step-1', {
  toolCallId: 'call-456',
  writes: {
    ops: [{ kind: 'append', key: 'notes', items: ['New finding'] }],
    warnings: [],
  },
  timestamp: Date.now(),
});

// Commit staged changes
await stateStore.promoteStaging('session-123', 'step-1');

// Rollback staged changes
await stateStore.discardStaging('session-123', 'step-1');

Distributed Coordination

// Atomic compare-and-set for status
const result = await stateStore.compareAndSetStatus(
  'session-123',
  ['running'], // Expected statuses (array)
  'interrupted' // New status
);

if (result.ok) {
  // result.newVersion is the post-increment version
} else {
  // result.currentStatus / result.currentVersion show actual stored values
}

// Increment resume count
const count = await stateStore.incrementResumeCount('session-123');

The run-start commit is the gate for every runtime entry; compareAndSetStatus is not part of it. Resume and retry winners bump resumeCount inside their run-start commit on the runtimes that count it.

Database Schema

The D1StateStore uses programmatic migrations. See d1-migrations.ts for the full schema. Core tables use session-centric naming (all prefixed with __agents_):

CREATE TABLE __agents_states (
  session_id TEXT PRIMARY KEY,
  agent_type TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active',
  step_count INTEGER NOT NULL DEFAULT 0,
  custom_state TEXT NOT NULL DEFAULT '{}',
  output TEXT,
  error TEXT,
  stream_id TEXT,
  parent_session_id TEXT,
  aborted INTEGER DEFAULT 0,
  abort_reason TEXT,
  created_at INTEGER NOT NULL,
  updated_at INTEGER NOT NULL
);

DurableObjectStreamManager

Stream management using Durable Objects.

Size limit (RM-90). The binding declares limits.maxChunkBytes (option maxChunkBytes, default DEFAULT_MAX_STREAM_CHUNK_BYTES = 1.8 MB, at most STREAM_DO_MAX_VALUE_BYTES = 2 MB, the stream DO's SQL value cap; a larger one makes the constructor throw a typed validation_error) and sends it with every write and end. The StreamServer measures the JSON it binds and answers a chunk over it with a 413 state_size_exceeded (cause: 'stream_chunk') before allocating a sequence. The binding revives that answer as core StateSizeExceededError (non-retryable), so a CF Workflows step is never retried for it. An end is never refused: a final output over the limit is not stored, the stream ends, and /info (metadata.endOmitted) and the end frame (omitted) name it. StreamServer also overrides partyserver's onException to rethrow the driver error, so any other SQL failure reports the real error instead of a 500 "undefined".

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

const streamManager = new DurableObjectStreamManager({
  streamNamespace: env.STREAMS,
  bufferSize: 100, // Chunks to buffer before flush
  flushInterval: 100, // Flush interval (ms)
  logger: console,
});

Methods

// Create writer (implicitly creates stream)
const writer = await streamManager.createWriter(streamId, agentId, agentType);
await writer.write(chunk);
await writer.close();

// Create reader
const reader = await streamManager.createReader(streamId);
for await (const chunk of reader) {
  // Process chunk
}

// End stream with the run's terminal status (required, CP-87)
await streamManager.endStream(streamId, undefined, { status: 'completed' });
await streamManager.endStream(streamId, output, { status: 'completed' }); // With final output
await streamManager.endStream(streamId, undefined, { status: 'interrupted' }); // A stopped run

// Fail stream
await streamManager.failStream(streamId, 'Error message');

// Claim the stream for a run that won its run-start commit (compare-and-set on the owner)
const claimed = await streamManager.claimStream(streamId, runId, { owner: previousOwner });

// Run-fenced transitions: apply only while `runId` still owns the stream (else a no-op)
await streamManager.endStream(streamId, output, { status: 'completed' }, { expectRunId: runId });
await streamManager.failStream(streamId, 'Error message', errorDetail, { expectRunId: runId });
await streamManager.pauseStream(streamId, { expectRunId: runId });

// Get stream info: null ONLY when the stream does not exist (a claimed stream exists); a failed stream
// carries info.metadata.error and info.metadata.errorDetail. A non-2xx answer from the stream
// Durable Object (/info) throws StreamConnectionError; it is never read as "no such stream".
const info = await streamManager.getStreamInfo(streamId);

// initStream resolves true when it created the stream and false ONLY when it already existed;
// a non-2xx answer (/init) throws StreamConnectionError
const created = await streamManager.initStream(streamId);

// Resumable reader
const reader = await streamManager.createResumableReader(streamId, {
  fromSequence: 100,
  waitForLive: true, // a terminal / missing stream means "not live yet" until the first chunk past the cursor
});
// reader.catchUpHead: the head when the reader attached; reader.caughtUp: the backlog boundary was passed

claimStream(streamId, runId, expect?) resolves true when runId owns the stream afterwards, false otherwise (nothing written; a missing stream is never created). Runtimes call it through core's claimStreamForCommittedRun. A chunk of the session's own stream stamped with a run other than the owner is dropped once a newer run has claimed it; a child's chunk in its parent's stream and run_terminated are not fenced.

Error text is stored sanitized (core). The stream Durable Object itself sanitizes each chunk on /write (core sanitizeStreamChunk) and the message and errorDetail on /fail, so getStreamInfo().metadata.error / errorDetail and every reader's frames carry no raw text, whatever the caller sent. The raw text stays in the session row and the log.

Reader buffering and drops

A reader never drops a chunk. Its buffer of unread chunks is bounded by count (maxBufferSize, default 10000) and by bytes (core READER_BUFFER_HIGH_WATER_BYTES, 8 MiB): once either bound is reached it stops reading its /sse body until the consumer drains half of it, so a slow consumer stalls the stream instead of growing the reader's memory. The onBufferOverflow option and the BufferOverflowEvent type are removed: a reader used to drop every chunk past maxBufferSize and only report it there. The backlog a reader reads before it returns (from its cursor to the route's caught-up frame) is capped at core READER_BACKLOG_MAX_BYTES (32 MiB); past it the reader ends with StreamTruncatedError (re-anchor from the snapshot).

When it ends (its terminal, a throw, or the consumer's close()), a reader logs one info line, SSE reader closed, with { streamId, maxBufferedBytes, reopens } (CP-136). maxBufferedBytes is the most unread bytes it ever held, which is its own read-ahead of the consumer. Anything the transport buffers after it (a response pipe, socket buffers, a client's queue) is not counted, so a byte count taken outside the reader cannot measure the bound. The Durable Object agent's reader (DOStreamManagerClient) logs the same fields as DO /sse reader closed. Core ReaderBufferBudget.highWaterBytes and StreamReopenBudget.reopenCount expose the two values.

On a drop (a body that ends with no terminal frame, or a body read that fails) a reader re-opens /sse itself from the highest stored sequence it took in, and throws the retryable StreamDropError only when a re-open cannot help (see resumable streams). WE C2b: StreamServer ends an evicted connection with evicted{resumeFrom, reason} (its unread-byte budget, staleness, a server error), and the reader acts on it as the DO readers do: a re-open by the reason (slow_consumer / stale off the cap), a typed StreamEvictedError / StreamRunChangedError when a re-open cannot help, StreamTruncatedError for a cursor below resumeFrom.

StreamServer

PartyServer-based Durable Object class for stream coordination. Must be exported from your worker.

Its /claim endpoint is the compare-and-set behind claimStream (expectOwner). During a rolling deploy, a new client talking to an old StreamServer gets the old, unconditional claim. Deploy the worker that hosts the stream Durable Objects before, or together with, the workers that run agents.

A request that throws answers 500 { error } with the message sanitized (core sanitizeErrorMessage); the raw text is logged. LockServer's 500 error is sanitized the same way.

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

// In worker.ts - re-export the class
export { StreamServer };

Custom Logger

Create a custom class with a logger:

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

const CustomStreamServer = StreamServer.withLogger(myLogger);

export { CustomStreamServer as StreamServer };

Migrations

D1 schema migrations.

import {
  runMigration,
  isMigrated,
  dropAllTables,
  getAgentsMigrationVersion,
  getAgentsTableNames,
  SCHEMA_MIGRATION_V1,
  CURRENT_SCHEMA_VERSION,
  TABLE_NAMES,
} from '@helix-agents/store-cloudflare';

// Run migration
await runMigration(env.AGENT_DB);

// Check if migrated
const migrated = await isMigrated(env.AGENT_DB);

// Get current version
const version = await getAgentsMigrationVersion(env.AGENT_DB);

// Drop all tables (for testing)
await dropAllTables(env.AGENT_DB);

• runMigration(db) runs on core's claim-first SQL migration runner. It is safe to call on every boot and from several isolates at once: each migration is one batch whose first statement records its version row, so a concurrent runner that loses the claim applies nothing (no duplicate column name, no data statement run twice). A migration that fails for any other reason rethrows its original error. A fresh database applies the collapsed schema and records every version in one batch.
• getAgentsMigrationVersion(db) returns the highest recorded version, or 0 only when the version table does not exist yet. A failed read throws; it is no longer reported as version 0.
• dropAllTables(db) also drops __agents_abandoned_run_starts.

Migration V17 (all nullable): __agents_checkpoints.kind / run_id / keep_count; __agents_runs.lease_until / start_digest / owner_token / owner_unknown_since; __agents_states.current_run_id / latest_truncate_checkpoint_id. Migration V18: __agents_runs.entry_step_count / input_row_count / parent_session_id / parent_tool_call_id and the abandoned_run_starts tombstone table. runMigration applies them. Migration V19 adds the nullable __agents_runs.completion_reason (RunMetadata.completionReason, written only for a stopWhen stop or a maxSteps cut-off); updateRunStatus writes it and getRun / getCurrentRun / listRuns read it back. Migration V20 (WA MR2) creates __agents_stop_requests (one row per (session_id, run_id), the run-scoped stop record behind requestStop / readStop) and drops __agents_interrupt_flags. Migration V22 (D-2 request keys) adds __agents_runs.request_key / request_digest and a unique index on (session_id, request_key); getRunByRequestKey reads it, and the run-start commit's claim guard refuses a held key (RunStartRejectedError('request_key_taken')).

Migration V21 adds a partial index on __agents_states(parent_session_id) and __agents_sub_session_refs.location. deleteSession cascades to the session's descendants and deletes their __agents_usage rows (see Deleting a session).

Migration V23 (DO companion lifecycle MR B) adds the held prior to __agents_sub_session_refs: nullable held_prior TEXT (JSON HeldPriorResult, read with core readStoredHeldPrior), outcome_generation INTEGER and held_generation INTEGER, behind holdPriorResult / releasePriorResult (the release matches the holder through json_extract only when json_valid(held_prior), so a corrupt slot is stale, never a statement error).

Migration V24 (RM-19) adds nullable workspace_ref TEXT to __agents_states: SessionState.workspaceRef as JSON, read with core readStoredWorkspaceRef, so a woken workspace run reopens its own workspace. A clone never inherits it.

getMessages with no limit reads the whole log from offset in 500-row pages, as one consistent read (core readMessagesToEnd, RM-01).

Migration SQL

// Get the SQL for manual migration
console.log(SCHEMA_MIGRATION_V1);

Errors

import {
  D1StateError, // Base D1 error
  StateNotFoundError, // State not found
  SubSessionRefNotFoundError,
  StreamConnectionError, // DO connection error
  StreamNotFoundError, // Stream not found
  SequenceConflictError, // Sequence mismatch
  D1CommitContentionError, // A guarded commit lost its CAS on every attempt
} from '@helix-agents/store-cloudflare';

try {
  const state = await stateStore.loadState(sessionId);
} catch (error) {
  if (error instanceof StateNotFoundError) {
    // Handle not found
  }
}

D1CommitContentionError extends StaleStateError and keeps its name, so retry paths gated on StaleStateError treat it as retryable. It carries attempts and divergence (what diverged on the last lost attempt: the state version, the current-run pointer, the checkpoint pointer, the fenced run's liveness or the run-start view).

Complete Example

wrangler.toml

name = "my-agent-worker"
main = "src/worker.ts"
compatibility_date = "2024-12-01"
compatibility_flags = ["nodejs_compat"]

[[d1_databases]]
binding = "AGENT_DB"
database_name = "my-agents-db"
database_id = "xxx-xxx-xxx"

[[durable_objects.bindings]]
name = "STREAMS"
class_name = "StreamServer"

[[migrations]]
tag = "v1"
new_sqlite_classes = ["StreamServer"]

worker.ts

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

// Re-export Durable Object
export { StreamServer };

interface Env {
  AGENT_DB: D1Database;
  STREAMS: DurableObjectNamespace;
}

export default {
  async fetch(request: Request, env: Env) {
    // Run migration on first request (or use scheduled task)
    await runMigration(env.AGENT_DB);

    // Create stores
    const { stateStore, streamManager } = createCloudflareStore({
      db: env.AGENT_DB,
      streams: env.STREAMS,
    });

    // Use stores...
    const state = await stateStore.loadState('session-123');

    return Response.json({ state });
  },
};

Migration Script

# Create database
npx wrangler d1 create my-agents-db

# Run migration (creates tables)
npx wrangler d1 execute my-agents-db --file=./migrations/0001_init.sql

D1UsageStore

D1-backed usage tracking storage. Track LLM tokens, tool executions, sub-agent calls, and custom metrics in Cloudflare Workers.

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

const usageStore = new D1UsageStore({
  database: env.AGENT_DB,
  tableName: 'usage_entries', // Optional, default: 'usage_entries'
});

Basic Usage

Pass to executor to enable usage tracking:

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

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

Methods

recordEntry

Record a usage entry (called internally by the framework).

await usageStore.recordEntry({
  kind: 'tokens',
  sessionId: 'session-123',
  stepCount: 1,
  timestamp: Date.now(),
  source: { type: 'agent', name: 'my-agent' },
  model: 'gpt-6-luna',
  tokens: { prompt: 100, completion: 50, total: 150 },
});

getEntries

Get usage entries for a run with optional filtering.

// All entries
const entries = await usageStore.getEntries('session-123');

// Filter by kind
const tokenEntries = await usageStore.getEntries('session-123', {
  kinds: ['tokens'],
});

// Filter by step range
const midRunEntries = await usageStore.getEntries('session-123', {
  stepRange: { min: 5, max: 10 },
});

// Pagination
const page = await usageStore.getEntries('session-123', {
  limit: 10,
  offset: 20,
});

getRollup

Get aggregated usage rollup.

// This agent's usage only
const rollup = await usageStore.getRollup('session-123');

// Include sub-agent usage (lazy aggregation)
const totalRollup = await usageStore.getRollup('session-123', {
  includeSubAgents: true,
});

exists

Check if usage data exists by inspecting getRollup() — it returns null when no entries exist for the session.

const rollup = await usageStore.getRollup('session-123');
const hasUsage = rollup !== null;

delete

Delete usage data for a run.

await usageStore.delete('session-123');

getEntryCount

Get entry count without fetching entries.

const count = await usageStore.getEntryCount('session-123');

findSessionIds

Find all tracked session IDs with optional filters.

// All session IDs
const sessionIds = await usageStore.findSessionIds();

// With filters
const filteredSessionIds = await usageStore.findSessionIds({
  agentType: 'researcher',
  limit: 100,
});

deleteOldEntries

Delete entries older than a specified age.

// Delete entries older than 7 days
await usageStore.deleteOldEntries(7 * 24 * 60 * 60 * 1000);

Database Schema

The D1 migration creates this table:

CREATE TABLE __agents_usage (
  id TEXT PRIMARY KEY,
  session_id TEXT NOT NULL,
  kind TEXT NOT NULL,
  step_count INTEGER NOT NULL,
  timestamp INTEGER NOT NULL,
  source_type TEXT NOT NULL,
  source_name TEXT NOT NULL,
  data TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

CREATE INDEX idx___agents_usage_session_id ON __agents_usage(session_id);
CREATE INDEX idx___agents_usage_kind ON __agents_usage(kind);
CREATE INDEX idx___agents_usage_timestamp ON __agents_usage(timestamp);

Migration

Run migration to create the usage table:

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

// Run on worker startup or via scheduled task
await runMigration(env.AGENT_DB);

Complete Example

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

interface Env {
  AGENT_DB: D1Database;
  STREAMS: DurableObjectNamespace;
}

export default {
  async fetch(request: Request, env: Env) {
    await runMigration(env.AGENT_DB);

    const { stateStore, streamManager } = createCloudflareStore({
      db: env.AGENT_DB,
      streams: env.STREAMS,
    });

    const usageStore = new D1UsageStore({ database: env.AGENT_DB });

    // Execute agent with usage tracking
    const handle = await executor.execute(agent, 'Task', { usageStore });
    await handle.result();

    // Get usage data
    const rollup = await handle.getUsageRollup();

    return Response.json({
      tokens: rollup?.tokens.total,
      toolCalls: rollup?.toolStats.totalCalls,
    });
  },

  // Scheduled cleanup
  async scheduled(event: ScheduledEvent, env: Env) {
    const usageStore = new D1UsageStore({ database: env.AGENT_DB });
    // Delete entries older than 30 days
    await usageStore.deleteOldEntries(30 * 24 * 60 * 60 * 1000);
  },
};

DurableObjectLockManager

Distributed lock manager using Durable Objects. Prevents concurrent execution of the same agent across Workers.

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

const lockManager = new DurableObjectLockManager({
  lockBinding: env.LOCKS,
});

wrangler.toml Configuration

[[durable_objects.bindings]]
name = "LOCKS"
class_name = "LockServer"

[[migrations]]
tag = "v2"
new_sqlite_classes = ["LockServer"]

Export the Durable Object

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

export { LockServer };

Methods

acquire

Acquire a lock with fencing token.

const lock = await lockManager.acquire('session-123', {
  ttlMs: 30000,
  owner: 'worker-instance-1',
});

if (lock) {
  console.log('Lock acquired');
  console.log('Fencing token:', lock.fencingToken);

  // Do work...

  // Refresh to extend TTL
  await lock.refresh();

  // Release when done
  await lock.release();
}

isLocked

Check if a lock is currently held.

const locked = await lockManager.isLocked('session-123');

Usage with Executor

import { CloudflareAgentExecutor } from '@helix-agents/runtime-cloudflare';
import { DurableObjectLockManager } from '@helix-agents/store-cloudflare';

const lockManager = new DurableObjectLockManager({ lockBinding: env.LOCKS });

const executor = new CloudflareAgentExecutor({
  workflowBinding: env.AGENT_WORKFLOW,
  stateStore,
  streamManager,
  lockManager, // Prevents concurrent execution
});

Fencing Tokens

Fencing tokens prevent split-brain in distributed scenarios:

const lock = await lockManager.acquire('session-123');

// Operations validate the fencing token
await stateStore.saveState(state, { fencingToken: lock.fencingToken });

Re-exported Types

For convenience, core types are re-exported:

import type {
  StateStore,
  StreamManager,
  StreamWriter,
  StreamReader,
  StreamChunk,
  ResumableStreamReader,
  ResumableReaderOptions,
  StreamInfo,
  ResumableStreamStatus,
  AgentState,
  Message,
  AgentStatus,
  StepWrites,
  LockManager,
  Lock,
  LockAcquireOptions,
} from '@helix-agents/store-cloudflare';

See Also

• Cloudflare Store Guide
• Cloudflare Runtime
• Checkpoints
• Distributed Coordination
• Research Assistant Cloudflare Example