Redis Storage

The Redis storage package (@helix-agents/store-redis) provides production-ready state and stream storage using Redis. It supports persistence, multi-process deployments, and atomic operations for parallel tool execution.

When to Use

Good fit:

• Production deployments
• Multi-process/multi-server architectures
• Agents requiring crash recovery
• Real-time streaming across services

Not ideal for:

• Simple development (use in-memory)
• Serverless with cold starts (connection overhead)
• Cloudflare Workers (use D1/Durable Objects)

Installation

npm install @helix-agents/store-redis ioredis

Setup

Basic Connection

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

// Create Redis client
const redis = new Redis({
  host: 'localhost',
  port: 6379,
  // password: 'your-password',
  // tls: {},  // For Redis Cloud/AWS ElastiCache
});

// Create stores
const stateStore = new RedisStateStore(redis);
const streamManager = new RedisStreamManager(redis);

With Connection URL

const redis = new Redis(process.env.REDIS_URL);
// e.g., redis://user:password@host:6379

const stateStore = new RedisStateStore(redis);
const streamManager = new RedisStreamManager(redis);

With Options

const stateStore = new RedisStateStore(redis, {
  prefix: 'myapp:agents:', // Key prefix (default: 'agent')
  ttl: 86400 * 7, // State TTL in seconds (default: 7 days)
});

const streamManager = new RedisStreamManager(redis, {
  prefix: 'myapp:streams:',
  ttl: 86400, // Stream TTL (default: 24 hours)
  maxChunks: 10000, // Max chunks per stream — applied via LTRIM (default: 10000, 0 = unlimited)
});

RedisStateStore

Session vs Run Identifiers

• sessionId: Primary key for all state operations. All Redis keys are scoped by sessionId.
• runId: Execution metadata identifying a specific run within a session.

The key structure uses sessionId as the primary scoping mechanism. Multiple runs within the same session share state.

Key Structure

The store uses a separated key structure for efficient atomic operations:

{prefix}:session:{sessionId}                      -> HASH (main state)
{prefix}:session:messages:{sessionId}             -> LIST (messages)
{prefix}:session:subsession:ids:{sessionId}       -> SET (sub-session IDs)
{prefix}:session:subsession:{sessionId}:{subId}   -> HASH (sub-session details)
{prefix}:session:customstate:{sessionId}          -> HASH (scalar custom state)
{prefix}:session:customstate:arraykeys:{sessionId} -> SET (array field names)
{prefix}:session:customstate:list:{sessionId}:{field} -> LIST (array fields)
{prefix}:session:stops:{sessionId}                -> HASH (run-scoped stop records: runId -> StopRequest JSON)

Basic Operations

// Create a session - one atomic Lua script (claim + full hash + initial
// custom state + indexes), so a concurrent loadState never sees it half-written
await stateStore.createSession('session-123', { agentType: 'my-agent' });

// Load state (reconstitutes from multiple keys)
const state = await stateStore.loadState('session-123');

// Save state (updates existing session; never writes a checkpoint)
await stateStore.saveState('session-123', state);

// Delete session (removes all related keys)
await stateStore.deleteSession('session-123');

Deleting a session

deleteSession first walks the tree: breadth-first from the root, reading each session's local refs (a location field on each ref) and its idx:sessions:parent:<id> ZSET, with a visited set so a cycle ends. It then deletes the descendants deepest first and the root last. Within one session, its other keys go first; its hash, its index entries and its sub-session refs go last, together, in one script. So a session whose delete did not finish (a crash or a dropped connection part-way) keeps its hash, its parent-index entry and its refs: it is still reachable from the root, and its location: 'remote' refs still exclude its remote children. A retried deleteSession of a delete that did not finish walks again and finishes the tree without deleting a remote child. A failed command in the per-session pipelines rejects the call before that session's hash goes. Once a session is fully deleted its refs are gone too, so a later deleteSession of it (for example a retry after a reply lost on a dropped connection) follows the parent index alone and deletes a remote child that shares the store and names it (FU-DELETE-SESSION-REMOTE-CHILD-AFTER-PARENT-GONE). Sessions named by a remote ref are not walked and not deleted.

The parent index follows the session hash: saveState and the commit script move a session's idx:sessions:parent:* entry whenever parentSessionId changes, so a child saved with its parent after its ref was written (the Temporal shape) still reaches the walk. A deleted session's own parent-index ZSET is not deleted; a surviving remote child stays listed there, as on the memory store. Usage is kept in a separate UsageStore and is not deleted (FU-USAGE-STORE-DELETE-SESSION). The idx:sessions:* ZSETs have no TTL while session keys expire on defaultTTL (FU-REDIS-INDEX-ZSET-TTL).

Fixed in this release: the dangling checkpoint pointer (RM-38), cloneSession (RM-27), listCheckpoints (RM-53)

• saveState no longer writes a checkpoint (RM-37, RM-38). It used to point the session at a checkpoint id that was never stored, so getLatestCheckpoint fell back to stepCount ordering and could return an earlier turn's checkpoint. saveState now never touches the pointer, and getLatestCheckpoint is pointer-only: it returns the checkpoint of the most recent commit, never the highest-stepCount row. A pointer naming a checkpoint that does not exist throws a typed state_checkpoint_not_found.
• cloneSession is one commit (a clone checkpoint) through the commit script (RM-27).
• listCheckpoints reads total and the page in one MULTI (RM-53), ordered by stepCount with the checkpoint id as tiebreaker. While a run is live a new checkpoint can still shift the offsets between pages; read it once the session is idle and de-duplicate by id.

The finding records stay open until their conformance cells land; see the conformance findings.

Upgrading: rebuild the session-status index once

The old saveState could leave the session-status secondary index (<prefix>:idx:sessions:status:*) out of step with a session's status. saveState now moves the index entry in the same script as the status write. rebuildIndexes() only ADDS index entries, it never removes a stale one, so after upgrading delete the <prefix>:idx:sessions:status:* sorted sets and then run await stateStore.rebuildIndexes() once: it re-adds every session under its current status, so listSessions() status filters are accurate for sessions written by the previous release. See the upgrade guide.

Atomic Operations

These operations are safe for concurrent access:

// Append messages - a commit (kind 'append'): rows + checkpoint + pointer in the
// commit script, fenced on commit.runId, only for an existing session
await stateStore.appendMessages('session-123', [{ role: 'assistant', content: 'Hello!' }], {
  streamSequence, // the live stream head (core resolveCommitSequence)
  runId, // the session's current live run (RunSupersededError otherwise)
});

// Merge custom state - applies StepWrites ops atomically
await stateStore.mergeCustomState('run-123', {
  ops: [
    { kind: 'replace', key: 'count', value: 5 }, // Scalar: HSET
    { kind: 'append', key: 'items', items: ['new'] }, // Array: RPUSH (append)
    { kind: 'replace', key: 'replaced', value: [1, 2] }, // Full replacement
  ],
  warnings: [],
});

// Update status - uses HSET
await stateStore.updateStatus('run-123', 'completed');

// Increment step count - uses HINCRBY (atomic)
const newCount = await stateStore.incrementStepCount('run-123');

Atomic Write (saveStateAndPromoteStaging)

The v7 atomic write primitive is a step commit run by the store's commit script (EVAL-loaded once, then EVALSHA-cached). The script executes the message append, state hash update, staging promotion, and checkpoint creation as a single atomic Redis operation — there is no client-side compensating-write window.

// Single round-trip, fully atomic on the server
await stateStore.saveStateAndPromoteStaging(
  sessionId,
  newState,
  pendingMessages,
  { stepId: 'step-7', stepCount: 7, streamSequence: 142, runId },
  { expectedVersion: state.version }
);

This satisfies cross-runtime invariant C-1: pendingClientToolCalls, clientToolCallOwnership, completed phase-1 messages, the checkpoint, and suspendedStepId all become visible together. Redis 6+ is required (already the documented minimum).

EVAL is required

There is no non-atomic fallback. The former allowSequentialFallback option was removed (P3.R3-BC-LUA-FALLBACK), so a Redis deployment that forbids EVAL/EVALSHA (some hardened managed offerings, certain proxy layers) cannot run this store: a Lua-script failure surfaces as a hard error instead of falling back to a sequential pipeline.

The commit script

Every commit (appendMessages, truncateMessages, commitState, promoteStaging / saveStateAndPromoteStaging, cloneSession), plus updateRunStatus and renewRunLease, runs through ONE Lua script. TypeScript pre-reads the view and passes every key the script may touch in KEYS; the script re-validates that view inside the atomic step (the current run, the live runs and their leases and owner tokens, the live message count a rewind was planned on) and plans every write before applying any, so an error leaves nothing written. The store clock is redis.call('TIME').

When the view moved since the pre-read (concurrent writers), the script answers RETRY and the store re-reads and re-sends with backoff. If that keeps happening on every attempt, the call throws RedisCommitContentionError (exported from @helix-agents/store-redis). It extends StaleStateError (and keeps its name), so code that already retries a stale write retries it too; lastReason names what diverged on the last attempt.

saveState atomicity

RedisStateStore.saveState is one Lua script (SAVE_STATE_ATOMIC_SCRIPT): the run fence (fenceRunId / fenceIdle), the version check, the main hash fields, the session-status index move and the full customState replacement apply together or not at all. It is not a commit: it writes no checkpoint and never changes checkpointId, latestTruncateCheckpointId or currentRunId.

createSession atomicity

createSession is one Lua script. It makes the create-if-absent claim, clears per-session keys left behind by a previous session with the same id, and writes the initial custom state, the secondary-index entries and every initial field (including status) with their TTLs. A concurrent loadState therefore sees either no session or the whole one, and a recreated session inherits no messages, runs, sub-session refs or checkpoint index (RM-50).

v7 SessionState Fields

The following fields are persisted as JSON-encoded entries on the existing __agents:state:{sessionId} hash — no new keys are introduced and existing keys remain read-compatible (records that pre-date this version simply return undefined for the new fields):

• suspendedAwaitingChildren — map of parentToolCallId → SuspendedChildWait
• suspendedStepId — mid-step suspension marker
• tracingContext — { traceId, rootSpanId } for trace continuity across resumes
• expiresAt — epoch-ms hint for operator GC via expiredSessionCleanup
• failureReason — discriminator for failed children (e.g. 'parent_suspended')
• completedClientToolCalls — root-session tombstone map for HITL idempotency

compareAndSetStatus Return Shape (v7)

The Lua-backed CAS now returns a discriminated union including the prior status and version on conflict, so callers can make richer retry decisions without an extra round-trip:

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

if (result.ok) {
  // result.newVersion = post-bump version
} else {
  // result.currentStatus, result.currentVersion = stored values at conflict
}

Parallel Tool Execution

The key structure enables safe parallel updates:

// Two tools running in parallel
await Promise.all([
  // Tool 1: Appends to items array
  stateStore.mergeCustomState(runId, {
    ops: [{ kind: 'append', key: 'items', items: ['from-tool-1'] }],
    warnings: [],
  }),
  // Tool 2: Appends to items array
  stateStore.mergeCustomState(runId, {
    ops: [{ kind: 'append', key: 'items', items: ['from-tool-2'] }],
    warnings: [],
  }),
]);

// Result: items = [...original, 'from-tool-1', 'from-tool-2']
// (order may vary, but both appends are preserved)

Message Queries

// Paginated messages (for UI)
const result = await stateStore.getMessages('run-123', {
  offset: 0,
  limit: 50,
  includeThinking: false,
});

// Message count
const count = await stateStore.getMessageCount('run-123');

Message Truncation

For crash recovery, messages can be truncated to match a checkpoint:

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

This is used internally during crash recovery when resuming from a checkpoint. If a crash occurred mid-step, messages added after the checkpoint are orphaned and need to be removed. The messageCount parameter comes from the checkpoint's messageCount field. The trim, the update of the session hash's messageCount (read by listSessions), the truncate checkpoint and the pointer move run in the commit script, so the listed count matches the live list length after a truncate (RM-22). A rewinding entry (resume from_checkpoint, retry()) does not call it: its run-start commit carries the rewind (startRun.truncateTo).

RedisStreamManager

Key Structure

{prefix}:stream:{streamId}       -> STREAM (Redis Stream for chunks)
{prefix}:stream:meta:{streamId}  -> HASH (stream metadata)

Basic Operations

// Create writer
const writer = await streamManager.createWriter('stream-123', 'run-123', 'researcher');

// Write chunks (uses XADD)
await writer.write({
  type: 'text_delta',
  agentId: 'run-123',
  agentType: 'researcher',
  timestamp: Date.now(),
  delta: 'Hello world',
});

// Close writer (releases resources, doesn't end stream)
await writer.close();

// Create reader (uses XREAD with blocking)
const reader = await streamManager.createReader('stream-123');
if (reader) {
  for await (const chunk of reader) {
    console.log(chunk);
  }
  await reader.close();
}

// End stream
await streamManager.endStream('stream-123', { result: 'done' });

// Or fail stream
await streamManager.failStream('stream-123', 'Error message');

Real-Time Streaming

Redis Streams provide real-time delivery:

// Writer (agent execution)
const writer = await streamManager.createWriter(streamId, agentId, agentType);
for (const token of tokens) {
  await writer.write({ type: 'text_delta', delta: token, ... });
  // Reader receives immediately
}

// Reader (UI/client) - receives chunks as they're written
const reader = await streamManager.createReader(streamId);
for await (const chunk of reader) {
  updateUI(chunk);  // Real-time updates
}

Resumable Streams

Support for client reconnection:

// Get stream info
const info = await streamManager.getStreamInfo('stream-123');
console.log(info?.totalChunks);
console.log(info?.latestSequence);
console.log(info?.status);

// Create resumable reader
const reader = await streamManager.createResumableReader('stream-123', {
  fromSequence: lastKnownSequence, // Resume from this point
});

if (reader) {
  console.log(`Starting from sequence ${reader.currentSequence}`);
  for await (const chunk of reader) {
    // Process chunks, track position
    savePosition(reader.currentSequence);
  }
}

Stream Cleanup

For crash recovery and stream reset scenarios:

// Clean up chunks beyond a specific step (atomic Lua script)
await streamManager.cleanupToStep('stream-123', stepCount);

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

cleanupToStep() removes all chunks with step > stepCount. This is used during crash recovery to remove orphaned chunks that were written after the checkpoint being resumed from. The operation uses an atomic Lua script to ensure consistency.

resetStream() clears the entire stream and its owner run. No runtime calls it: a session's stream keeps every run's chunks, and each run's startSequence marks where its chunks begin.

Stream ownership and fences

• claimStream(streamId, runId, { owner }) is one Lua compare-and-set on the stream's currentRunId: it applies only while the owner is still the one the caller read, and returns false (nothing written) otherwise or when the stream does not exist.
• endStream / failStream / pauseStream with a run fence check the owner and transition inside the same CAS script, so a finished run's late call never ends or pauses a newer run's stream.
• A chunk of the stream's own session stamped with a runId is dropped once a newer run owns the stream; the check runs inside the chunk-append Lua script.
• createResumableReader(streamId, { fromSequence, waitForLive: true }) parks on a terminal or missing stream until the new run makes it live, and every reader exposes catchUpHead / caughtUp.

See Stream ownership and the run fence.

Stream TTL

Streams auto-expire to prevent unbounded growth:

const streamManager = new RedisStreamManager(redis, {
  ttl: 3600, // 1 hour
});

// Stream keys expire after TTL
// Completed streams can be read until expiry

Complete Example

import Redis from 'ioredis';
import { JSAgentExecutor } from '@helix-agents/runtime-js';
import { RedisStateStore, RedisStreamManager } from '@helix-agents/store-redis';
import { VercelAIAdapter } from '@helix-agents/llm-vercel';

// Setup
const redis = new Redis(process.env.REDIS_URL);
const stateStore = new RedisStateStore(redis, { prefix: 'myapp:' });
const streamManager = new RedisStreamManager(redis, { prefix: 'myapp:' });

const executor = new JSAgentExecutor(stateStore, streamManager, new VercelAIAdapter());

// Execute agent
const handle = await executor.execute(ResearchAgent, 'Research AI agents', {
  sessionId: 'my-session-1',
});

// Stream in real-time
const stream = await handle.stream();
if (stream) {
  for await (const chunk of stream) {
    process.stdout.write(chunk.type === 'text_delta' ? chunk.delta : '');
  }
}

// Get result
const result = await handle.result();
console.log('\\nResult:', result.output);

// Cleanup (optional)
await stateStore.deleteSession(handle.sessionId);

// Close connection when done
await redis.quit();

Production Considerations

Connection Pooling

Use connection pooling for high-throughput:

const redis = new Redis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: 3,
  enableReadyCheck: true,
  connectTimeout: 10000,
});

Cluster Support

Redis Cluster is not supported. Use a single Redis primary: standalone, or with replicas and Sentinel failover. RedisStateStore's atomic operations are multi-key Lua scripts (every commit touches the session hash, its message list, run hashes, checkpoint keys and global secondary indexes), and those keys carry no shared hash tag:

• The global secondary-index keys (<prefix>:idx:sessions:*) hash to different slots from a session's own keys.
• Some older scripts build per-field custom-state list keys inside the script instead of declaring them in KEYS (the commit script declares every key it touches).

On a cluster these scripts fail with CROSSSLOT errors or touch undeclared keys. RedisStreamManager has not been verified on Cluster either. The key layout would need hash tags to support Cluster, which is not implemented.

Error Handling

Handle Redis connection errors:

redis.on('error', (err) => {
  console.error('Redis error:', err);
});

redis.on('connect', () => {
  console.log('Redis connected');
});

// Graceful shutdown
process.on('SIGTERM', async () => {
  await redis.quit();
  process.exit(0);
});

Monitoring

Monitor Redis memory and connections:

# Memory usage
redis-cli INFO memory

# Connected clients
redis-cli INFO clients

# Key count
redis-cli DBSIZE

# Specific key patterns
redis-cli KEYS "myapp:state:*" | wc -l

Backup and Recovery

Agent state can be recovered from Redis after crashes:

// After process restart
const handle = await executor.getHandle(MyAgent, savedSessionId);

if (handle) {
  const { canResume, reason } = await handle.canResume();
  if (canResume) {
    // Resume from saved state
    const newHandle = await handle.resume();
    const result = await newHandle.result();
  }
}

Limitations

No Transactions Across Keys

Atomic operations work per-key, not across multiple keys:

// This is NOT atomic across both updates
await stateStore.updateStatus(sessionId, 'completed');
await stateStore.mergeCustomState(sessionId, {
  ops: [{ kind: 'replace', key: 'finalResult', value: result }],
  warnings: [],
});

// If process crashes between these, state may be inconsistent

Memory Usage

Monitor Redis memory with many concurrent agents:

// Set maxmemory and policy
// redis.conf: maxmemory 1gb
// redis.conf: maxmemory-policy allkeys-lru

// Or use TTL aggressively
const stateStore = new RedisStateStore(redis, { ttl: 3600 });

Network Latency

Each operation is a network round-trip. Batch when possible:

// Multiple sequential calls - multiple round-trips (and checkpoints)
await stateStore.appendMessages(sessionId, [msg1], commit);
await stateStore.appendMessages(sessionId, [msg2], commit);

// Single call - one round-trip, one commit
await stateStore.appendMessages(sessionId, [msg1, msg2], commit);

Next Steps

• In-Memory Storage - For development
• Cloudflare Storage - For edge deployment
• Temporal Runtime - Uses Redis storage