Storage Overview

Storage in Helix Agents consists of two components: State Stores for persisting agent state, and Stream Managers for real-time event streaming. Both are defined as interfaces, enabling different implementations for different environments.

Why Separate Components?

State and streaming have different requirements:

Concern	State Store	Stream Manager
Purpose	Persist agent state	Real-time event delivery
Access Pattern	Read-modify-write	Append-only, subscribe
Consistency	Strong (atomic updates)	Eventual (ordering matters)
Retention	Until explicit delete	Temporary (TTL-based)

By separating them, you can:

• Use in-memory streams with Redis state
• Scale streaming infrastructure independently
• Apply different retention policies

Session-Centric Model

All state store operations use sessionId as the primary key. This is a fundamental design principle:

• Messages are stored per session, not duplicated per run
• Runs are lightweight execution metadata within a session
• Checkpoints reference message boundaries by count, not by copying messages

This design enables efficient storage (O(n) messages vs O(n²) for run-centric models) and natural conversation continuity. See Concepts: Session vs Run for details.

Available Implementations

Package	State Store	Stream Manager	Use Case
@helix-agents/store-memory	InMemoryStateStore	InMemoryStreamManager	Development, testing
@helix-agents/store-redis	RedisStateStore	RedisStreamManager	Production (self-hosted)
@helix-agents/store-postgres	PostgresStateStore	(any)	Production (Postgres-backed)
@helix-agents/store-cloudflare	D1StateStore	DOStreamManager	Cloudflare Workers (Workflows path)
@helix-agents/runtime-cloudflare (DO stores)	DOStateStore	DOStreamManager	Cloudflare DO runtime

State Store Interface

The SessionStateStore interface defines how agent state is persisted. sessionId is the primary key for all operations:

interface SessionStateStore {
  // Session lifecycle
  createSession(sessionId: string, options?: CreateSessionOptions): Promise<void>;
  sessionExists(sessionId: string): Promise<boolean>;
  deleteSession(sessionId: string): Promise<void>;

  // State operations
  loadState(sessionId: string): Promise<SessionState | null>;
  // Never a commit: writes no checkpoint, never moves the latest-checkpoint pointer.
  saveState(
    sessionId: string,
    state: SessionState,
    opts?: { fenceRunId?: string; fenceIdle?: { runId: string | null } }
  ): Promise<void>;
  // A commit (kind 'state'): the terminal-state commit
  commitState<TState, TOutput>(
    sessionId: string,
    state: SessionState<TState, TOutput>,
    commit: MessageCommit,
    opts?: { expectedVersion?: number }
  ): Promise<{ checkpointId: string; newVersion: number }>;

  // A commit (kind 'step'): appendMessages + state + promoteStaging + checkpoint in one shot
  saveStateAndPromoteStaging<TState, TOutput>(
    sessionId: string,
    state: SessionState<TState, TOutput>,
    appendMessages: Message[],
    checkpointMeta: { stepId: string; stepCount: number; streamSequence: number; runId: string },
    options?: { expectedVersion?: number }
  ): Promise<{ checkpointId: string; newVersion: number }>;

  // Commits (kind 'input' with commit.startRun, else 'append' / 'truncate')
  appendMessages(sessionId: string, messages: Message[], commit: MessageCommit): Promise<void>;
  truncateMessages(sessionId: string, keepCount: number, commit: MessageCommit): Promise<number>;

  // Atomic operations (for parallel tool execution)
  mergeCustomState(sessionId: string, writes: StepWrites): Promise<{ warnings: string[] }>;
  updateStatus(sessionId: string, status: SessionStatus): Promise<void>;
  incrementStepCount(sessionId: string): Promise<number>;

  // v7: discriminated-union return for richer conflict handling
  compareAndSetStatus(
    sessionId: string,
    expectedStatuses: SessionStatus[],
    newStatus: SessionStatus,
    options?: { interruptContext?: InterruptContext; error?: string; expectedVersion?: number }
  ): Promise<
    | { ok: true; newVersion: number }
    | { ok: false; currentStatus: SessionStatus; currentVersion: number }
  >;

  // Run-scoped stop records (WA MR2; all stores)
  requestStop(sessionId: string, req: RequestStopInput): Promise<RequestStopResult>;
  readStop(sessionId: string, runId: string): Promise<StopRequest | null>;

  // Sub-agent tracking
  addSubSessionRefs(sessionId: string, refs: SubSessionRef[]): Promise<void>;
  updateSubSessionRef(sessionId: string, update: SubSessionRefUpdate): Promise<void>;

  // Message queries (for UI/recovery)
  getMessages(sessionId: string, options?: GetMessagesOptions): Promise<PaginatedMessages>;
  getMessageCount(sessionId: string): Promise<number>; // must be a LIVE row count

  // Cross-session listing (admin / cleanup)
  listSessions(options?: ListSessionsOptions): Promise<PaginatedSessions>;
}

Commits and checkpoints

Every write that changes the message log or what a snapshot shows is a commit: one atomic store operation that also writes a checkpoint { id, kind, runId, messageCount, streamSequence, stepCount, state, keepCount } and moves the session's latest-checkpoint pointer to it. kind (core CommitKind) says which write it was:

kind	Written by
input	appendMessages with commit.startRun: a run-start commit (rows may be empty)
append	appendMessages without startRun
step	saveStateAndPromoteStaging / promoteStaging
state	commitState (the terminal-state commit, on completion, failure and stop)
truncate	truncateMessages, or a run-start commit's truncateTo rewind
clone	cloneSession

• MessageCommit = { streamSequence, runId, startRun? } is a required argument of appendMessages / truncateMessages (breaking for callers and custom stores). streamSequence is the live stream head at the commit (core resolveCommitSequence; NO_STREAM_SEQUENCE = -1 when no stream manager is wired; a failed head read throws state_checkpoint_sequence_unavailable, never a silent 0).
• Every commit that carries a runId is fenced: it writes nothing and throws RunSupersededError (state_run_superseded) unless runId is the session's current live run.
• saveState is not a commit: it writes no checkpoint and never moves the pointer (RM-37). Runtimes pass { fenceRunId } (or { fenceIdle } for a write made when no run is live). createCheckpoint writes a checkpoint row with kind: null and never moves the pointer.
• getLatestCheckpoint returns the checkpoint the pointer names (the most recent commit), never the highest-stepCount row.
• The run-start commit. appendMessages(sessionId, rows, { streamSequence, runId, startRun }) applies, atomically and on the store's own clock: idempotency by runId (a different input throws RunStartRejectedError('idempotency_mismatch')), a tombstone from abandonRunStart (RunStartRejectedError('entry_abandoned')), a compare-and-set on startRun.expectCurrentRunId (RunStartConflictError when the current run is live, else RunStartRejectedError('expected_mismatch_idle')) and on the current run's status startRun.expectCurrentRunStatus (expected_run_status_changed), the owner guard (same ownerToken, a lapsed JS lease, or a runtime-asserted ownerless run, else RunStartConflictError('run_active'); skipped for a current run named by startRun.finishCurrent whose K5 the store's own reads verify, CP-84), the session gate (the session status read in the same atomic step must be in startRun.entryStatuses, else entry_status_changed), supersedes every other live run, applies startRun.truncateTo, applies startRun.resumedCompanionRef (a resumed companion's parent ref turns running in the same step; a missing ref throws RunStartRejectedError('resumed_companion_ref_missing'); a store that cannot do this atomically must refuse the field typed), then creates the run, the rows and the input checkpoint.
• Run methods. renewRunLease (the JS lease, a CAS on the store clock), setRunOwnerUnknownSince (CF Workflows) and abandonRunStart (the run-start fail-safes) are new. updateRunStatus is sticky: a terminal status (completed, failed, interrupted, superseded) never changes again.

The contract suites in @helix-agents/core/testing (commit-checkpoint-operations, run-start-operations, fence-operations, checkpoint-operations) pin these rules for custom stores. See the upgrade guide.

Message counts must be truthful (custom stores)

Runtimes compare every history read with getMessageCount (core loadLLMHistory); a mismatch fails the run with state_history_incomplete. So getMessageCount must count live rows, never read a denormalized column. If your store keeps a denormalized count (for listSessions), truncateMessages must update it atomically with the delete, and every checkpoint must record the live count as its messageCount. The contract suite in @helix-agents/core/testing (stateStoreContractTests) checks both.

Operator: expiredSessionCleanup

@helix-agents/agent-server exports expiredSessionCleanup(stateStore, options), an operator-driven sweep that uses SessionState.expiresAt to garbage-collect abandoned sessions (e.g., HITL waits where the user never returned). Run on a schedule (cron, Cloudflare cron trigger) to reclaim storage. All five v7 stores (memory, Redis, postgres, D1, DO) honor options.error so reaped sessions are recorded with a consistent failure reason.

Atomic Operations

The atomic operations enable safe concurrent modifications from parallel tool execution:

// Multiple tools can safely append messages
await Promise.all([
  stateStore.appendMessages(sessionId, [toolResult1]),
  stateStore.appendMessages(sessionId, [toolResult2]),
]);

// Multiple tools can safely update different state keys
await Promise.all([
  stateStore.mergeCustomState(sessionId, {
    ops: [{ kind: 'replace', key: 'key1', value: 'value1' }],
    warnings: [],
  }),
  stateStore.mergeCustomState(sessionId, {
    ops: [{ kind: 'replace', key: 'key2', value: 'value2' }],
    warnings: [],
  }),
]);

State Merge Semantics

The mergeCustomState operation applies a StepWrites op-list — the canonical write representation produced by ImmerStateTracker.getStepWrites():

// Array delta mode: new items are appended using 'append' ops
// Tool 1: draft.items.push('a') → op { kind: 'append', key: 'items', items: ['a'] }
// Tool 2: draft.items.push('b') → op { kind: 'append', key: 'items', items: ['b'] }
// Result: items = [...original, 'a', 'b']

// Scalar values: last write wins ('replace' op)
// Tool 1: draft.count = 5  → op { kind: 'replace', key: 'count', value: 5 }
// Tool 2: draft.count = 10 → op { kind: 'replace', key: 'count', value: 10 }
// Result: count = 5 or 10 (depends on timing)

// Explicit delete: 'delete' op (not null assignment)
// { kind: 'delete', key: 'temp' }
// Result: temp key is removed

Run Tracking

The state store also tracks execution runs within sessions. A run represents a single execution (via execute() or resume()), while a session represents the overall conversation:

interface SessionStateStore {
  // ... other methods ...

  // Run tracking
  createRun(sessionId: string, runId: string, metadata: RunMetadata): Promise<void>;
  updateRunStatus(runId: string, status: RunStatus, updates?: RunStatusUpdate): Promise<void>;
  getCurrentRun(sessionId: string): Promise<RunRecord | null>;
  listRuns(sessionId: string): Promise<RunRecord[]>;
}

When runs are created: every entry (execute(), resume() in any mode, retry(), a client-tool continuation, a sub-agent or companion turn) creates its run inside its run-start commit, with the next turn number. Nothing creates a run outside it on any runtime.

Run status values:

• running - Execution in progress
• suspended_client_tool / suspended_awaiting_children / suspended_step_partial - Waiting at a HITL boundary
• completed - Finished successfully with output
• failed - Terminated with error
• interrupted - User-requested pause (can be resumed)
• superseded - Another run took the session over (never a failure)

Every runtime writes the run's terminal status after its terminal-state commit (Temporal used to leave it running, RM-47; the record stays open until its conformance cell lands).

Example: Querying run history:

// Get the current/most recent run
const currentRun = await stateStore.getCurrentRun(sessionId);
console.log(currentRun?.status); // 'running' | 'completed' | 'failed' | 'interrupted'

// List all runs for a session (useful for debugging/auditing)
const runs = await stateStore.listRuns(sessionId);
for (const run of runs) {
  console.log(`Turn ${run.turn}: ${run.status} (${run.stepCount} steps)`);
}

Stream Manager Interface

The StreamManager interface defines how events are streamed:

interface StreamManager {
  // Writer lifecycle
  createWriter(streamId: string, runId: string, agentType: string): Promise<StreamWriter>;

  // Reader lifecycle
  createReader(streamId: string): Promise<StreamReader | null>;

  // Stream control (fence: apply only while expectRunId still owns the stream)
  endStream(
    streamId: string,
    finalOutput?: unknown,
    terminal?: {
      status: 'completed' | 'interrupted';
      state?: Record<string, unknown>;
      usage?: UsageRollup;
    },
    fence?: { expectRunId: string }
  ): Promise<void>;
  failStream(
    streamId: string,
    error: string,
    errorDetail?: ErrorDetail,
    fence?: { expectRunId: string }
  ): Promise<void>;
  pauseStream(streamId: string, fence?: { expectRunId: string }): Promise<void>;
  resumeStream(streamId: string): Promise<void>;

  // Optional: make runId the stream's owner run, a compare-and-set on the owner read
  claimStream?(
    streamId: string,
    runId: string,
    expect?: { owner: string | null }
  ): Promise<boolean>;

  // Resumable streams
  createResumableReader(
    streamId: string,
    options?: ResumableReaderOptions // fromSequence, waitForLive, ...
  ): Promise<ResumableStreamReader | null>;
  getStreamInfo(streamId: string): Promise<StreamInfo | null>;

  // Content reconstruction
  getAllChunks(streamId: string): Promise<StreamChunk[]>;
  getChunksFromStep(streamId: string, fromStep: number): Promise<StreamChunk[]>;
  getChunksFromSequence(streamId: string, fromSequence: number): Promise<StreamChunk[]>;
}

Stream ownership and the run fence

A session's stream is shared by all of its runs, so a late call from a finished or superseded run must not act on a newer run's stream:

• claimStream (CP-85) makes a run the stream's owner. Runtimes call it only through core claimStreamForCommittedRun, right after the run won its run-start commit; it is a compare-and-set on the owner the caller read (expect.owner), so it never overwrites a newer run's claim. A sub-agent writing into its parent's stream never claims it. A manager whose claimStream resolves undefined is treated as having claimed (the old unconditional behaviour).
• Fenced transitions. endStream / failStream / pauseStream with a fence apply only while fence.expectRunId still owns the stream; otherwise they are a no-op. Every runtime passes its own run id.
• Fenced chunks. A chunk of the stream's own session stamped with a runId is dropped once a newer run owns the stream (core chunkFenceRunId). A child's chunk in its parent's stream, a chunk with no runId and run_terminated are not fenced.
• waitForLive (ResumableReaderOptions): until the reader yields its first chunk past fromSequence, a terminal status, or a stream that does not exist yet, means "not live yet": the reader parks instead of ending, so a resume in the window right after a run-start commit sees the new run. A manager that ignores it keeps the old behaviour.
• catchUpHead / caughtUp. Every resumable reader exposes the backlog boundary its initial read covered (catchUpHead) and a promise that resolves once that backlog was yielded (caughtUp), so a consumer can tell replayed content from live content without a quiet period.

The shared contract cases (streamManagerContractTests, including WL.* and RF.claim.cas) are in @helix-agents/core/testing.

Resumable Stream Features

Different implementations support different resumability features:

Feature	InMemory	Redis	Cloudflare DO
createResumableReader	Yes	Yes	Yes
getStreamInfo	Yes	Yes	Yes
getAllChunks	Yes	Yes	Yes
getChunksFromStep	Yes	Yes	Yes
pauseStream / resumeStream	Yes	Yes	Yes
claimStream (CAS) and fences	Yes	Yes	Yes
waitForLive	Yes	Yes	Yes
Sequence tracking	Yes	Yes	Yes
Multi-process support	No	Yes	Yes
maxChunks LTRIM cap	No	Yes	n/a (DO SQLite)

Key methods for resumability:

• createResumableReader: Creates a reader that tracks position with sequence numbers. Use fromSequence option to resume from a specific position.
• getStreamInfo: Returns stream status, total chunks, and latest sequence number. Returns null when the stream does not exist (every manager except the in-process DOStreamManager, RM-71; contract case G5.info-null-on-not-found). Use to check if a stream is still active before connecting. A failed stream's metadata carries failStream's message (error) and structured errorDetail (every built-in manager; contract case G5.fail-detail-on-info).
• getAllChunks: Returns all chunks in the stream. Used for full content reconstruction on page refresh.
• getChunksFromStep: Returns chunks from a specific step onwards. More efficient than getAllChunks when checkpoint data is available.

For comprehensive resumability documentation, see Resumable Streams Guide.

Writer/Reader Pattern

Writers emit chunks, readers consume them:

// Create writer (usually done by runtime)
const writer = await streamManager.createWriter(streamId, agentId, agentType);
await writer.write({ type: 'text_delta', delta: 'Hello', ... });
await writer.close();  // Close writer, NOT the stream

// Create reader (usually done by client/UI)
const reader = await streamManager.createReader(streamId);
if (reader) {
  try {
    for await (const chunk of reader) {
      console.log(chunk);
    }
  } finally {
    await reader.close();  // Always close reader
  }
}

Stream Lifecycle

graph TB
    Created["Stream Created (first writer)"]

    Created --> Writers["Writers emit chunks"]
    Writers --> Readers["Readers receive in real-time"]

    Created --> End["endStream() called"]
    End --> Complete["Readers complete normally"]
    Complete --> Historical["New readers get historical chunks"]

    Created --> Fail["OR failStream() called"]
    Fail --> Error["Readers receive error"]
    Error --> Null["New readers get null"]

Choosing Storage

Development

Use in-memory stores for fast iteration:

import { InMemoryStateStore, InMemoryStreamManager } from '@helix-agents/store-memory';

const stateStore = new InMemoryStateStore();
const streamManager = new InMemoryStreamManager();

Pros: Zero setup, fast, easy debugging Cons: Data lost on restart, single process only

Production (Self-Hosted)

Use Redis for durability and multi-process:

import { RedisStateStore, RedisStreamManager } from '@helix-agents/store-redis';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);
const stateStore = new RedisStateStore(redis);
const streamManager = new RedisStreamManager(redis);

Pros: Persistent, multi-process, production-ready Cons: Requires Redis infrastructure

Production (Cloudflare)

Use D1 and Durable Objects for edge deployment:

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

const stateStore = new D1StateStore(env.DB);
const streamManager = new DOStreamManager(env.STREAM_MANAGER);

Pros: Global edge, serverless, integrated with Cloudflare Cons: Cloudflare-specific, different API from other stores

Swapping Stores

Because stores implement standard interfaces, swapping is straightforward:

// Environment-based selection
const createStores = () => {
  if (process.env.NODE_ENV === 'production') {
    const redis = new Redis(process.env.REDIS_URL);
    return {
      stateStore: new RedisStateStore(redis),
      streamManager: new RedisStreamManager(redis),
    };
  }

  return {
    stateStore: new InMemoryStateStore(),
    streamManager: new InMemoryStreamManager(),
  };
};

const { stateStore, streamManager } = createStores();
const executor = new JSAgentExecutor(stateStore, streamManager, llmAdapter);

State Data Model

Session state includes:

interface SessionState<TState, TOutput> {
  // Identity
  sessionId: string; // Primary key for all operations
  agentType: string; // Agent type name
  streamId?: string; // Stream identifier (defaults to sessionId)

  // Hierarchy
  parentSessionId?: string; // Parent if this is a sub-agent
  rootSessionId?: string; // Root pointer for cross-session ownership

  // Execution state
  customState: TState; // Your custom state schema
  stepCount: number; // LLM call count
  status: SessionStatus; // 'active' | 'completed' | 'failed' | 'interrupted' | 'paused' | 'suspended_*'

  // Output / failure
  output?: TOutput; // Final structured output
  error?: string;
  failureReason?: string; // v7: cascade discriminator (e.g., 'parent_suspended')

  // v7: Stateless suspension fields
  suspendedAwaitingChildren?: Record<string, SuspendedChildWait>;
  suspendedStepId?: string;
  pendingClientToolCalls?: Record<string, PendingClientToolCall>;
  completedClientToolCalls?: Record<string, number>; // root-only tombstones
  clientToolCallOwnership?: ClientToolCallOwnership; // root-only
  interruptContext?: InterruptContext;
  tracingContext?: { traceId: string; rootSpanId: string };
  expiresAt?: number; // operator GC

  // User context
  userId?: string;
  tags?: string[];
  metadata?: Record<string, string>;

  // Versioning (for optimistic concurrency)
  version: number;
  resumeCount: number;

  // Timestamps
  createdAt: number;
  updatedAt: number;
}

The fields tagged "v7" are persisted across all v7 stores. On D1 / DO they pack into a single suspension_context TEXT column (V9 / V5 respectively); on Redis they are JSON-encoded fields on the existing __agents:state:{sessionId} hash; on Postgres they are JSONB columns. See packages/core/src/types/session.ts for the canonical type definition.

Note: Messages are stored separately in the state store (not embedded in state) for O(1) append operations.

Querying Across Sessions

State stores, usage stores, and memory stores support cross-session queries for admin dashboards and analytics:

• listSessions() on SessionStateStore — browse and filter sessions by user, agent type, status, tags, metadata, and date range
• queryUsage() on UsageStore — query usage entries across sessions for cost tracking and analytics
• listEntities() on MemoryStore — list distinct entity/source pairs with memory counts

All methods return paginated results with { total, offset, limit, hasMore }.

See Querying Guide for complete documentation and examples.

Next Steps

• In-Memory Storage - Development and testing
• Redis Storage - Production deployment
• Cloudflare Storage - Edge deployment