AbstractRuntime 0.4.3

AbstractRuntime: a durable graph runner designed to pair with AbstractCore.

AbstractRuntime

AbstractRuntime is a durable workflow runtime (interrupt → checkpoint → resume) with an append-only execution ledger.

It is designed for long-running workflows that must survive restarts and explicitly model blocking (human input, timers, external events, subworkflows) without keeping Python stacks alive.

Version: 0.4.3 (see pyproject.toml) • Python: 3.10+

Status: pre-1.0 (API may evolve). For production use, pin versions and follow CHANGELOG.md.

AbstractFramework ecosystem

AbstractRuntime is one component of the wider AbstractFramework ecosystem:

• AbstractRuntime (this repo) — durable workflow kernel (src/abstractruntime/core/*)
• AbstractCore — LLM + tools integration (wired via src/abstractruntime/integrations/abstractcore/*)
  Repo: lpalbou/abstractcore

At a high level, hosts define workflow graphs (WorkflowSpec) and AbstractRuntime executes them durably. When nodes request LLM/tool work (EffectType.LLM_CALL, EffectType.TOOL_CALLS), those effects are typically handled via AbstractCore.

flowchart LR
  Host["Host app / orchestrator"] -->|"WorkflowSpec"| RT["AbstractRuntime"]
  RT -->|"LLM_CALL / TOOL_CALLS"| AC["AbstractCore"]
  AC -->|"results / waits"| RT

Install

Core runtime:

pip install abstractruntime

AbstractCore integration (LLM + tools):

pip install "abstractruntime[abstractcore]"

The abstractcore extra installs AbstractCore 2.13.5 or newer so the hardened server auth model, provider-key header routing, prompt-cache control plane, and current tool catalog are available.

MCP worker entrypoint (default toolsets over stdio):

pip install "abstractruntime[mcp-worker]"

Quick start (pause + resume)

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"

What’s included (v0.4.3)

Kernel (dependency-light):

• workflow graphs: WorkflowSpec (src/abstractruntime/core/spec.py)
• durable execution: Runtime.start/tick/resume (src/abstractruntime/core/runtime.py)
• durable waits/events: WAIT_EVENT, WAIT_UNTIL, ASK_USER, EMIT_EVENT
• append-only ledger (StepRecord) + node traces (vars["_runtime"]["node_traces"])
• retries/idempotency hooks: src/abstractruntime/core/policy.py
• runtime-aware limits (_limits) with a default iteration budget of 50 (docs/limits.md)

Durability + storage:

• stores: in-memory, JSON/JSONL, SQLite (src/abstractruntime/storage/*)
• durable command inbox primitives (idempotent, append-only): CommandStore, CommandCursorStore (src/abstractruntime/storage/commands.py, src/abstractruntime/storage/sqlite.py)
• artifacts + offloading (store large payloads by reference)
• snapshots/bookmarks (docs/snapshots.md)
• tamper-evident hash-chained ledger (docs/provenance.md)

Drivers + distribution:

• scheduler: create_scheduled_runtime() (src/abstractruntime/scheduler/*)
• VisualFlow compiler + WorkflowBundles (src/abstractruntime/visualflow_compiler/*, src/abstractruntime/workflow_bundle/*)
• VisualFlow multi-entry execution lowering for fan-in routes and per-entry input overrides (docs/workflow-bundles.md)
• run history export: export_run_history_bundle(...) (src/abstractruntime/history_bundle.py)

Optional integrations:

• AbstractCore (LLM + tools, prompt cache, tool approval waits): docs/integrations/abstractcore.md
• comms toolset gating (email/WhatsApp/Telegram): docs/tools-comms.md

Built-in scheduler (zero-config)

from abstractruntime import create_scheduled_runtime

sr = create_scheduled_runtime()
run_id, state = sr.run(my_workflow)

if state.status.value == "waiting":
    state = sr.respond(run_id, {"text": "yes"})

sr.stop()

For persistent storage:

from abstractruntime import create_scheduled_runtime, JsonFileRunStore, JsonlLedgerStore

sr = create_scheduled_runtime(
    run_store=JsonFileRunStore("./data"),
    ledger_store=JsonlLedgerStore("./data"),
)

Documentation

Document	Description
Getting Started	Install + first durable workflow
API Reference	Public API surface (imports + pointers)
Docs Index	Full docs map (guides + reference)
FAQ	Common questions and gotchas
Architecture	Component map + diagrams
Overview	Design goals, core concepts, and scope
Integrations	Integration guides (AbstractCore)
Snapshots	Named checkpoints for run state
Provenance	Tamper-evident ledger documentation
Evidence	Artifact-backed evidence capture for web/command tools
Limits	_limits namespace and RuntimeConfig
WorkflowBundles	.flow bundle format (VisualFlow distribution)
MCP Worker	abstractruntime-mcp-worker CLI
Changelog	Release notes
Contributing	How to build/test and submit changes
Security	Responsible vulnerability reporting
Acknowledgments	Credits
ROADMAP	Prioritized next steps

Development

python -m venv .venv
source .venv/bin/activate
python -m pip install -U pip
python -m pip install -e ".[abstractcore,mcp-worker,test,docs]"
python -m pytest -q

See CONTRIBUTING.md for contribution guidelines and doc conventions.