Runtime & Services Reference

Reference documentation for the production runtime backends in substrate.infrastructure.runtime. All implementations conform strictly to the kernel protocols defined in substrate.kernel.runtime.

EventLog

Append-only PostgreSQL Write-Ahead Log implementing EventLogProtocol. Backed by substrate_event_log with transactional advisory locking and LISTEN / NOTIFY streaming.

from substrate.infrastructure.runtime.event_log import EventLog
from substrate.kernel.runtime.log_entry import RunLogEntry

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

# Append a durable turn with optimistic sequence check
new_seq = await event_log.append(
    run_id=run_id,
    entry=RunLogEntry(kind="turn.completed", payload={"output": "Done"}),
    expected_seq=4
)

# Stream events reactively in real-time (PostgreSQL LISTEN / NOTIFY)
async for entry in event_log.tail(run_id, from_seq=0):
    print(f"[{entry.seq}] {entry.kind}: {entry.payload}")

Key Methods

• append(run_id, entry, expected_seq) — Acquires a pg_advisory_xact_lock(hash(run_id)), validates sequence monotonicity, writes the row, and fires SELECT pg_notify('substrate_evlog', run_id).
• read(run_id, from_seq=0) — Historical async iterator over committed events.
• tail(run_id, from_seq=0) — Real-time reactive stream waking via asyncio.Event callbacks on notification with a 2.0s safety poll fallback.

SignalBus

Durable event bus implementing SignalBusProtocol. Controls asynchronous signals, human approvals, and schedule timers for suspended runs.

from substrate.infrastructure.runtime.signal_bus import SignalBus

signal_bus = SignalBus(pool)

# Send an asynchronous signal or human approval to a suspended agent
await signal_bus.signal(
    run_id=run_id,
    name="human_approval",
    payload={"approved": True, "reviewer": "alice@company.com"}
)

# Idempotently consume a signal during replay
result = await signal_bus.consume(
    run_id=run_id,
    name="human_approval",
    effect_id="eff_01j9a8b"
)

Key Methods

• signal(run_id, name, payload) — Inserts into substrate_signals, sets run queue status from 'suspended' to 'pending', and wakes workers via SELECT pg_notify('substrate_sched', run_id).
• consume(run_id, name, effect_id) — Idempotent claim: replays of the same effect return the exact same payload.
• timer(run_id, at) — Schedules a wall-clock wakeup deadline without holding active worker threads.

Inbox

Point-to-point transactional mailbox implementing InboxProtocol. Handles agent-to-agent message queues with deduplication and delivery acknowledgments.

from substrate.infrastructure.runtime.inbox import Inbox

inbox = Inbox(pool)

# Deliver a message to an actor's mailbox
delivered = await inbox.deliver(agent_id=target_actor, msg=message)

# Drain pending messages
messages = await inbox.drain(agent_id=target_actor, max=100)

# Acknowledge processed message
await inbox.ack(agent_id=target_actor, msg_id=message.id)

Key Methods

• deliver(agent_id, msg, notify=True) — Inserts with ON CONFLICT (agent_id, msg_id) DO NOTHING for idempotent delivery.
• drain(agent_id, max=100) — Fetches queued messages in FIFO order.
• ack(agent_id, msg_id) — Deletes successfully processed messages.
• nack(agent_id, msg_id) — Increments retry count; moves to dead-letter queue when retries exceed threshold.

Scheduler

Distributed task scheduler implementing SchedulerProtocol. Allocates work across distributed worker pools using PostgreSQL concurrency primitives.

from substrate.infrastructure.runtime.scheduler import Scheduler

scheduler = Scheduler(pool)

# Lease work for an active worker pod
leases = await scheduler.lease(worker_id="worker_pod_9a8f", capacity=4)
for lease in leases:
    print(f"Executing run {lease.run_id} for {lease.agent_id}")

# Release completed or suspended lease
await scheduler.release(lease, status="completed")

Key Methods

• lease(worker_id, capacity) — Claims pending runs using SELECT ... FOR UPDATE SKIP LOCKED with a 30s heartbeat TTL.
• release(lease, status, wake_on=None) — Transitions lease state to 'completed', 'failed', or 'suspended'.
• reclaim_orphans() — Automatically requeues runs whose leases expired due to node eviction or OOM crashes.