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
@helix-agents/store-cloudflare	D1StateStore	DOStreamManager	Cloudflare Workers

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>;
  saveState(sessionId: string, state: SessionState): Promise<void>;

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

  // 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>;
}

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, { values: { key1: 'value1' } }),
  stateStore.mergeCustomState(sessionId, { values: { key2: 'value2' } }),
]);

State Merge Semantics

The mergeCustomState operation uses smart merging:

// Array delta mode: new items are appended
// Tool 1: draft.items.push('a')
// Tool 2: draft.items.push('b')
// Result: items = [...original, 'a', 'b']

// Scalar values: last write wins
// Tool 1: draft.count = 5
// Tool 2: draft.count = 10
// Result: count = 5 or 10 (depends on timing)

// Null means delete
// draft.temp = null
// 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:

• execute() creates a new run with turn: 1
• resume() creates a new run with incremented turn number

Run status values:

• running - Execution in progress
• completed - Finished successfully with output
• failed - Terminated with error
• interrupted - User-requested pause (can be resumed)

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
  endStream(streamId: string, finalOutput?: unknown): Promise<void>;
  failStream(streamId: string, error: string): Promise<void>;

  // Optional: Resumable streams
  createResumableReader?(
    streamId: string,
    options?: ReaderOptions
  ): Promise<ResumableStreamReader | null>;
  getStreamInfo?(streamId: string): Promise<StreamInfo | null>;
  pauseStream?(streamId: string): Promise<void>;
  resumeStream?(streamId: string): Promise<void>;

  // Optional: Content reconstruction (for mid-stream refresh)
  getAllChunks?(streamId: string): Promise<StreamChunk[]>;
  getChunksFromStep?(streamId: string, fromStep: number): Promise<StreamChunk[]>;
}

Resumable Stream Features

Different implementations support different resumability features:

Feature	InMemory	Redis	Cloudflare DO
createResumableReader	No	Yes	Yes
getStreamInfo	Yes	Yes	Yes
getAllChunks	Yes	Yes	Yes
getChunksFromStep	Yes	Yes	Yes
pauseStream / resumeStream	No	Yes	Yes
Sequence tracking	No	Yes	Yes
Multi-process support	No	Yes	Yes

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. Use to check if a stream is still active before connecting.
• 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

  // Hierarchy
  parentSessionId?: string; // Parent if this is a sub-agent

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

  // Output
  output?: TOutput; // Final structured output
  error?: string; // Error message if failed

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

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

Next Steps

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