AbstractRuntime
AbstractRuntime is a durable workflow runtime: interrupt, checkpoint, resume, with an append-only execution ledger (the record of every step a run takes). Workflows survive restarts and wait for people, timers, events or child runs without keeping a Python stack alive. It also runs automations on a schedule or on request, and replays session history as a window of whole turns.
pip install abstractruntime # remote providers, no local engines
pip install "abstractruntime[apple]" # local engines on Apple silicon; [gpu] for NVIDIAA persistent graph runner
You define workflow graphs (WorkflowSpec). Nodes never run side effects themselves: they return a StepPlan that requests an effect (a model call, tool calls, a question to the user, a wait), and the runtime executes it, checkpoints the run and appends a record to the ledger. When a node requests model or tool work, AbstractCore handles it.
Pause, then resume from any process
A blocked run persists a WaitState. resume() continues at the next node, so the waiting node is not run again.
- ✓ Durable waits:
ASK_USER,WAIT_UNTIL,WAIT_EVENT,EMIT_EVENT, subworkflows - ✓ Stores: in-memory, JSON/JSONL and SQLite, a durable command inbox, artifacts and offloading, snapshots
- ✓ Provenance, opt-in: wrap a ledger store in
HashChainedLedgerStorefor a tamper-evident chain (verify_ledger_chain()). Entity diaries use it; the gateway’s run ledgers do not - ✓ Limits and ids: a default budget of 20 iterations; a run id you choose is created only when no run already uses it
from abstractruntime import Effect, EffectType, Runtime, StepPlan, WorkflowSpec
from abstractruntime.storage import InMemoryLedgerStore, InMemoryRunStore
def ask(run, ctx):
return StepPlan(
node_id="ask",
effect=Effect(
type=EffectType.ASK_USER,
payload={"prompt": "Continue?"},
result_key="user_answer",
),
next_node="done",
)
def done(run, ctx):
answer = run.vars.get("user_answer") or {}
text = answer.get("text") if isinstance(answer, dict) else None
return StepPlan(node_id="done", complete_output={"answer": text})
wf = WorkflowSpec(workflow_id="demo", entry_node="ask", nodes={"ask": ask, "done": done})
rt = Runtime(run_store=InMemoryRunStore(), ledger_store=InMemoryLedgerStore())
run_id = rt.start(workflow=wf)
state = rt.tick(workflow=wf, run_id=run_id)
assert state.status.value == "waiting"
state = rt.resume(
workflow=wf,
run_id=run_id,
wait_key=state.waiting.wait_key,
payload={"text": "yes"},
)
assert state.status.value == "completed"
print(state.status.value, state.output) # completed {'answer': 'yes'}What the runtime does
A small, import-light kernel, plus integrations for models, tools, memory and distribution.
Stop reaches the running call
Runtime.cancel_run(...) stops the model or tool call in flight: it ends as a cancelled ledger record and is never retried. Hosts can also pause at step boundaries with tick(..., step_gate=...).
Scheduled and on-request runs
A workflow runs on schedule@1 (fixed UTC intervals) or manual@1 (only when asked) as a durable controller run. Each firing is an occurrence: a child run with a deterministic id that counts as one session turn.
A whole-turn history window
Replayed session history keeps the newest whole messages up to 50,000 estimated tokens. No message is ever cut, and the run records what was replayed and dropped.
Live token streaming
Runtime.set_live_delta_sink(sink) plus _runtime.stream: true delivers llm.delta and llm.delta_end events while an answer is generated, outside the ledger. The durable record still holds the final answer.
Tool approval and a tool ceiling
An explicit allowed_tools list can only narrow across child runs, and approval policy never widens it. ApprovalToolExecutor runs safe tool batches at once and asks for approval before the others.
VisualFlow and bundles
The VisualFlow compiler runs the workflows you draw in AbstractFlow, packed as .flow workflow bundles (a portable package of a workflow graph, its subflows and metadata), with multi-entry execution. export_run_history_bundle(...) exports a portable replay of a run.
AbstractCore integration
LLM_CALL, TOOL_CALLS and MODEL_RESIDENCY effects, generated voice, music, image and video outputs, and a config_facade that passes Core's models, engines and host jobs through to hosts.
Memory and entities
MEMORY_KG_* effects backed by AbstractMemory, and an entity runtime with per-entity homes, leases and visit waits.
Workspace tools and comms
File and shell tools scoped to a run's workspace (with host-only protected folders), and opt-in email, WhatsApp and Telegram tools. abstractruntime-mcp-worker exposes AbstractCore toolsets over MCP.
An automation is a run
AbstractRuntime runs automations itself, as ordinary durable runs, so a runtime and a run store are enough. A host such as AbstractGateway exposes them over HTTP.
Controller and occurrences
The controller runs the packaged flow abstractframework.automation-controller@1.0.0; its run id is the automation id. At most one occurrence runs at a time, and a crash never loses an occurrence or starts one twice.
Commands and context
Pause, resume, run now, revise, stop the current occurrence, or archive the automation. Each occurrence starts fresh by default, or joins a growing session that sees the previous occurrences as history.
Quiet by default
An occurrence asks for attention only when its output says notify, when it still fails after its last retry, or while it waits on a person. Fork a discussion at any occurrence, in its own workspace.
The history window
Session replay keeps the most recent turns that fit HISTORY_REPLAY_MAX_TOKENS (50,000 estimated tokens), newest first, as whole messages. There is no message-count cap and no character cap. When the newest turn alone is larger than the window, it is kept whole and the report says oversize_turn_kept. Images count as 512 tokens each.
- ✓ Applies to growing automations, discussion seeds, host session seeding and the entity chat and visit drivers
- ✓ When older turns are dropped, the oldest kept message starts with a labeled
[#TRUNCATION: ...]line - ✓ Budgets are met by selecting whole turns, never by cutting content, as the framework’s truncation policy requires (selection, never truncation: ADR-0026; see the runtime CHANGELOG 0.7.0 and API reference)
from abstractruntime import HISTORY_REPLAY_MAX_TOKENS, window_transcript
messages = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello"},
]
history = window_transcript(messages) # max_tokens=50_000 by default
print(HISTORY_REPLAY_MAX_TOKENS) # 50000
print(history.report["policy"]) # most_recent_whole_turns
print(history.report["dropped_messages"])More ways in
A zero-config scheduler, a runtime wired to AbstractCore, approval-gated tools and live streaming.
from abstractruntime import Effect, EffectType, Runtime, StepPlan, WorkflowSpec
from abstractruntime.storage import InMemoryLedgerStore, InMemoryRunStore
def ask(run, ctx):
return StepPlan(
node_id="ask",
effect=Effect(
type=EffectType.ASK_USER,
payload={"prompt": "Continue?"},
result_key="user_answer",
),
next_node="done",
)
def done(run, ctx):
answer = run.vars.get("user_answer") or {}
text = answer.get("text") if isinstance(answer, dict) else None
return StepPlan(node_id="done", complete_output={"answer": text})
my_workflow = WorkflowSpec(workflow_id="demo", entry_node="ask", nodes={"ask": ask, "done": done})
from abstractruntime import create_scheduled_runtime, JsonFileRunStore, JsonlLedgerStore
sr = create_scheduled_runtime(
run_store=JsonFileRunStore("./data"),
ledger_store=JsonlLedgerStore("./data"),
)
run_id, state = sr.run(my_workflow)
if state.status.value == "waiting":
state = sr.respond(run_id, {"text": "yes"})
sr.stop()
print(state.status.value, state.output)from abstractruntime.integrations.abstractcore import create_local_runtime
# LLM_CALL and TOOL_CALLS effects run through AbstractCore
rt = create_local_runtime(provider="mlx", model="mlx-community/Qwen3.5-4B-4bit")
print(type(rt).__name__, type(rt.run_store).__name__)def write_file(path: str, content: str) -> str: # W2: stub for the name the page assumes
return "ok"
from abstractruntime.integrations.abstractcore import (
ApprovalToolExecutor,
MappingToolExecutor,
ToolApprovalPolicy,
)
tools = ApprovalToolExecutor(
delegate=MappingToolExecutor({"write_file": write_file}),
policy=ToolApprovalPolicy(),
)
print(type(tools).__name__)events = []
runtime.set_live_delta_sink(events.append)
run_id = runtime.start(workflow=wf, vars={"_runtime": {"stream": True}})
runtime.tick(workflow=wf, run_id=run_id)
# events: {"kind": "llm.delta", "text": "Hello", "channel": "content", ...}
# {"kind": "llm.delta_end", "reason": "completed", ...}Checked against AbstractRuntime 0.7.1 (Runtime.set_live_delta_sink); a fragment that needs a runtime and a workflow wf; not executed for this page.