The Runtime (L1)

What this is

The kernel page on runtime contracts describes a shape — a handful of pure Protocols (EventLogProtocol, Journal, InboxProtocol, SchedulerProtocol, SupervisorProtocol, SignalBusProtocol, FollowGraph, FanoutStrategy) with no I/O, no sockets, no database. This page describes the first real machine that fills in that shape.

It is the Stage-0, in-process implementation: everything runs inside one Python process, on one asyncio event loop, with no serialization — a leased run is just an asyncio.Task that stays alive in memory. There is no Postgres, no Redis, no network. You can pip-free, docker-free run a durable agent loop on your laptop with nothing but python.

The one idea that makes it all worth it

The agent author writes async def run(self, ctx, inbox) once. Every ctx.llm(), ctx.tool(), ctx.spawn() call is journaled under the hood. That same agent code runs unchanged against the Postgres/Redis backends in infrastructure/runtime/ — you swap the backends, not the agent. Dev and production are literally the same call sites.

# Dev (Stage 0) — everything in-process, zero infra
async with Runtime() as rt:
    await rt.register(my_agent)
    run_id = await rt.submit(my_agent.id, boot_msg)

# Production (Stage 1) — durable backends injected by the infra factory
from substrate.infrastructure.runtime import build_postgres_runtime
async with build_postgres_runtime(postgres_url=..., redis_url=...) as rt:
    await rt.register(my_agent)
    run_id = await rt.submit(my_agent.id, boot_msg)   # identical

Here is the cast this page covers, each with a one-line analogy.

The Runtime facade

What & why: Runtime is one object you create, hand a few agents, and submit messages to. It owns the eight backends (defaulting each to its in-memory implementation), wires the inbox wakeup hook, and starts a single Worker. Think of it as the building manager: it connects the power, hangs the mailboxes, hires the one driver, and gives you a front desk to drop off work.

It lives at agents/runtime/runtime.py and is exported as Runtime.

Construction — inject or default

Every backend is an optional constructor argument. Pass nothing and you get the full in-memory stack. Pass durable ones and the same Runtime runs against Postgres/Redis — this is exactly what build_postgres_runtime does.

class Runtime:
    def __init__(
        self,
        *,
        event_log: EventLogProtocol | None = None,
        inbox: InboxProtocol | None = None,
        journal: Journal | None = None,
        scheduler: SchedulerProtocol | None = None,
        signal_bus: SignalBusProtocol | None = None,
        follow_graph: FollowGraph | None = None,
        fanout: FanoutStrategy | None = None,
    ) -> None:
        self._event_log = event_log or InMemoryEventLog()
        self._journal   = journal   or InMemoryJournal()
        self._scheduler = scheduler or InMemoryScheduler()
        self._inbox     = inbox     or InMemoryInbox()
        self._inbox.set_deliver_hook(self._on_inbox_deliver)   # the wakeup wire
        ...

The public API

Four verbs, plus lifecycle. All return fast — work happens on the Worker.

submit and the duplicate-run trap

submit does the delivery and the enqueue itself. The subtle bit is notify=False:

async def submit(self, agent_id, msg, *, priority=5, tenant="default", max_retries=3):
    run_id = new_run_id()
    self._scheduler.register_run(run_id, agent_id)
    # notify=False: submit() enqueues its OWN run below. If the deliver-hook
    # also fired, it would find no active run yet and spawn a DUPLICATE.
    await self._inbox.deliver(agent_id, msg, notify=False)
    await self._scheduler.enqueue(run_id, priority=priority, tenant=tenant,
                                  retry_policy=RunRetryPolicy(max_retries=max_retries))
    return run_id

The inbox deliver-hook: wake vs spawn

When a message lands in a dormant agent's mailbox via the normal path (notify=True), the Inbox calls a sync hook, which schedules _handle_inbox_delivery. That method makes the central wake-vs-spawn decision using scheduler.find_run_for_agent() — the same API on both the in-memory and Postgres schedulers, so no private attributes are touched.

async def _handle_inbox_delivery(self, agent_id: AgentId) -> None:
    result = await self._scheduler.find_run_for_agent(agent_id)
    if result is None:
        await self._spawn_run_for_inbox(agent_id)   # no active run -> spawn fresh
        return
    run_id, status = result
    if status == RunStatus.SUSPENDED:
        await self._scheduler.wake_suspended(run_id)  # dormant run -> wake it
    # PENDING / RUNNING -> no-op: the active run will drain this message itself

Start / stop as a context manager

start() constructs the InMemorySupervisor (it needs references to the EventLogProtocol, InboxProtocol, Journal, and SchedulerProtocol) and the single Worker, then starts the Worker's poll loop. async with calls start/stop for you.

async with Runtime() as rt:        # -> start() : builds SupervisorProtocol + Worker, starts poll loop
    await rt.register(agent)
    await rt.submit(agent.id, msg)
    ...                            # Worker drives runs in the background
# -> stop() : cancels the poll loop and all in-flight agent tasks

The Worker — the run loop

What & why: The Worker is the engine that turns a queued run into a running agent. It polls the Scheduler for leased runs, builds a RunContext, drains the inbox, calls agent.run(...), and — win, lose, or cancel — files the paperwork (append a terminal log entry, ack/nack messages, release the lease).

It lives at agents/runtime/worker.py.

The poll loop

_poll_loop runs every POLL_INTERVAL (0.05 s). Each tick it leases up to capacity=10 runs and launches each as its own asyncio.Task. Many agents run concurrently because each is an independent task; the lease capacity bounds how many start per tick.

async def _poll_loop(self) -> None:
    while self._running:
        leases = await self._scheduler.lease(worker_id=self._worker_id, capacity=10)
        for lease in leases:
            agent = self._registry.get(lease.agent_id)
            if agent is None:
                continue   # not registered yet — hold the lease, it expires + retries
            task = asyncio.create_task(self._run_agent(lease, agent))
            self._tasks[lease.run_id] = task
        await asyncio.sleep(self.POLL_INTERVAL)

One run, start to finish (_run_agent)

This is the heart of the page. For one leased run, the Worker:

1. Builds a fresh RunContext — wiring in all backends plus the agent's LLM client and a ToolInvoker built from the agent's declared tools (_build_tool_invoker).
2. Appends run.started to the EventLog if last_seq < 0 (fresh run).
3. Drains the inbox for this agent (up to 100 messages).
4. Dispatches lifecycle hooks (RUN_START) and runs through the agent's middleware pipeline if present, otherwise calls agent.run directly.
5. Heartbeats in the background every 15 s so a long LLM call never loses a Postgres lease (the in-memory scheduler's heartbeat is a no-op).
6. On success — ack every drained message, optionally clear run-scoped history (_maybe_clear_run_history), append run.completed, release with COMPLETED, and record completion with the Supervisor (this is what wakes a parent waiting in join).
7. On cancel (asyncio.CancelledError / CancellationError) — append run.cancelled, release CANCELLED.
8. On any other exception — branch: a guardrail stop (MiddlewareTermination) or budget stop (BudgetExhaustedError) is expected, so it acks the messages and logs run.failed with a friendly status; a real AgentCrashError nacks the messages (so the run can be retried) and logs the crash.

async def _run_agent(self, lease, agent):
    ctx = RunContext(meta=meta, event_log=..., journal=..., inbox=..., ...,
                     llm_client=getattr(agent, "model", None),
                     tool_invoker=self._build_tool_invoker(agent), agent=agent)
    if (await self._event_log.last_seq(run_id)) < 0:
        await self._event_log.append(run_id, RunLogEntry(..., kind="run.started"),
                                     expected_seq=-1)
    inbox_msgs = await self._inbox.drain(agent.id, max=100)
    heartbeat_task = asyncio.create_task(_heartbeat())   # renews lease every 15s
    try:
        if middleware is not None:
            await middleware.execute(ctx, lambda c: agent.run(c, inbox_msgs))
        else:
            await agent.run(ctx, inbox_msgs)
        for msg in inbox_msgs:
            await self._inbox.ack(agent.id, msg.id)
        await self._event_log.append(run_id, RunLogEntry(..., kind="run.completed"), ...)
        await self._scheduler.release(lease, status=RunStatus.COMPLETED)
        await self._supervisor.record_completion(run_id, RunStatus.COMPLETED)
    except (asyncio.CancelledError, CancellationError):
        ...  # append run.cancelled, release CANCELLED
    except Exception as exc:
        ...  # guardrail/budget -> ack + run.failed ; crash -> nack + run.failed
    finally:
        heartbeat_task.cancel()
        ...  # dispatch RUN_END hook, drop run from _tasks / _tokens

RunContext — the agent's toolkit and notebook

What & why: RunContext is the concrete ctx passed to agent.run(ctx, inbox). It satisfies the kernel's minimal AgentRunContext Protocol but offers the full rich surface an agent author actually uses: ctx.llm(), ctx.tool(), ctx.spawn(), ctx.join(), ctx.ask(), ctx.send(), ctx.emit(), ctx.check(), and more. Every effectful call is journaled, so a replay never re-bills the model or re-sends an email.

It lives at agents/runtime/context/ (a small package: journal.py, llm.py, tool.py, messaging.py, supervision.py — not a single file).

Hierarchical effect paths and the EffectCache

At-most-once used to be keyed by a flat _step_seq counter looked up in a separate Journal backend. That's been replaced: the context now keeps a scope stack (_path_stack) rather than a flat counter, and effect lookups/records go through an EffectCache (agents/runtime/effect_cache.py) folded straight from the EventLog's own effect.result entries — not a separate Journal store. _alloc_path() allocates a hierarchical path ("0", "0.1", "1", …) so a tool's own internal journaled calls (e.g. a nested ctx.uuid()) get stable paths of their own without desyncing sibling effect ids, even when the outer call itself replays as a cache hit. The generic helper is _journaled():

async def _journaled(self, kind, args, fn):
    path = self._alloc_path()
    effect_id = Effect.make_id(self.run_id, path, kind, args)
    cached = self._lookup_effect(effect_id)     # EffectCache — pure in-memory, no I/O
    if cached:
        if cached.status == "error":
            raise RuntimeError((await self._resolve_effect_value(cached)).get("error", "journaled error"))
        return await self._resolve_effect_value(cached)   # HIT — do NOT re-run
    self._enter_scope()                          # open a child path scope for fn()'s own calls
    try:
        result = await fn()                       # MISS — run the real effect
        await self._record_effect(effect_id, "ok", result or {})
        return result
    except Exception as exc:
        await self._record_effect(effect_id, "error", {"error": str(exc)})
        raise
    finally:
        self._exit_scope()

This is the same lookup -> execute -> record dance described in Durability, just backed by the EffectCache/EventLogProtocol rather than a Journal. ctx.llm() and ctx.tool() inline the same pattern (so they can stream and serialize their own results) rather than calling _journaled directly, but the contract is identical. Values over 64KB are transparently offloaded to BlobStore and referenced by artifact_ref, dereferenced lazily by _resolve_effect_value() only on a genuine cross-process replay hit.

The capability surface

How ask and sleep actually suspend (Stage 0)

In Stage 0 the coroutine is never serialized — it stays alive as an asyncio Task. So ctx.ask() and ctx.sleep_until_signal() suspend by awaiting an asyncio.Event inside the InMemorySignalBus, the lightweight in-process stand-in for the durable Wakeup mechanism. ask sets reply_to and correlation_id, delivers to the target, then waits on a reply:{id} signal:

async def ask(self, target, msg, *, timeout, idempotency_key=None):
    enriched = msg.model_copy(update={"reply_to": self.run_id, "correlation_id": cid})
    await self._inbox.deliver(target_agent, enriched)
    await self._scheduler.wake_agent(target_agent)        # or wake_suspended(run)
    try:
        payload = await self._signal_bus.wait_for_signal(self.run_id, f"reply:{cid}",
                                                         timeout=timeout)
        return AskOutcome(kind="replied", result=..., last_seq=0)
    except asyncio.TimeoutError:
        # peek target status -> timed_out vs target_failed vs target_cancelled
        return AskOutcome(kind=kind, handle=handle, last_seq=last_seq)

The in-memory backends

Each backend is a single-process, single-event-loop implementation of one kernel Protocol. None is crash-durable — that is the whole point of Stage 0: zero infra, fast tests, identical surface. They all live in agents/runtime/backends/.

A one-line tour:

• InMemoryEventLog — a dict[run_id, list[RunLogEntry]]. append enforces optimistic concurrency (raises ConcurrentAppendError when expected_seq doesn't match the real tail), read yields a finite slice, tail yields then blocks on an asyncio.Event for new entries (this powers live SSE streaming).
• InMemoryInbox — per-agent message store with dedup by Message.id, per-sender FIFO ordering, and a retry counter that dead-letters after max_retries nacks. Holds the on_deliver wakeup hook the Runtime wires in.
• InMemoryScheduler — an asyncio.PriorityQueue (lower number = higher priority) plus status/lease/agent maps. enqueue coalesces (a run already pending is not added twice), lease hands out leases with a 30 s expiry, heartbeat is a no-op (single process), and release re-enqueues FAILED runs per their RunRetryPolicy. It owns find_run_for_agent — the wake-vs-spawn decider.
• InMemorySupervisor — spawn journals the child run id via the EffectCache (so replay returns the same child and never duplicates it), delivers the boot message with notify=False, and logs child.spawned in the parent's EventLog. join suspends on a consume-based child:{run_id} signal (via SignalBusProtocol) rather than blocking on an asyncio.Event; cancel recurses through the subtree.
• InMemorySignalBus — run_id -> name -> (asyncio.Event, payload_box). signal fired before a run waits is buffered (delivered eagerly), so a signal is never lost. timer schedules an asyncio.sleep that fires the __timer__ signal. Exposes the non-Protocol wait_for_signal helper used by ask and sleep_until_signal.
• PushAllFanout — publish simply async for follower in graph.followers_of(topic): inbox.deliver(follower, msg). Fine for normal agents; a pull model for celebrity agents is unbuilt.
• InMemoryFollowGraph — two defaultdicts (followers_of and following) keyed by "{type}/{source}"; follow is idempotent (a set), unfollow is safe on an absent subscription.

Where this lives

Next: Context, Compaction & Memory Backends — how an agent's conversation history is stored, trimmed, and summarised so a long run fits in the model's context window.