Pragmatic

The quickstart.

Five minutes from cargo add to a crash-proof, replayable agent. Every snippet on this page runs against version 0.4 of the crates.

1 · Install

Two crates. pragmatic is the runtime, with zero dependencies. pragmatic-anthropic gives you Claude as an Oracle; skip it and implement the Oracle trait over your own model client if you prefer.

Cargo.toml

[dependencies]
pragmatic = "0.4"
pragmatic-anthropic = "0.4"   # or implement Oracle for your own client
2 · Write the agent

An agent is a plain function over Ctx. Route model calls through ctx.oracle so they are journaled, route external writes through ctx.effect so they run exactly once, and mark the function #[pragmatic::durable].

src/main.rs

use pragmatic::{Ctx, Fault, Value};

#[pragmatic::durable]
fn research(ctx: &mut Ctx) -> Result<Value, Fault> {
    let plan = ctx.oracle("plan the research task")?;   // journaled

    let mut findings = Vec::new();
    for step in 0..10 {
        findings.push(ctx.oracle(format!("probe {step}: {plan}"))?);
    }

    // A durable effect: journaled before it runs (write-ahead),
    // its result journaled after. Replay reuses the result;
    // the world is touched exactly once.
    ctx.effect("publish", format!("{} findings", findings.len()), |arg| {
        // your real side effect: S3 write, webhook, DB insert...
        Ok(Value::from(format!("s3://reports/{arg}")))
    })
}
3 · Run it durably

src/main.rs

use pragmatic::Runtime;
use pragmatic_anthropic::AnthropicOracle;

let oracle = AnthropicOracle::from_env()?     // reads ANTHROPIC_API_KEY
    .model("claude-sonnet-5")
    .max_tokens(1024);

let mut rt = Runtime::on_dir("./journals", oracle)?;
let report = rt.run("research-42", research)?;
println!("{}", report.output);

Every oracle call's realized outcome now lands in ./journals/research-42.journal: append-only, hash-chained, synced to disk as the run proceeds.

4 · Crash, recover

Kill the process anywhere, say step nine of a thousand. Then:

src/main.rs

let report = rt.resume("research-42", research)?;

The nine recorded steps are read back from the journal, with zero model calls and zero duplicate effects; recording continues at step nine. The agent function cannot tell the difference.

5 · Replay, exactly

src/main.rs

let audit = rt.replay("research-42", research)?;   // the model is NEVER called
assert_eq!(audit.trace, report.trace);

Strict replay reproduces the run bit for bit, nondeterministic model behavior included, for time-travel debugging and audit. During replay the oracle is swapped for one that refuses to sample, so "the model is never called" is enforced, not promised.

6 · Inspect

Install the CLI once, and every journal directory becomes inspectable from the terminal or the browser.

your terminal

cargo install pragmatic-cli

pragmatic runs   --dir ./journals            # every run, chain status
pragmatic show   research-42 --dir ./journals
pragmatic verify --all --dir ./journals      # walk the tamper-evident chain
pragmatic export research-42 --dir ./journals -o run.html
pragmatic serve  --dir ./journals            # live console on 127.0.0.1:7171
Async agents

Same API, awaited. Executor-agnostic and still zero-dependency: drive it from tokio, smol, or the built-in pragmatic::block_on.

src/main.rs

use pragmatic::{AsyncCtx, AsyncRuntime, Fault, Value};

#[pragmatic::durable]
async fn research(ctx: &mut AsyncCtx<'_, MyOracle>) -> Result<Value, Fault> {
    let plan = ctx.oracle("plan the task").await?;
    ctx.effect("publish", plan, async |arg| {
        Ok(Value::from(format!("s3://{arg}")))   // awaited tool call
    }).await
}

let mut rt = AsyncRuntime::on_dir("./journals", oracle)?;
let report = rt.run("research-42", research).await?;

Implement AsyncOracle over your async HTTP client for native async model calls; every sync Oracle also works as-is.

Python

your terminal

pip install pragmatic-runtime

research.py

import pragmatic

rt = pragmatic.Runtime("./journals", my_model_fn)   # (str) -> str

def research(ctx):
    plan = ctx.oracle("plan the task")
    return ctx.effect("publish", plan, lambda arg: f"s3://{arg}")

report = rt.run("research-42", research)
report = rt.resume("research-42", research)   # crash-recover, no re-sampling
audit  = rt.replay("research-42", research)   # model never called
assert audit.trace == report.trace and rt.verify("research-42")

Journals written from Python are byte-compatible with the Rust runtime and every CLI command above.

Where next