Runtime

indusagi.runtime is the engine that drives one model conversation from a prompt to settlement. Imported as import indusagi.runtime (or from indusagi import runtime), it pairs a pure finite-state reducer with a single stateful conductor that carries out the side effects the reducer asks for.

The runtime is split in two. The pure core is an immutable RunSnapshot advanced by a side-effect-free reducer (cadence) that returns Effects and never touches I/O. The thin stateful conductor (Agent / create_agent) interprets those effects — reach the model, run tools, condense history, persist, publish — and folds the resulting Signals back into the reducer until the run settles or faults. It owns the full agent loop: streamed emissions fold into an assistant turn, requested tools are dispatched with bounded concurrency, history is condensed when it nears the model's window, settled history is persisted to a content-addressed session DAG, and progress is published as RunEvents.

Table of Contents

• Quickstart
• Public API
• Sub-directories
• Effect / Signal loop
• Driving the pure FSM directly
• Events and snapshots
• Tool dispatch
• Compaction
• Sessions and persistence
• Key concepts
• Notable behavior
• Relationship to neighbors

Quickstart

create_agent builds a runnable Agent from a static AgentConfig. submit drives one prompt to settlement; the model entrypoint is injectable, so a test can script it without a network. A Channel is just an async iterable of Emissions.

import asyncio
from indusagi.runtime import create_agent, AgentConfig, AgentDeps
from indusagi.llmgateway.contract import (
    Conversation, StreamOptions, TextEmission, DoneEmission,
    AssistantTurn, TextBlock, Usage,
)


async def scripted(conversation: Conversation, options: StreamOptions):
    yield TextEmission(delta="Hello, ")
    yield TextEmission(delta="world!")
    reply = AssistantTurn(blocks=(TextBlock(text="Hello, world!"),))
    yield DoneEmission(reply=reply, usage=Usage(input_tokens=10, output_tokens=3))


def invoke(conversation, options):
    return scripted(conversation, options)


async def main() -> None:
    agent = create_agent(
        AgentConfig(model="some-model"),
        AgentDeps(invoke_model=invoke),
    )
    final = await agent.submit("hi")   # string sugar -> a single user text turn
    print(final.phase)                 # 'settled'
    print(final.messages[-1])          # the folded AssistantTurn


asyncio.run(main())

create_agent(config, deps=None) accepts an AgentConfig — model (a catalog id) is required; system, tools, max_output_tokens, thinking, compaction, and max_turns are optional. submit takes a bare string (sugar for one user turn) or a sequence of Turns and returns the terminal RunSnapshot.

Public API

The package barrel (indusagi.runtime) exposes the host-facing factory, its handle, the type vocabulary callers branch on, and the pure transition machine for advanced hosts.

Effects, signals, and the lower-layer drivers live on the sub-packages below.

Sub-directories

Internal layering is strict: contract (types) ← cadence (pure reducer) ← turn / dispatch / memory / store / ledger / wire (single-purpose drivers) ← conductor (the only orchestrator that touches all of them).

Effect / Signal loop

The reducer requests work as effects; the conductor performs each and feeds results back as signals, closing the loop until a terminal phase. The agent loop is encoded as a transition table keyed on (phase, signal.kind), not nested control flow:

submit                         → invoke_model              · phase invoking
emission (text/thinking)       → fold + publish delta       · phase streaming
emission (tool start/delta)    → fold + tool_started        · phase streaming
emission (usage)               → accumulate usage           · (phase kept)
stream_end · has tool calls    → run_tool × N               · phase dispatching
stream_end · no tool calls     → persist + settled          · phase settled
tool_settled · more pending    → mark done, keep waiting    · phase dispatching
tool_settled · all done        → append tool turn,          · phase invoking
                                  invoke_model (next turn)
compacted                      → splice summary, invoke     · phase invoking
abort / fault / error          → faulted                    · phase faulted

A submit seeds a blank AssistantTurn and emits an InvokeModelEffect (phase → invoking). Streamed EmissionSignals fold into the open assistant turn (text/thinking deltas merge, tool-call blocks open, raw JSON argument fragments concatenate; usage accumulates) and publish TextDeltaEvent/ThinkingDeltaEvent/ToolStartedEvent. StreamEndSignal seals the buffered tool args, then either settles (no tool calls → PersistEffect + SettledEvent) or dispatches (one RunToolEffect per call, phase → dispatching). ToolSettledSignals arrive in any order; once every pending tool is 'done' the reducer re-invokes the model.

The conductor's _drive() is the heartbeat: pull a signal from a future-backed _SignalQueue, step the reducer, swap in the new snapshot, and perform the effects. A run is genuinely finished only when it is terminal AND no effect is in flight AND no signal is buffered — so a steering submit/compacted/fault that lands while a trailing persist/publish is still draining can legitimately reopen the run.

Driving the pure FSM directly

Advanced hosts can bypass the conductor and drive the reducer themselves. cadence(config) returns a step(state, signal) -> Transition; it performs no I/O and never awaits.

from indusagi.runtime import cadence, initial_snapshot, AgentConfig
from indusagi.runtime.contract import SubmitSignal
from indusagi.llmgateway.contract import UserTurn, TextBlock

step = cadence(AgentConfig(model="some-model"))            # also exported as `step`
state = initial_snapshot(session_id="s1", model="some-model")  # phase 'idle'

signal = SubmitSignal(input=(UserTurn(blocks=(TextBlock(text="hi"),)),))
transition = step(state, signal)
print(transition.state.phase)        # 'invoking'
for effect in transition.effects:
    print(type(effect).__name__)     # InvokeModelEffect, PublishEffect

Events and snapshots

subscribe taps the live RunEvent stream and returns a disposer. A SnapshotAccumulator folds the snapshot-bearing events into the single latest RunSnapshot; attach(ledger) wires it to the agent's ledger and returns its own disposer.

from indusagi.runtime import create_agent, AgentConfig
from indusagi.runtime.contract import TextDeltaEvent
from indusagi.runtime.ledger import SnapshotAccumulator

agent = create_agent(AgentConfig(model="some-model"))


def on_event(event):
    if isinstance(event, TextDeltaEvent):
        print(event.delta, end="")


unsubscribe = agent.subscribe(on_event)

acc = SnapshotAccumulator()
acc_dispose = acc.attach(agent._ledger)   # or feed acc.observe(event) from on_event

# ... await agent.submit(...) elsewhere ...
latest = acc.get_snapshot()               # most recent RunSnapshot or None
unsubscribe()
acc_dispose()

The simplest path is to skip the accumulator entirely: agent.submit(...) already resolves the terminal RunSnapshot, and agent.snapshot() reads the latest state synchronously at any time.

Tool dispatch

When AgentConfig.tools is present, the conductor builds a Scheduler over tools.runner. One stream-end round's tool calls run as a single batch through the scheduler at most DEFAULT_CONCURRENCY (8) at a time, each under a child CancelToken chained to the run's active token. ToolSettledSignals are emitted in completion order, not request order, and each appends a ToolResultBlock to the trailing ToolTurn. If a tool is requested but no runner is configured, the round faults with 'tool_failed' rather than hanging.

Compaction

Condensation is not a reducer transition — it happens lazily at the conductor's invocation boundary. Before opening a stream, if should_compact() trips on the live snapshot's history, the older prefix is distilled into one summary turn and a compacted signal redirects the reducer to re-invoke over the shrunken history. The functions are usable standalone:

from indusagi.runtime.memory import estimate_context_tokens, should_compact, find_cut_point
from indusagi.runtime.contract import CompactionPolicy
from indusagi.llmgateway.catalog import get_card
from indusagi.llmgateway.contract import UserTurn, TextBlock

messages = (UserTurn(blocks=(TextBlock(text="x" * 4000),)),)
print(estimate_context_tokens(messages))   # ~ len/4 + 4 per turn

card = get_card("some-model")
if card is not None:
    policy = CompactionPolicy(trigger_ratio=0.8, keep_recent=8)
    print(should_compact(messages, card, policy))
    print(find_cut_point(messages, policy.keep_recent))   # tool-safe boundary index

The summary is filed under the user role (marked with SUMMARY_HEADING, [condensed earlier context]) so it stays valid input for any provider. A pre-check on find_cut_point(...) <= 1 avoids spending a model call when history is already as small as the policy allows, and stops the invocation gate from re-triggering forever.

Sessions and persistence

When an AgentDeps.store is injected, settled history is appended to a content-addressed SessionGraph. Each SessionNode's id is a SHA-256 hash of (parent, turn, created_at); branching is two nodes sharing a parent, and the model's linear context is the root→leaf path. Re-appending an identical turn is a no-op head advance, so re-persisting an unchanged prefix writes nothing new. Persistence is advisory — a store error is swallowed so an otherwise complete run still settles.

import asyncio
from indusagi.runtime.store import SessionStore, SessionGraph
from indusagi.llmgateway.contract import UserTurn, TextBlock


async def main() -> None:
    store = SessionStore("/tmp/indus-sessions")
    graph = SessionGraph("sess-1")
    node = graph.append(UserTurn(blocks=(TextBlock(text="remember this"),)))
    await store.append_node("sess-1", node)

    reloaded = await store.load_session("sess-1")
    leaf = reloaded.leaf
    if leaf is not None:
        print(reloaded.path_to(leaf))         # root->leaf turns
    print(await store.list_sessions())        # ['sess-1']


asyncio.run(main())

An Agent with a store can await agent.resume(session_id) to re-seat its in-memory snapshot on the rehydrated history; the phase returns to 'idle' so the next submit continues the conversation from where it left off.

Key concepts

• Pure reducer / cadence — cadence(config) returns a side-effect-free step(state, signal) -> Transition. It derives a fresh immutable RunSnapshot and a flat tuple of Effects per signal; it never awaits or does I/O.
• Effect / Signal loop — the reducer requests work (invoke_model, run_tool, persist, compact, publish); the conductor performs each and feeds results back as signals (emission, stream_end, tool_settled, compacted, abort, fault).
• RunSnapshot (event-sourced state) — the single immutable source of truth, never mutated in place; every transition produces a new snapshot via dataclasses.replace.
• Content-addressed session DAG — history is an immutable DAG keyed by content hash; the model's context is the root→leaf path.
• Lazy compaction gate — condensation fires at the conductor's invocation boundary, not as a reducer transition.
• Steering submit — a submit() arriving mid-run is folded into the live signal queue rather than starting a competing loop; the caller awaits the shared driving task.
• Turn budget — the drive loop counts InvokeModelEffects and force-faults with 'turn_budget' once they would exceed max_turns (default 64), guarding against a model that loops forever.
• Bounded-concurrency tool dispatch — one round's calls run as a single batch at most DEFAULT_CONCURRENCY (8) at a time, each under a child CancelToken.

Notable behavior

• The compaction gate in _on_invoke_model reads the live self._snapshot.messages, not the conversation payload captured in the effect, so it always sizes the freshest history.
• AbortSignalInput is named that way (not AbortSignal) to avoid shadowing the DOM global.
• RunPhase includes 'compacting', but the reducer never sets it — compaction is handled at the conductor's invocation boundary via compacted signals, so the phase is effectively unused.
• abort() hard-cancels the model-stream tasks in addition to firing the cooperative token, because a connector parked on a network read may not poll the token until its next chunk.
• CancelledError is consistently re-raised and never folded into a fault or swallowed by ledger fan-out.
• The wire/projectors.py projectors are identity passthroughs today (history is already stored in gateway turn shape) — a deliberate seam, not dead code.

Relationship to neighbors

The runtime depends on LLM Gateway for its provider-neutral vocabulary (Turn, Block, Emission, Conversation, StreamOptions, Channel, Usage, ModelCard, ToolDescriptor, ThinkingLevel) and for the stream function plus catalog.get_card used by the default ModelInvoker and the compaction window sizing. It deliberately does not re-export those gateway types — they have a single source of truth in the gateway, imported straight from indusagi.llmgateway or indusagi.llmgateway.contract.

Downstream, the Capabilities tool kernel and the Shell App build a ToolBox/ToolRunner, then call create_agent(...) and submit/subscribe. The high-level agent facade is the ergonomic front door over this engine, and the ai facade wraps the gateway it talks to.

Back to the Architecture overview.