@helix-agents/store-postgres

PostgreSQL implementations of store interfaces for production use. Provides durable session-state, lock-manager, and usage-tracking implementations that work across all runtimes (JS, Temporal, DBOS, Cloudflare via Neon/Hyperdrive).

Streaming: This package does NOT provide a stream manager. Pair PostgresStateStore with RedisStreamManager from @helix-agents/store-redis (recommended) or InMemoryStreamManager for live event streaming.

Installation

# Node.js / Temporal / DBOS — use pg
npm install @helix-agents/store-postgres pg

# Cloudflare Workers / edge — use Neon serverless
npm install @helix-agents/store-postgres @neondatabase/serverless

# Cloudflare Workers via Hyperdrive — no extra dependency required
npm install @helix-agents/store-postgres

Quick Start

import { Pool } from 'pg';
import {
  createPgAdapter,
  PostgresMigrations,
  PostgresStateStore,
  PostgresLockManager,
  PostgresUsageStore,
} from '@helix-agents/store-postgres';

// 1. Wrap your underlying client in an adapter
const pool = new Pool({ connectionString: process.env.POSTGRES_URL });
const adapter = createPgAdapter(pool);

// 2. Run migrations (idempotent, advisory-locked, transactional)
const { current } = await new PostgresMigrations(adapter).run();
console.log(`Schema is at V${current}`);

// 3. Construct stores
const stateStore = new PostgresStateStore(adapter);
const lockManager = new PostgresLockManager(adapter);
const usageStore = new PostgresUsageStore(adapter);

Session vs Run Identifiers

• sessionId: Primary key for all state operations. A session contains all messages, state, and checkpoints.
• runId: Execution metadata identifying a specific run within a session.

All store methods use sessionId as the primary key for lookups and storage.

Adapter Factories

Each store accepts a PostgresAdapter so the same store implementations work across pg, Neon, and Hyperdrive without changes.

createPgAdapter(pool: Pool)

Wraps a pg.Pool. Each transaction() checks out a PoolClient, issues BEGIN/COMMIT/ROLLBACK, and releases it on completion.

import { Pool } from 'pg';
import { createPgAdapter } from '@helix-agents/store-postgres';

const pool = new Pool({ connectionString: process.env.POSTGRES_URL });
const adapter = createPgAdapter(pool);

Use for: Node.js, Temporal workers, DBOS workers, long-lived servers.

createNeonAdapter(pool)

Wraps a @neondatabase/serverless Pool. The Neon transport supports the same query/transaction surface as pg, so the same store implementations run unchanged against Neon's HTTP/WebSocket transport.

import { Pool } from '@neondatabase/serverless';
import { createNeonAdapter } from '@helix-agents/store-postgres';

const pool = new Pool({ connectionString: process.env.NEON_URL });
const adapter = createNeonAdapter(pool);

Use for: Neon serverless, Cloudflare Workers (with Neon as origin), edge runtimes.

createHyperdriveAdapter(hyperdrive)

Wraps a Cloudflare Hyperdrive binding. Allows running PostgresStateStore from inside a Cloudflare Worker, with Hyperdrive handling connection pooling and edge-routing to your origin Postgres.

import { createHyperdriveAdapter } from '@helix-agents/store-postgres';

// In a Cloudflare Worker fetch handler:
const adapter = createHyperdriveAdapter(env.HYPERDRIVE);

Custom adapters

PostgresAdapter is a small interface (query, transaction, close). Any Postgres client can be wrapped to plug into the same stores.

Migrations

PostgresMigrations.run() applies all pending migrations, one version at a time, in three phases: the version's transactional statements commit under pg_advisory_xact_lock; its CREATE INDEX CONCURRENTLY statements (if any) run outside a transaction; then its version row commits under the same lock. Every statement is idempotent, so concurrent runners record each version once (see Concurrent migrators). The current schema is V19.

Version	Description
V1	Session-centric schema with all base tables and indexes.
V2	Add remote_json column to __agents_sub_session_refs for remote agents.
V3	Add pending_client_tool_calls JSONB to __agents_states for client-executed tools.
V4	Add root_session_id TEXT and client_tool_call_ownership JSONB to __agents_states.
V5	Add completed_client_tool_calls JSONB for the durable already_completed marker.
V6	Add mode column on states + completion_reason on runs (consumed by runtime-dbos).
V7	Add suspension_context JSONB + partial index on expiresAt for stateless HITL/client-tool flows.
V8	Add composite (session_id, status, turn) index on __agents_runs for efficient listRuns({ status }) queries (built CONCURRENTLY to avoid lock contention on hot-path writes).
V9	Denormalize message_count INTEGER onto __agents_states, eliminating the O(N×M) correlated COUNT(*) subquery in listSessions. Backfills existing rows on apply.
V10	StepWrites unification: replaces __agents_staging.patches + .merge_changes JSONB columns with a single writes JSONB column. Migration TRUNCATEs the staging table (ephemeral data); checkpoint replay handles in-flight steps. Forward-only.
V11	Add completion_delivered to sub-session refs.
V12	Add forced_completion JSONB to states (deterministic forced completion).
V13	Add error_detail JSONB to states (structured terminal failure detail).
V14	Checkpoint commit kind / run_id / keep_count; run lease, start digest and owner columns.
V15	Run entry columns and the abandoned_run_starts tombstone table.
V16	__agents_stop_requests (one row per (session_id, run_id), the run-scoped stop record behind requestStop / readStop); drops __agents_interrupt_flags.
V17	Add location ('local' or 'remote') to __agents_sub_session_refs (SubSessionRef.location, used by the deleteSession cascade).
V18	__agents_runs.request_key / request_digest and a per-session unique index on (session_id, request_key) (D-2 request keys). getRunByRequestKey reads it; the run-start commit checks the holder under the session row lock, and a racing duplicate-key insert (23505) is answered RunStartRejectedError('request_key_taken'), never raw.
V19	The held prior (DO companion lifecycle MR B): nullable held_prior JSONB, outcome_generation INTEGER and held_generation INTEGER on __agents_sub_session_refs (holdPriorResult / releasePriorResult; every reopen bumps the generation).
V20	Add workspace_ref JSONB to states (SessionState.workspaceRef, RM-19; nullable, no backfill).

const migrations = new PostgresMigrations(adapter, {
  tablePrefix: '__agents_', // default; must match `[a-zA-Z_][a-zA-Z0-9_]*`, at most 34 chars
});

const { applied, current } = await migrations.run();
// applied: [17]      (versions just applied)
// current: 17        (final schema version)

Migrations record applied versions in __agents_migrations, and serialize concurrent runs through __agents_migration_lease. Re-running on an up-to-date database is a no-op. Multiple tenants can share a database by passing distinct tablePrefix values.

Upgrading: PostgreSQL 12 or newer

From this release, PostgresMigrations.run() refuses a server older than PostgreSQL 12 (it reads pg_stat_progress_create_index, added in 12), before it creates or changes anything: @helix-agents/store-postgres migrations need PostgreSQL 12 or newer (server_version_num …). Upgrade the server first; schemas already migrated on an older server are unchanged and continue from where they are once the server is 12+.

Upgrading: table prefix at most 34 characters

From this release, tablePrefix must be at most 34 characters (MAX_TABLE_PREFIX_LENGTH, exported). Every store and PostgresMigrations refuses a longer one at construction. Postgres truncates identifiers to 63 bytes, and the longest name the migrations derive from the prefix adds 29 characters to it (idx_<prefix>states_expires_at_partial). With a longer prefix, names were silently truncated, and from 39 characters on, two indexes' names collided, so one of them was never created.

If your prefix is longer, rename the tables (and their indexes) to a shorter prefix before upgrading, or start a fresh schema with one.

Concurrent migrators

Several processes can call run() against one database at once (app instances or workers starting together). The contract:

• One migrator at a time. run() first takes the migration lease, one row in <prefix>migration_lease (holder, expires_at by the database clock), and holds it until it finishes. Other migrators wait for it. Without it, migrators overlap, and a CREATE INDEX CONCURRENTLY (which waits for every older transaction) deadlocks against another migrator's transaction that waits for the build's table lock, or against a second migrator's build of the same index.
  • The holder renews the lease every leaseTtlMs / 3 (default ttl 30 s) and releases it at the end. A crashed holder's lease is taken over leaseTtlMs after its last renewal.
  • The last statement of each phase transaction (a version's statements; its version row) renews the lease and proves it is still the run's (clock_timestamp()), so a run that lost its lease rolls that phase back and never records a version: it fails that attempt with MigrationLeaseLostError and re-enters. No lock is held while a phase runs, so a long phase (a backfill on a big table) never blocks its own renewal. The lease is also checked before each concurrent index build.
  • Renewal runs on its own pooled connection while a build runs: give the adapter's pool at least 2 connections.
  • The migrations and lease tables themselves are created under pg_advisory_xact_lock (a bare concurrent CREATE TABLE IF NOT EXISTS is not race-free in Postgres: the loser fails on pg_type_typname_nsp_index).
  • Serialization holds only among migrators of this version or later: upgrade every instance that runs migrations.
• Waiting: peerWaitMs. The total time one run() waits on others, across its re-entries: for the lease while another migrator holds it, and for an index build still running on one of its tables (a crashed holder's backend can outlive it, or an operator's build). It is spent only while there is something to wait for.
  • The default is 10 minutes because the holder may be building an index CONCURRENTLY on a large runs or states table, which can take minutes and also waits for every transaction already touching the table.
  • When it runs out, run() throws MigrationLeaseTimeoutError (naming the holder) or MigrationIndexNotReadyError (a build), and this process's start fails. The timed-out attempt records nothing; versions the run recorded on an earlier attempt stay recorded. The holder keeps going and records its versions, so a restart finds them applied. Lower it to fail a waiting instance's start sooner, under a supervisor that restarts it.
• Crashed builds. An INVALID index left by a crashed CREATE INDEX CONCURRENTLY is dropped before the version is rebuilt. A build still in progress also reads indisvalid = false, so each candidate is first retired under SHARE UPDATE EXCLUSIVE on its table, which no live build (of any role, in any phase) lets it take: re-checked by OID and renamed to idx_<prefix>dead_<oid>. Only that name is then dropped. A busy table (its lock not granted within 5 s, or within what is left of peerWaitMs) is skipped and logged (postgres.migration.invalid_index.busy); phase 3 then refuses that version until a re-entry retires it.
• Invariant. A version row is written only when every index that version builds concurrently exists and is valid. If one is not, that attempt fails with MigrationIndexNotReadyError (exported). A recorded version never covers an INVALID index.
• Retries. run() re-enters on SQLSTATE 40P01 (a deadlock, e.g. against an operator's DDL), MigrationIndexNotReadyError or MigrationLeaseLostError, up to 5 attempts with jittered backoff, and logs each retry at warn (postgres.migration.retry). Every step is idempotent, so a re-entry redoes only what is left, and applied lists every version the run recorded over all its attempts. Any other error is thrown at once.
• Options. peerWaitMs must be a finite number ≥ 0 and leaseTtlMs at least 100 ms; others are refused at construction. A mixed-case tablePrefix works: the DDL's names are unquoted, so Postgres folds them to lower case, and the runner looks names up the same way (App_ and app_ name the same tables).
• Requirements. PostgreSQL 12 or newer (run() checks server_version_num and refuses older servers). A build on one of its tables is found through pg_stat_progress_create_index and scoped to the table through pg_locks, which every role can read, so a build by another role (even without pg_read_all_stats) is waited for when it is on one of its tables, and never for an unrelated table.

const migrations = new PostgresMigrations(adapter, {
  peerWaitMs: 120_000, // total wait on the lease holder / a build on our tables per run() (default 600_000)
  leaseTtlMs: 30_000, // a crashed holder's lease is taken over after this (default 30_000)
  logger: { warn: (event, data) => log.warn(event, data) },
});

PostgresStateStore

Full implementation of SessionStateStore from @helix-agents/core.

const stateStore = new PostgresStateStore(adapter, {
  tablePrefix: '__agents_', // default
  logger: customLogger, // optional
});

await stateStore.createSession('session-123', { agentType: 'my-agent' });
const state = await stateStore.loadState('session-123');
await stateStore.saveState('session-123', state);

Size limits

PostgresStateStore declares no finite limit (limits = UNLIMITED_STORE_LIMITS), so a runtime never refuses a write in advance for it. A value larger than Postgres's own jsonb cap fails with SQLSTATE 54000 (program_limit_exceeded); the write paths that bind jsonb (staging, the step commit's state, rows and checkpoint, saveState, commitState, appendMessages, the run-start commit, createSession, cloneSession, mergeCustomState, createCheckpoint, patchMetadata, updateStatus) map it to core's StateSizeExceededError (state_size_exceeded, non-retryable) with limitBytes = POSTGRES_MAX_JSONB_BYTES (268 435 455, 2^28 − 1, the cap Postgres's message names) and the original error as cause, instead of a retryable CommitNotAppliedError. The failed statement writes nothing (a commit's transaction rolls back). isPostgresSizeError(err) is the check, by the SQLSTATE code, never the message text; both are exported.

Deleting a session

deleteSession runs in one transaction. It locks the root state row (SELECT … FOR UPDATE), reads the id set with one WITH RECURSIVE query (the session's location = 'local' refs plus the sessions whose parent_session_id is it, minus the sessions it names through a remote ref; UNION stops cycles), then deletes WHERE session_id = ANY($ids) from every session-keyed table, usage included, and the state rows last. The parent_session_id index (idx___agents_states_parent) already existed, so the walk needs no new index. See Deleting a session.

Atomic save + promote (v7)

PostgresStateStore implements the v7 saveStateAndPromoteStaging primitive via a single BEGIN; UPDATE/INSERT/DELETE; COMMIT transaction. State, appended messages, staged changes, and the new checkpoint are all persisted atomically — the cross-runtime invariant C-1 contract that lets runtimes suspend without leaking partial writes.

const { checkpointId, newVersion } = await stateStore.saveStateAndPromoteStaging(
  'session-123',
  nextState,
  appendMessages,
  { stepId, stepCount, streamSequence },
  { expectedVersion: previousVersion } // throws StaleStateError on mismatch
);

Compare-and-set status

const result = await stateStore.compareAndSetStatus('session-123', ['running'], 'interrupted');

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

The version field is incremented on every successful CAS. Stores reject saveState calls whose expectedVersion doesn't match, which prevents stale processes from clobbering newer state.

Checkpoints

const checkpoint = await stateStore.getCheckpoint('cpv1-...');
const latest = await stateStore.getLatestCheckpoint('session-123');
const result = await stateStore.listCheckpoints('session-123', { limit: 10 });

Paged listCheckpoints can repeat or skip checkpoints (RM-54)

listCheckpoints orders the page by step_count alone, with no tiebreaker. stepCount restarts every turn, so a long multi-turn session has many tied checkpoints, and their order can differ between page queries. In a probe with 600 checkpoints over about 200 turns, paging at limit: 50 returned 41 checkpoints twice and never returned 41 others. Dedupe by id and do not treat a paged walk as complete. See RM-54 in the conformance findings.

v7 SessionState fields

PostgresStateStore persists the full v7 stateless-suspension surface:

• suspendedAwaitingChildren / suspendedStepId / tracingContext / expiresAt — packed into suspension_context JSONB (V7 migration). The partial index on suspension_context->>'expiresAt' supports operator-side cleanup of abandoned suspended sessions.
• pendingClientToolCalls / completedClientToolCalls / clientToolCallOwnership — dedicated JSONB columns (V3/V4/V5).
• failureReason — discriminator on status === 'failed' (e.g. 'parent_suspended' for the γ-cascade).
• mode — 'standard' | 'persistent', write-once on first execute() (V6).
• version / resumeCount — optimistic concurrency + resume tracking.

See @helix-agents/core for field semantics.

PostgresLockManager

Distributed lock manager backed by Postgres advisory locks. Returns a fencing token so split-brain scenarios can be rejected by downstream operations.

import { PostgresLockManager } from '@helix-agents/store-postgres';

const lockManager = new PostgresLockManager(adapter, {
  tablePrefix: '__agents_',
  defaultTTLMs: 30_000,
});

const lock = await lockManager.acquire('session-123', { ttlMs: 30_000 });
if (lock) {
  console.log('Fencing token:', lock.fencingToken);

  // Refresh during long operations
  await lock.refresh();

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

Use with executor

import { JSAgentExecutor } from '@helix-agents/runtime-js';

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

PostgresUsageStore

Records token / tool / sub-agent / custom usage entries; supports filter, range, and rollup queries.

import { PostgresUsageStore } from '@helix-agents/store-postgres';

const usageStore = new PostgresUsageStore(adapter);

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 },
});

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

// Filtered
const tokenOnly = await usageStore.getEntries('session-123', { kinds: ['tokens'] });

// Rollup
const rollup = await usageStore.getRollup('session-123', { includeSubAgents: true });

Querying

List Sessions

import type { ListSessionsOptions } from '@helix-agents/core';

const result = await stateStore.listSessions({
  agentType: 'chat-agent',
  status: 'active',
  orderBy: { field: 'updatedAt', direction: 'desc' },
  limit: 25,
});

Query Usage

import type { QueryUsageOptions } from '@helix-agents/core';

const result = await usageStore.queryUsage({
  agentType: 'chat-agent',
  kinds: ['tokens'],
  dateRange: { from: Date.now() - 7 * 24 * 60 * 60 * 1000 },
  limit: 100,
});

See the Querying Guide for complete documentation.

Streaming

This package does not include a stream manager. Pair the state store with one of:

• RedisStreamManager from @helix-agents/store-redis (recommended for production)
• InMemoryStreamManager from @helix-agents/store-memory (dev/tests)
• DurableObjectStreamManager from @helix-agents/store-cloudflare (Cloudflare deploys)

Integration Tests

Integration tests run against a real Postgres instance. The repo's docker-compose.test.yml exposes a Postgres 16 service on port 5433:

# Start Postgres
docker compose -f docker-compose.test.yml up -d postgres-store

# Run integration tests
POSTGRES_URL=postgresql://test:test@localhost:5433/helix_agents_test \
  pnpm --filter @helix-agents/store-postgres run test:integration

The harness creates a unique tablePrefix per test run, applies all migrations via PostgresMigrations.run(), and drops the prefixed tables on teardown. The full integration suite covers state store contract, lock manager, usage store, atomic suspension writes, migration sequencing, and concurrent CAS scenarios.

When to Use

Choose Postgres stores when:

• Running in production with an existing Postgres operational footprint
• You need transactional guarantees for state + messages + checkpoints
• You want a single durable backend across runtimes (JS / Temporal / DBOS / Cloudflare)
• You need Hyperdrive or Neon for edge deploys

Consider other stores when:

• Developing locally (@helix-agents/store-memory)
• Postgres infrastructure isn't available (@helix-agents/store-redis)
• Deploying primarily to Cloudflare without origin Postgres (@helix-agents/store-cloudflare)

See Also

• Helix concepts — session/run model, stateless suspension, hooks
• @helix-agents/core — SessionStateStore interface
• v6 → v7 upgrade guide
• Distributed Coordination — CAS operations, version management
• Client-Executed Tools