Agent Substrate

PostgreSQL WAL, Event Log & Real-Time Reactivity

In production environments, worker pods get evicted by Kubernetes, nodes encounter Out-Of-Memory (OOM) errors, and spot instances get reclaimed mid-turn.

Agent Substrate guarantees zero data loss, deterministic replay, and zero duplicate side-effects via its append-only PostgreSQL Write-Ahead Log (WAL) and an event-driven reactivity layer powered by PostgreSQL LISTEN / NOTIFY.

The Durability Architecture

Agent Substrate strictly decouples execution compute from persistent state. Agents never write raw SQL or manage database connections directly. Instead, the runtime uses a two-tier architecture:

  1. Kernel Protocols (substrate.kernel.runtime): Pure abstract protocols (EventLogProtocol, SignalBusProtocol, InboxProtocol).
  2. Infrastructure Backends (substrate.infrastructure.runtime): High-performance PostgreSQL implementations wrapped around asyncpg.

1. Append-Only Event Log & Optimistic Concurrency

Every state delta, model token turn, and tool execution result is recorded sequentially in substrate_event_log:

sql
CREATE TABLE substrate_event_log (
    run_id  TEXT        NOT NULL,
    seq     INTEGER     NOT NULL,
    kind    TEXT        NOT NULL,
    payload JSONB       NOT NULL DEFAULT '{}',
    ts      TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (run_id, seq)
);

Advisory Locking (pg_advisory_xact_lock)

To guarantee that two concurrent workers racing on the same agent run never clobber each other or interleave sequences out of order, EventLog.append() acquires a transactional PostgreSQL advisory lock hashed to the run_id:

python
async with conn.transaction():
    await conn.execute("SELECT pg_advisory_xact_lock($1)", lock_key)
    current = await conn.fetchval(
        "SELECT COALESCE(MAX(seq), -1) FROM substrate_event_log WHERE run_id = $1",
        run_id,
    )
    if current != expected_seq:
        raise ConcurrentAppendError(
            f"expected seq {expected_seq}, got {current}",
            run_id=run_id,
            expected_seq=expected_seq,
            actual_seq=current,
        )
    # Insert new event at current + 1

If an append conflicts with a concurrent worker or network retry, a ConcurrentAppendError is raised immediately, preserving linear state history.

2. Event-Driven Streaming via LISTEN / NOTIFY

Polling the database (SELECT ... WHERE seq > last_seq) introduces latency and exhausts database IOPS. Instead, Agent Substrate uses PostgreSQL's native LISTEN and pg_notify for sub-millisecond, push-based reactivity.

How Real-Time Reactivity Works

  1. Transactional Notification: When an event is committed, append() notifies the channel inside the same transaction:

    python
    await conn.execute("SELECT pg_notify($1, $2)", "substrate_evlog", run_id)

    Because pg_notify is transactional in PostgreSQL, notifications are delivered only when the transaction successfully commits. If a rollback occurs, no spurious notifications are fired.

  2. Dedicated Connection for LISTEN: LISTEN requires a persistent, long-lived connection. The EventLog allocates a dedicated asyncpg connection exclusively for notifications, ensuring client connection pools are never blocked or starved by streaming consumers.

  3. In-Memory Coroutine Waking: When a notification packet lands, _on_notify() wakes in-memory asyncio.Event waiters for that specific run_id, allowing real-time tailing with near-zero latency.

  4. Safety Fallback Polling (Zero Missed Events): Because PostgreSQL notifications are delivered best-effort over TCP without message retention, a 2.0-second safety poll backstop runs concurrently. If a temporary network blip occurs, the listener self-heals and drains missing records automatically.

python
from substrate.infrastructure.runtime.event_log import EventLog

event_log = EventLog(pool, dsn="postgresql://...")

# Stream agent execution events in real time
async for entry in event_log.tail(run_id, from_seq=0):
    print(f"[{entry.seq}] {entry.kind}: {entry.payload}")

3. Hot Waking & The Signal Bus

When an agent enters hibernation waiting for human approval or an external webhook, it suspends execution and yields its worker thread.

To wake a suspended agent without active polling:

  1. An approval action or webhook calls SignalBus.signal(run_id, name, payload).
  2. The signal is stored in substrate_signals.
  3. The run status in substrate_run_queue is transitioned from 'suspended' to 'pending'.
  4. PostgreSQL sends a notification:
    python
    await conn.execute("SELECT pg_notify('substrate_sched', $1)", run_id)
  5. Background schedulers wake up instantly and lease the run to the next available worker.

4. Pod Eviction & Crash Replay

If a worker node crashes mid-execution (e.g. at Step 3 of a 5-step task):

  1. A standby worker leases the run from substrate_run_queue.
  2. The worker replays the event log from seq = 0 to seq = 3.
  3. Memoized Effect Hits: Completed tool calls and model responses are loaded from the database cache in under 5ms. The agent does NOT re-query external APIs, duplicate database writes, or double-bill model tokens.
  4. Execution seamlessly resumes at Step 4.

Next Steps

Was this page helpful?
Suggest editsRaise issue