Ironflow

Ironflow runs automations written in Rust. A workflow is a list of steps (a shell command, an HTTP call, an AI agent, a human approval) written as a normal async function. Ironflow runs each step, records its output, duration and cost, pauses when a human has to decide, and shows everything live in a web dashboard.

flowchart LR
    T["Button, CLI,<br/>webhook or cron"] --> R["Run"]
    R --> S1["Step: shell"]
    S1 --> S2["Step: AI agent"]
    S2 --> A{"Human approval"}
    A -->|approved| S3["Step: publish"]
    A -->|rejected| X["Run stops"]

Words used in this book

Where to start

• New here? Quick Start: run the demo in five minutes.
• Ready to write code? Writing a Workflow.
• Looking up how something works? The Concepts pages.
• Driving Ironflow from a terminal or an AI assistant? Interfaces.
• Deploying it? Running the Server and Running a Worker.
• Curious about the internals? Architecture.

The API reference of every crate is on docs.rs.

Quick Start

This page takes you from nothing to a running workflow in three stages: run the demo, drive it from a terminal, then start your own project.

1. Run the demo

You need Rust 1.94+, Node 22+ with pnpm for the dashboard, and, for agent steps, the Claude Code CLI.

git clone https://gitlab.com/ThomasTartrau/ironflow.git
cd ironflow

Build the dashboard first: the API server embeds ironflow-dashboard/dist/ when it compiles.

(cd ironflow-dashboard && pnpm install && pnpm build)

Start the API server. It serves the REST API and the dashboard on http://localhost:3000, with a dozen example workflows already registered:

IRONFLOW_ENV=development cargo run -p ironflow-example-server

IRONFLOW_ENV=development lets the server boot without secrets: it generates them and logs the worker token (start workers with WORKER_TOKEN=...).

In a second terminal, start a worker. The server stores runs; the worker is the process that executes them.

WORKER_TOKEN=<token from the server log> cargo run -p ironflow-example-worker

Open http://localhost:3000, create an account, open deploy-approval and click Run. The run executes its steps, then stops on an approval gate: approve it and the last step runs.

The example server keeps runs in memory, so they are gone when it stops. See Running the Server to use Postgres.

2. Drive it from a terminal

Create an API key in the dashboard, under API keys, then:

cargo install ironflow-cli

export IRONFLOW_URL=http://localhost:3000
export IRONFLOW_API_KEY=irfl_...

ironflow-cli workflow list
ironflow-cli run create ci-pipeline
ironflow-cli logs <run-id>

The CLI, the Rust SDK and the MCP server are described in Interfaces.

3. Start your own project

A project has three parts: a library with your workflows, a server binary and a worker binary. Installation shows the layout, and the example server and worker are the code to start from.

With Claude Code, the Ironflow plugin scaffolds that project for you with /ironflow setup.

Then write your first workflow with Writing a Workflow.

Installation

Prerequisites

• Rust 1.94+ (see rust-version in Cargo.toml)
• A running PostgreSQL instance (for production; in-memory store available for development)

Add dependencies

Add the crates you need to your Cargo.toml:

[dependencies]
ironflow-engine = "0.1"   # Workflow handler, context, engine
ironflow-api = "0.1"      # REST API server
ironflow-worker = "0.1"   # Background worker
ironflow-store = "0.1"    # Storage backends
ironflow-core = "0.1"     # Shell, agent providers

Minimal project structure

A typical Ironflow project has three parts:

1. A library crate with your workflow handlers
2. A server binary that exposes the API and serves the dashboard
3. A worker binary that executes workflows

my-project/
├── src/
│   └── lib.rs          # Your workflow handlers
├── src/bin/
│   ├── server.rs       # API server
│   └── worker.rs       # Background worker
└── Cargo.toml

See the example server and example worker for complete working code.

Running the Server

The API server exposes a REST API for managing workflows, runs, and steps. It also serves the web dashboard.

Example server

The repository includes a complete example server:

//! ironflow API server example.
//!
//! ```sh
//! cargo run -p ironflow-example-server
//! ```
//!
//! The dashboard is served automatically via the `dashboard` feature in `ironflow-api`.
//!
//! Environment:
//! - `IRONFLOW_ENV` (`production`, or `development` to boot without secrets;
//!   unset requires them)
//! - `IRONFLOW_INSECURE_COOKIES` (`1` or `true` drops the `Secure` cookie flag, ignored in production)
//! - `DATABASE_URL` (required in production)
//! - `JWT_SECRET` (required, >= 32 bytes; random per process in development)
//! - `WORKER_TOKEN` (required, >= 32 bytes; random per process in development,
//!   logged at startup for the worker)
//! - `PORT` (default: 3000)
//! - `DASHBOARD_DIR` (optional: overrides the embedded dashboard with a filesystem path)
//! - `ALLOWED_ORIGINS` (comma-separated list; omit to allow same-origin only)
//! - `TRUSTED_PROXIES` (optional: comma-separated IPs or CIDR ranges of the
//!   reverse proxies whose forwarding headers the rate limiters and the HTTPS
//!   redirect believe; unset, the TCP peer is the client and nothing is
//!   redirected to HTTPS)
//! - `TRUSTED_PROXY_HEADER` (default: `x-forwarded-for`; `x-real-ip` reads
//!   only `X-Real-IP` from a trusted proxy, with no fallback between the two)
//! - `WEBHOOK_URL` (optional: outbound webhook for run events)
//! - `ARTIFACTS_DIR` (optional: filesystem root for step artifacts; unset
//!   leaves artifacts disabled and the artifact routes answer `501`)
//! - `ARTIFACT_MAX_BYTES` (optional: per-artifact size limit, default 100 MiB)
//! - `IRONFLOW_DEFAULT_RUN_MAX_COST_USD` (optional: default per-run cost cap in
//!   USD, applied when neither the run creation request nor the workflow
//!   handler declares one; unset means no cap)
//! - `IRONFLOW_MONTHLY_COST_LIMIT_USD` (optional: global cost quota for the
//!   current calendar month in UTC; beyond it, creating a run returns
//!   `429 MONTHLY_BUDGET_EXCEEDED` while in-flight runs continue)
//! - `PURGE_MAX_AGE_DAYS`, `PURGE_MAX_RUNS_PER_WORKFLOW`, `PURGE_DRY_RUN`,
//!   `PURGE_INTERVAL_SECS`, `PROVIDER_ACCOUNT_USAGE_RETENTION_DAYS` and
//!   `SIGNAL_RETENTION_DAYS` (optional: retention of the purger, see
//!   `ServerConfig`)
//! - `IRONFLOW_SEED` (optional: when set to any value, seeds development data
//!   at startup -- users, runs, steps, API keys)
//!
//! The server runs with `ExecutionMode::Workers`: a run woken by a signal, an
//! approval or a human input goes back to `Pending`. Run `ironflow-example-worker`
//! (`examples/worker`) alongside, otherwise woken runs wait in `Pending`.
//!
//! `GET /api/v1/metrics` serves Prometheus metrics (the `prometheus` feature)
//! to an admin session or a `stats_read` API key owned by an admin.

use std::net::SocketAddr;
use std::process;
use std::sync::Arc;

use axum::http::header::{AUTHORIZATION, CONTENT_TYPE};
use axum::http::{HeaderValue, Method};
use tokio::net::TcpListener;
use tower_http::cors::CorsLayer;
use tracing::{info, warn};
use tracing_subscriber::EnvFilter;

use ironflow_api::config::ServerConfig;
use ironflow_api::purger::RunPurger;
use ironflow_api::routes::{RouterConfig, create_router};
use ironflow_api::sse::SseBroadcaster;
use ironflow_api::sse_limiter::SseLimits;
use ironflow_api::state::AppState;
use ironflow_artifacts::blob_store::BlobStore;
use ironflow_artifacts::local::LocalBlobStore;
use ironflow_auth::jwt::JwtConfig;
use ironflow_core::provider::AgentProvider;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_engine::artifact::DirectArtifactSink;
use ironflow_engine::budget::BudgetConfig;
use ironflow_engine::engine::{Engine, ExecutionMode};
use ironflow_engine::notify::{AuditLogSubscriber, Event, WebhookSubscriber, WorkflowEventBus};
use ironflow_store::crypto::{KeyRing, SECRET_KEYS_ENV};
use ironflow_store::memory::InMemoryStore;
use ironflow_store::store::Store;
use xtask::seed::{SeedOptions, seed_store};

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,ironflow=debug".parse().expect("valid filter")),
        )
        .init();

    let config = ServerConfig::from_env().unwrap_or_else(|e| {
        eprintln!("{e}");
        process::exit(1);
    });

    let mut store = InMemoryStore::new();

    let key_ring = KeyRing::from_env().unwrap_or_else(|e| {
        eprintln!("invalid secret key configuration: {e}");
        process::exit(1);
    });

    let has_key_ring = key_ring.is_some();
    match key_ring {
        Some(ring) => {
            info!(
                active_version = ring.active_version(),
                configured_versions = ?ring.versions(),
                "secret store enabled"
            );
            store.set_key_ring(ring);
        }
        None => {
            info!("{SECRET_KEYS_ENV} not set, secret store disabled");
        }
    }

    let store: Arc<dyn Store> = Arc::new(store);

    // A secret encrypted with a key that is no longer configured is
    // unreadable. Fail here rather than at the first workflow that needs it.
    if has_key_ring {
        let status = store.secret_key_status().await.unwrap_or_else(|e| {
            eprintln!("cannot read secret key versions: {e}");
            process::exit(1);
        });

        if !status.is_consistent() {
            let missing: Vec<String> = status.missing.iter().map(|v| v.to_string()).collect();
            eprintln!(
                "secret key versions present in database but missing from configuration: {}\n\
                 set {SECRET_KEYS_ENV} to include them, or rotate before removing a key",
                missing.join(", ")
            );
            process::exit(1);
        }
    }
    if std::env::var("IRONFLOW_SEED").is_ok() {
        info!("IRONFLOW_SEED set, seeding development data...");
        let seed_opts = SeedOptions {
            force: false,
            artifacts_dir: config.artifacts_dir.clone(),
        };
        seed_store(&*store, &seed_opts).await.unwrap_or_else(|e| {
            warn!("seed skipped: {e}");
        });
    }

    let provider = Arc::new(ClaudeCodeProvider::new());

    if !config.cookie_secure {
        warn!("IRONFLOW_INSECURE_COOKIES is set: session cookies are sent without the Secure flag");
    }

    let jwt_config = Arc::new(JwtConfig {
        secret: config.jwt_secret.clone(),
        access_token_ttl_secs: 900,
        refresh_token_ttl_secs: 604800,
        cookie_domain: None,
        cookie_secure: config.cookie_secure,
    });

    let budget = BudgetConfig::from_env();
    info!(
        default_run_max_cost_usd = ?budget.default_run_max_cost_usd,
        monthly_cost_limit_usd = ?budget.monthly_cost_limit_usd,
        "cost guardrails loaded"
    );

    // Go through `build_engine` so the mode is the one the test pins.
    let mut engine = build_engine(store.clone(), provider, budget);
    ironflow_workflows::register_all(&mut engine).expect("failed to register workflows");

    // Artifacts stay off until a storage root is configured. The API and any
    // in-process run then share the same backend, so a file a step produces is
    // downloadable from the same server that stored it.
    let blob_store: Option<Arc<dyn BlobStore>> = config.artifacts_dir.as_ref().map(|dir| {
        info!(
            dir = %dir.display(),
            max_bytes = config.artifact_max_bytes,
            "artifact storage enabled"
        );
        Arc::new(LocalBlobStore::new(dir).max_bytes(config.artifact_max_bytes))
            as Arc<dyn BlobStore>
    });

    if let Some(ref blob) = blob_store {
        engine.set_artifact_sink(Arc::new(DirectArtifactSink::new(
            blob.clone(),
            store.clone(),
        )));
    }

    if let Some(ref webhook_url) = config.webhook_url {
        info!(url = %webhook_url, "registering webhook subscriber");
        engine.subscribe(
            WebhookSubscriber::new(webhook_url),
            &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
        );
    }

    // Every engine event is persisted for `GET /api/v1/audit-logs`: who approved,
    // rejected, paused or resumed what. Log lines and provider account events do
    // not go through the engine publisher, so they are not duplicated here.
    engine.subscribe(AuditLogSubscriber::new(store.clone()), Event::ALL);
    let sse_broadcaster = SseBroadcaster::new();
    let event_sender = sse_broadcaster.sender();
    engine.subscribe(sse_broadcaster, Event::ALL);

    let event_bus = WorkflowEventBus::new();
    engine.set_event_bus(event_bus.clone());

    let engine = Arc::new(engine);

    let cors = build_cors(&config);

    let mut state = AppState::new(
        store.clone(),
        engine.clone(),
        jwt_config,
        config.worker_token.clone(),
        event_sender,
    )
    .with_event_bus(event_bus)
    .with_sse_limits(SseLimits {
        per_principal: config.sse_max_connections_per_principal.map(|n| n as usize),
        global: config.sse_max_connections.map(|n| n as usize),
    });
    if let Some(blob) = blob_store {
        state = state.with_blob_store(blob);
    }

    let shutdown = state.spawn_background_tasks().await;
    tokio::spawn(
        RunPurger::from_config(store.clone(), &config)
            .with_blob_store(state.blob_store.clone())
            .run(shutdown.clone()),
    );
    let router_config = RouterConfig {
        dashboard_dir: config.dashboard_dir.clone(),
        rate_limit_auth: config.rate_limit_auth,
        rate_limit_general: config.rate_limit_general,
        trusted_proxies: config.trusted_proxies.clone(),
        enforce_https: config.is_production,
    };
    // The rate limiters key on the TCP peer address: connect info is required.
    let app = create_router(state, router_config)
        .layer(cors)
        .into_make_service_with_connect_info::<SocketAddr>();

    let addr = format!("0.0.0.0:{}", config.port);
    let listener = TcpListener::bind(&addr).await.expect("bind address");

    info!("==============================================");
    info!("  ironflow server on http://{addr}");
    info!(
        "  environment: {}",
        if config.is_production {
            "production"
        } else {
            "development"
        }
    );
    info!("==============================================");

    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            tokio::signal::ctrl_c().await.expect("ctrl+c handler");
            info!("shutting down...");
            shutdown.cancel();
        })
        .await
        .expect("serve");
}

/// Builds the server's engine. The server resumes no run itself: a run woken by a
/// signal, an approval, a human input or the Waker goes back to `Pending` and a
/// worker, which owns the agent and the tools, finishes it.
fn build_engine(
    store: Arc<dyn Store>,
    provider: Arc<dyn AgentProvider>,
    budget: BudgetConfig,
) -> Engine {
    Engine::new(store, provider)
        .with_budget_config(budget)
        .with_execution_mode(ExecutionMode::Workers)
}

/// Build CORS layer from config.
///
/// - If `allowed_origins` is set: only those origins are permitted (comma-separated).
/// - If unset: no extra origins are allowed (same-origin only).
///
/// Credentials (cookies) are always allowed so JWT cookies work cross-origin.
fn build_cors(config: &ServerConfig) -> CorsLayer {
    let methods = vec![Method::GET, Method::POST, Method::PUT, Method::DELETE];
    let headers = vec![AUTHORIZATION, CONTENT_TYPE];

    match config.allowed_origins {
        Some(ref raw) => {
            let origins: Vec<HeaderValue> = raw
                .split(',')
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .filter_map(|s| match s.parse::<HeaderValue>() {
                    Ok(v) => Some(v),
                    Err(err) => {
                        warn!(origin = s, %err, "ignoring invalid CORS origin");
                        None
                    }
                })
                .collect();

            info!(?origins, "CORS: allowing configured origins");

            CorsLayer::new()
                .allow_origin(origins)
                .allow_methods(methods)
                .allow_headers(headers)
                .allow_credentials(true)
        }
        None => {
            info!("CORS: no ALLOWED_ORIGINS set, same-origin only");

            CorsLayer::new()
                .allow_methods(methods)
                .allow_headers(headers)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn server_engine_resumes_woken_runs_on_workers() {
        let engine = build_engine(
            Arc::new(InMemoryStore::new()),
            Arc::new(ClaudeCodeProvider::new()),
            BudgetConfig::new(),
        );

        assert_eq!(
            engine.execution_mode(),
            ExecutionMode::Workers,
            "a Local server would resume woken runs itself, without the worker's agent or tools"
        );
    }
}

Environment variables

No secret is built into the binary. JWT_SECRET and WORKER_TOKEN must be at least 32 bytes and must not start with ironflow-dev-; generate them with openssl rand -hex 32. Only IRONFLOW_ENV=development boots without them. Production also requires DATABASE_URL. Any violation aborts at boot with every error listed.

Secret encryption keys

IRONFLOW_SECRET_KEYS holds one or more versioned AES-GCM keys, as version:hex pairs separated by commas. Each key is 64 hex characters (32 bytes):

IRONFLOW_SECRET_KEYS="1:0123...ef,2:fedc...10"
IRONFLOW_SECRET_ACTIVE_KEY_VERSION=2

Every key in the ring can decrypt; only the active one encrypts. Without any key the secret store stays off and workflows reading secrets fail.

IRONFLOW_SECRET_KEY (a single unversioned key) is still accepted and read as version 1, so existing deployments keep working. It is deprecated: when IRONFLOW_SECRET_KEYS is also set, it is ignored with a warning.

The server refuses to start if a stored secret uses a key version absent from the configuration, and names the missing versions. That is the safety net behind the rotation procedure below.

Rotating the encryption key

# 1. Add the new key without activating it, then restart.
#    New secrets stay on version 1; version 2 is merely available.
IRONFLOW_SECRET_KEYS="1:<hexA>,2:<hexB>"
IRONFLOW_SECRET_ACTIVE_KEY_VERSION=1

# 2. Activate version 2, then restart.
#    New secrets use version 2; older ones stay readable.
IRONFLOW_SECRET_ACTIVE_KEY_VERSION=2

# 3. Re-encrypt the existing secrets.
ironflow-cli secret rotate

# 4. Confirm version 1 is no longer used by any secret.
ironflow-cli secret key-status

# 5. Drop version 1, then restart.
IRONFLOW_SECRET_KEYS="2:<hexB>"

Step 1 is kept separate from step 2 so rolling back to the previous deployment stays possible for as long as nothing has been encrypted with the new key.

secret rotate works in batches and is safe to interrupt: secrets already re-encrypted are skipped on the next run, and every secret stays readable throughout. It re-encrypts in place: the ID, key and timestamps of a secret never change. A secret that cannot be decrypted is skipped and reported, and the command exits non-zero.

secret key-status reports which versions are configured, which are used by stored secrets, and which can be retired.

Transport security

Session cookies are HttpOnly, SameSite=Lax and Secure. Browsers treat http://localhost as a secure context, so Secure cookies work in local development. For a plain-HTTP setup on another origin, set IRONFLOW_INSECURE_COOKIES=1; the flag is ignored when IRONFLOW_ENV=production.

In production the example server also sets RouterConfig::enforce_https: TLS terminates at your reverse proxy, and any request that arrives with X-Forwarded-Proto: http is answered with a 308 Permanent Redirect to https://<host><path> (the host comes from X-Forwarded-Host, then Host). The method and body are preserved. Requests without X-Forwarded-Proto, such as in-cluster health probes and worker traffic, are not redirected. Every response carries Strict-Transport-Security.

The redirect reads X-Forwarded-Proto and X-Forwarded-Host only when the TCP peer is listed in TRUSTED_PROXIES. From any other peer these headers are client input: they are ignored and the request is served as is. Set TRUSTED_PROXIES in production, or nothing is redirected (the server logs a warning at startup).

The reverse proxy must therefore be listed in TRUSTED_PROXIES, set X-Forwarded-Proto (and X-Forwarded-Host if it rewrites Host), and overwrite any value sent by the client.

Rate limiting and client IP

An unauthenticated request is rate limited under its client IP: the address of the TCP peer. Serve the router with into_make_service_with_connect_info::<SocketAddr>(), as the example server does; without it the IP limit is skipped and a warning is logged once.

Forwarding headers are written by the client, so they are ignored unless the peer is listed in TRUSTED_PROXIES. Behind a reverse proxy, list it (TRUSTED_PROXIES=10.0.0.0/8). Only the header named by TRUSTED_PROXY_HEADER is read then. With the default, x-forwarded-for, X-Forwarded-For is read from the right, trusted hops are skipped and the first other address is the client. With x-real-ip, X-Real-IP is the client. The other header is never read, not even as a fallback. Without TRUSTED_PROXIES, every user shares the proxy’s bucket.

A request with an API key is counted under that key only once its secret is verified. A secret that does not verify counts on the IP bucket. Unverified secrets are checked at most 10 times per minute per client and key, then answered 429. Only an admin may set an API key’s rate_limit_override (a member gets 403), and it is honoured only for keys owned by an admin.

Sign-in and sign-up are also counted per targeted email, so attempts on one account spread over many addresses are limited too. The two routes have separate counters: a sign-up flood does not lock the account out of sign-in.

Password change (PATCH /api/v1/auth/password) sits behind the same limiter, counted per signed-in user: guesses at the current password beyond RATE_LIMIT_AUTH per minute answer 429 before the password is checked, and each wrong current password is logged with the user id.

Accounts and passwords

A password chosen at sign-up, password change or admin user creation must be 12 to 128 characters, not a common password (nor one padded with digits and symbols, like Password2024!), must not contain the email or username, and must not be repetitive. Otherwise the request answers 400 WEAK_PASSWORD with the rule it broke.

Sign-up answers 204 without a session, whether the email was free or already registered: the dashboard signs in next with the same credentials. A taken username still answers 409 DUPLICATE_USERNAME.

The first account created on an empty database becomes admin. The decision is atomic: concurrent sign-ups on a fresh install cannot each see an empty table, so exactly one of them is admin.

Sign-in on an unknown email runs one Argon2 verification against a throwaway hash, like a wrong password, so the answer time does not tell whether the email has an account. Both answer 401 INVALID_CREDENTIALS and are logged without the email.

Sign-up has no email verification, so it still tells whether an email has an account: a sign-up followed by a sign-in with the chosen password succeeds only if the email was free. When email enumeration matters, build the server without the sign-up feature in production and create accounts as an admin, or keep sign-up behind an invite flow.

API key scopes

A request authenticated with an API key (Authorization: Bearer irfl_...) must hold the scope of the route, otherwise it answers 403 INSUFFICIENT_SCOPE before any lookup, so a key without the scope cannot tell whether a run or a secret exists. The admin scope implies every other scope. Paths below are relative to /api/v1.

Routes that were admin-only stay so: POST /runs, cancel, retry, replay, pause and resume of a run, POST /signals, GET /metrics, and every admin route also need the key’s owner to be an admin, checked at request time, so demoting the owner takes effect at once (403 FORBIDDEN). Demoting a user also removes from each of their keys every scope a member may not hold (only workflows_read, runs_read and stats_read remain) and disables the keys left with no scope; promoting them again does not restore those scopes. Deleting a user deletes their keys, and a key whose owner no longer exists answers 401 TOKEN_REVOKED. Approving, rejecting and answering a gate keep their assignee and delegation rules on top of the scope.

On GET /events, a member session, or a key without the admin scope or owned by a member, only receives run events (run_*, step_*, approval_*, log_line, retry_forced, signal_*). User sign-in, sign-up and sign-out, Provider Account, schedule and workflow pause/resume events reach only an admin session or an admin-scope key owned by an admin. An open GET /events stream re-reads these rights every 30 seconds: it is closed when its key is revoked, disabled, expired or loses runs_read, or when its user or key owner is deleted, and it falls back to run events only when the user is demoted or the key loses its admin scope. Within those 30 seconds, events still go through.

Schedule, template and delegation routes are not admin-only: a key owned by a member reaches them exactly as that member’s dashboard session does. Revoking a delegation still needs to be its delegator or an admin, and listing still shows a non-admin only the delegations they are part of.

Signals and Provider Accounts have their own scopes (signals_send, accounts_read, accounts_manage). Dashboard sessions (JWT) are not scope-checked: only the account’s admin flag applies to them.

Metrics

Built with the prometheus feature, the server exposes GET /api/v1/metrics. It needs an admin session, or an API key with the stats_read (or admin) scope owned by an admin; an unauthenticated request gets 401. Create a key for the scraper (ironflow-cli api-key create prometheus --scope stats_read) and hand it to Prometheus as a bearer token:

scrape_configs:
  - job_name: ironflow
    metrics_path: /api/v1/metrics
    authorization:
      type: Bearer
      credentials_file: /etc/prometheus/ironflow-api-key
    static_configs:
      - targets: ["ironflow:3000"]

Audit log

The example server records every engine event in the audit log read by GET /audit-logs (and the Audit page of the dashboard) through AuditLogSubscriber. A custom server must subscribe it itself, otherwise the audit log stays empty:

engine.subscribe(AuditLogSubscriber::new(store.clone()), Event::ALL);

Log lines and Provider Account events do not go through the engine publisher, so they are neither recorded nor duplicated. Who acted is the entry’s user_id, also present in the payload of approval_granted, approval_rejected (a gate decision or a human input rejection, with its reason), workflow_paused and workflow_resumed. For an API key it is the key’s owner; the event keeps the key name as its label (approved_by, rejected_by). A decision taken by the SLA timer (system:timeout) has no user_id. A pause is recorded on every request, a resume only when the workflow was paused.

GET /events sends a lagged event, whatever its filters, to a client that read slower than events were published: data: {"missed": n} counts the events it lost. Its view is stale; reload it.

Retention

The example server starts a RunPurger built from these variables: it purges old runs, Provider Account usage history and received signals. A custom server must start it itself with RunPurger::from_config, otherwise these variables are ignored:

let shutdown = state.spawn_background_tasks().await;
tokio::spawn(
    RunPurger::from_config(store.clone(), &config)
        .with_blob_store(state.blob_store.clone())
        .run(shutdown.clone()),
);

Running

IRONFLOW_ENV=development cargo run -p ironflow-example-server

The server starts on http://localhost:3000. The dashboard is available at the root URL. In development it generates its secrets and logs the worker token (start workers with WORKER_TOKEN=...). Without IRONFLOW_ENV=development, set JWT_SECRET and WORKER_TOKEN (openssl rand -hex 32): no secret is built in.

Pausing runs and workflows

An administrator can hold work without cancelling it. Every route below needs an admin account (403 otherwise):

An unknown run or workflow answers 404. Pausing a finished, already paused or sub-workflow run, or resuming a run that is not paused, answers 400. The CLI offers the same actions as ironflow-cli run pause|resume <id> and ironflow-cli workflow pause|resume <name>. See Pausing runs and workflows for the lifecycle details.

Running a Worker

Workers poll the API for pending runs, acquire leases, and execute workflow handlers.

Example worker

//! ironflow worker example.
//!
//! ```sh
//! cargo run -p ironflow-example-worker
//! ```
//!
//! Environment:
//! - `API_URL` (default: http://localhost:3000)
//! - `WORKER_TOKEN` (required: the server's token)
//! - `CONCURRENCY` (default: 2)
//! - `POLL_INTERVAL_SECS` (default: 2)

use std::env;
use std::process;
use std::sync::Arc;
use std::time::Duration;

use tracing::info;
use tracing_subscriber::EnvFilter;

use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_worker::WorkerBuilder;
use ironflow_workflows::handlers;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,ironflow=debug".parse().expect("valid filter")),
        )
        .init();

    let api_url = env::var("API_URL").unwrap_or_else(|_| "http://localhost:3000".to_string());
    let worker_token = env::var("WORKER_TOKEN")
        .ok()
        .filter(|token| !token.is_empty())
        .unwrap_or_else(|| {
            eprintln!(
                "WORKER_TOKEN is required: use the server's token (in IRONFLOW_ENV=development, \
                 the server logs the one it generated at startup)"
            );
            process::exit(1);
        });
    let concurrency: usize = env::var("CONCURRENCY")
        .ok()
        .and_then(|c| c.parse().ok())
        .unwrap_or(2);
    let poll_interval: u64 = env::var("POLL_INTERVAL_SECS")
        .ok()
        .and_then(|p| p.parse().ok())
        .unwrap_or(2);

    let mut builder = WorkerBuilder::new(&api_url, &worker_token)
        .provider(Arc::new(ClaudeCodeProvider::new()))
        .concurrency(concurrency)
        .poll_interval(Duration::from_secs(poll_interval));

    // Same list as the server: one source of truth for both binaries.
    for handler in handlers() {
        builder = builder.register(handler);
    }

    let worker = builder.build().expect("failed to build worker");

    info!("==============================================");
    info!("  ironflow worker");
    info!("  API: {api_url}");
    info!("  Concurrency: {concurrency}");
    info!("==============================================");

    if let Err(e) = worker.run().await {
        tracing::error!("worker error: {e}");
    }
}

Environment variables

Running

WORKER_TOKEN=<token logged by the server> cargo run -p ironflow-example-worker

Scaling

To increase throughput, start multiple workers. Each worker polls independently and acquires leases on runs, so no coordination is needed beyond the API server.

WorkflowHandler

A WorkflowHandler is the core abstraction in Ironflow. It defines a named workflow as imperative Rust code.

The trait

pub trait WorkflowHandler: Send + Sync {
    fn name(&self) -> &str;
    fn description(&self) -> &str;
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a>;

    // Optional methods
    fn category(&self) -> Option<&str> { None }
    fn input_schema(&self) -> Option<Value> { None }
    fn default_labels(&self) -> HashMap<String, String> { HashMap::new() }
    fn source_code(&self) -> Option<&str> { None }
    fn priority(&self) -> i16 { 0 }
}

Example: a greeting workflow

use std::collections::HashMap;

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

/// Input payload for the greeting workflow.
///
/// Derives [`JsonSchema`] so the dashboard can render a dynamic form.
#[derive(Deserialize, JsonSchema)]
struct GreetingInput {
    /// Person to greet.
    name: String,
    /// Greeting language (en, fr, es).
    #[serde(default = "default_language")]
    language: String,
    /// Number of times to repeat the greeting.
    #[serde(default = "default_repeat")]
    repeat: u32,
    /// Whether to output in uppercase.
    #[serde(default)]
    uppercase: bool,
}

fn default_language() -> String {
    "en".to_string()
}

fn default_repeat() -> u32 {
    1
}

pub struct Greeting;

impl WorkflowHandler for Greeting {
    fn name(&self) -> &str {
        "greeting"
    }

    fn category(&self) -> Option<&str> {
        Some("examples")
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<GreetingInput>())
    }

    fn default_labels(&self) -> HashMap<String, String> {
        HashMap::from([("project".to_string(), "ironflow".to_string())])
    }

    fn description(&self) -> &str {
        "A demo workflow that greets someone. \
         Shows how input_schema generates a dynamic form in the dashboard."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("greeting.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: GreetingInput = ctx.input().await?;

            let greeting = match input.language.as_str() {
                "fr" => format!("Bonjour, {} !", input.name),
                "es" => format!("Hola, {}!", input.name),
                _ => format!("Hello, {}!", input.name),
            };

            let mut message = (0..input.repeat)
                .map(|_| greeting.as_str())
                .collect::<Vec<_>>()
                .join("\n");

            if input.uppercase {
                message = message.to_uppercase();
            }

            // `name` is untrusted: pass it as an argument, never inside a
            // command line, so no shell ever parses it.
            ctx.shell("greet", ShellConfig::exec("printf", &["%s\n", &message]))
                .await?;

            Ok(())
        })
    }
}

Key points

• name() must be unique across all registered handlers. It identifies the workflow in the API and the database.
• execute() receives a WorkflowContext to create steps. Steps are persisted as they complete.
• input_schema() returns a JSON Schema derived from a #[derive(JsonSchema)] struct. The dashboard renders it as a dynamic form.
• source_code() optionally embeds the handler source for display in the dashboard.
• sub_workflows() lists the handlers this one calls through ctx.workflow, for the call graph. Build it from the handlers, sub_workflow_names(&[&Collect]), never from hand-written names.
• priority() sets the queue priority of the runs, from -100 to 100 (out of range values are clamped). Workers pick the highest priority first, then the oldest; nothing is preempted or aged. See Engine & Worker.

Typed input for sub-workflows

A handler called as a sub-workflow declares its input type with TypedWorkflow. ctx.workflow(&Collect, CollectInput { .. }) then accepts nothing else, and input_schema() is derived from the same type:

#[derive(Serialize, Deserialize, JsonSchema)]
struct CollectInput {
    host: String,
}

impl WorkflowHandler for Collect {
    fn name(&self) -> &str { "collect" }
    fn input_schema(&self) -> Option<Value> { Self::typed_input_schema() }
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> { /* .. */ }
}

impl TypedWorkflow for Collect {
    type Input = CollectInput;
}

// In the parent:
let child = ctx.workflow(&Collect, CollectInput { host: "db-1".into() }).await?;
let report: Option<CollectReport> = child.output()?;

A child without input uses type Input = ();.

Run output

A handler publishes a typed result with ctx.set_output(&value). The value is serialized to JSON and persisted on the run (Run.output) when the run ends: completed, warning, failed or cancelled. WorkflowHandler::execute does not change: it still returns Result<(), EngineError>.

#[derive(Serialize, Deserialize)]
struct CollectReport {
    disks: Vec<String>,
}

// In the child:
ctx.set_output(&CollectReport { disks })?;

• The last call wins. A handler that sets nothing leaves the output empty.
• A failed run keeps the output set before the error, so a verdict survives a handler that returns Err after reaching it.
• set_output fails with EngineError::Serialization when the value does not serialize to JSON.
• While planning (plan_handler, the execution plan API), set_output does nothing.
• A suspended run (approval, human input, signal, delay) writes no output; the handler replays on resume and sets it again.

The output is returned by GET /api/v1/runs/{id} (run.output, omitted when empty) and shown in the run page of the dashboard. A parent reads its child output with child.output::<T>(), see sub-workflow steps.

Worker tags

A workflow that must run on a given kind of host (a GPU, a region, a network) declares the worker tags every one of its runs requires:

impl WorkflowHandler for Transcode {
    fn name(&self) -> &str {
        "transcode"
    }

    fn required_worker_tags(&self) -> Vec<String> {
        vec!["gpu".into()]
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move { /* steps */ Ok(()) })
    }
}

Only a worker built with all of these tags (WorkerBuilder::tags) takes the run. Tags given when the run is created are added to this list. The default is empty: any worker that registered the workflow may take it. See Routing runs to workers.

Registration

Handlers are registered in the Engine before starting the server or worker:

let mut engine = Engine::new(store, provider);
engine.register(Box::new(Greeting))?;

See Writing a Workflow for a step-by-step guide.

Runs

A Run is one execution of a workflow. It is created by a trigger, waits in the queue as pending, is picked by a worker, and ends completed, failed, warning or cancelled. Every step it executes is persisted with its output, duration and cost. An agent step built from an AgentRole also records the role name as agent_role, so steps and costs can be grouped by role across workflows.

This page covers how runs are created and the options that control it. How workers pick them, with concurrency groups and priority, is in Engine & Worker.

Triggers

A run can also carry a scheduled_at: set it at creation and the run stays pending until that time.

Idempotent runs

POST /api/v1/runs accepts an optional Idempotency-Key header. Sending the same key again returns the run it already created instead of starting a second one. It protects against webhook replays and client retries after a network timeout.

Rules:

• A key is at most 255 printable ASCII characters and must not be empty.
• A key is global, not scoped per workflow: reusing one across two workflows is a conflict. Prefix it (github:...) to keep sources apart.
• A key stays bound for 24 hours. Past that window it is released and the next call with it creates a new run.
• The run returned on replay is the original one whatever its state, including failed or cancelled. Use POST /api/v1/runs/:id/retry to run it again deliberately; a retry never inherits the key.
• Only the workflow and the payload decide replay versus conflict. Labels are merged with the handler defaults server-side and are not compared.

# Same key twice: one run, the second call answers 200.
curl -X POST https://ironflow.example.com/api/v1/runs \
  -H "Authorization: Bearer $TOKEN" \
  -H "Idempotency-Key: github:8f4e2a10" \
  -H "Content-Type: application/json" \
  -d '{"workflow": "deploy", "payload": {"env": "prod"}}'

A conflict names the run holding the key:

{
  "error": {
    "code": "IDEMPOTENCY_KEY_CONFLICT",
    "message": "idempotency key already used with a different request",
    "details": { "run_id": "0199c3f0-..." }
  }
}

The SDK, the CLI and the MCP server all carry the key:

#![allow(unused)]
fn main() {
use ironflow_sdk::IronflowClient;
use ironflow_sdk::types::CreateRunRequest;

async fn example(request: &CreateRunRequest) -> Result<(), Box<dyn std::error::Error>> {
let client = IronflowClient::new("http://localhost:3000", "irfl_...");
client.create_run_idempotent(request, "github:8f4e2a10").await?;
Ok(())
}
}

ironflow-cli run create deploy --payload '{"env":"prod"}' \
  --idempotency-key github:8f4e2a10

A webhook received by the standalone runtime exposes the provider delivery id, ready to use as the key.

With the prometheus feature, ironflow_run_idempotency_total{outcome} counts created, replayed and conflict outcomes.

Exclusive runs

POST /api/v1/runs also accepts an optional concurrency_key in the body. While a run holding the same key is not finished (pending, running, sleeping, retrying, awaiting approval), a second creation is refused with 409 CONCURRENCY_CONFLICT naming that run. The key is released when the holder completes, fails, ends with a warning or is cancelled. Use it to keep one active run per issue, branch or tenant.

curl -X POST https://ironflow.example.com/api/v1/runs \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"workflow": "fix-issue", "payload": {"issue": 12}, "concurrency_key": "issue:12"}'

{
  "error": {
    "code": "CONCURRENCY_CONFLICT",
    "message": "concurrency key \"issue:12\" is held by active run 0199c3f0-...",
    "details": { "key": "issue:12", "run_id": "0199c3f0-..." }
  }
}

• A key is at most 255 bytes and must not be empty. It is global, like an idempotency key: prefix it to keep sources apart.
• An idempotent replay (same Idempotency-Key, same request) is answered before the exclusivity check, so it returns the original run instead of a conflict.
• A retry (POST /api/v1/runs/:id/retry) inherits the key and answers 409 if an active run took it since.
• The CLI takes --concurrency-key issue:12 and the MCP create_run tool a concurrency_key argument.

A sub-workflow takes the key through ctx.workflow_with; a conflict completes the step with the run holding the key instead of failing the parent (see Exclusive children).

A concurrency key refuses a second run. To queue it instead, use a concurrency group.

Schedules

A schedule creates a run of a workflow on a cron expression. It is declared in code by the handler, or created through the API, the CLI, the MCP server or the dashboard. The server fires it; workers execute the runs like any other run.

Declaring a schedule

A handler returns its schedule from schedule(). The policy builders set the timezone, the catch-up of missed occurrences and the overlap with a run still active:

use std::sync::LazyLock;
use std::time::Duration;

use ironflow_engine::prelude::*;

static MORNING: LazyLock<CronSchedule> = LazyLock::new(|| {
    CronSchedule::new("0 9 * * *")
        .and_then(|s| s.with_timezone("Europe/Paris"))
        .map(|s| {
            s.with_catchup(CatchupPolicy::All)
                .with_catchup_max(5)
                .with_catchup_window(Duration::from_secs(3 * 86_400))
                .with_overlap(OverlapPolicy::Skip)
        })
        .expect("valid schedule")
});

impl WorkflowHandler for MorningReport {
    fn name(&self) -> &str { "morning-report" }
    fn schedule(&self) -> Option<&CronSchedule> { Some(&MORNING) }
    // ...
}

At startup the server syncs every handler schedule to the database: a new handler creates its schedule, a changed expression, priority or policy updates it, a removed handler deletes it.

Catch-up

When no server fired a schedule for a while (a deployment, an outage), its occurrences pile up. The catch-up policy decides what happens to them on the next tick:

• catchup_max bounds all, from 1 to 1000. Default: 10.
• catchup_window bounds every policy: an occurrence older than the window is dropped. From one minute to 30 days. Default: one day.
• An occurrence is on time when it is less than the on-time grace old: twice the ticker interval, and never less than a minute. The window used is never narrower than this grace, so an on-time occurrence is never dropped as too old.

Each run carries its occurrence. The occurrence is part of the run idempotency key, so two servers firing the same backlog create each run once.

A paused schedule is never caught up: resuming it computes the next occurrence from now, whatever the policy.

Overlap

The overlap policy decides what an occurrence does while a run of the same schedule is still active (pending, running or waiting):

skip gives every run of the schedule the concurrency key schedule:<id>. A manual trigger of the schedule contends for the same key: it fails with 409 Conflict while a run is active.

With catchup = all and overlap = skip, the first catch-up run takes the key and the next occurrences of the same backlog are skipped as overlap.

Timezone and daylight saving time

The expression is evaluated in an IANA timezone, UTC by default. 0 9 * * * in Europe/Paris fires at 9:00 Paris time all year: 7:00 UTC in summer, 8:00 UTC in winter.

• A local time skipped by the spring change (2:30 in Paris on the last Sunday of March) fires once, at the end of the gap (3:00).
• A local time repeated by the autumn change fires once, on its first pass.

An invalid timezone is rejected when the schedule is created. A stored timezone the server no longer knows disables the schedule, with the reason in last_error, like an invalid expression.

Reading the occurrence in a workflow

ctx.trigger() returns how the run was triggered. For a scheduled run it carries the schedule and the occurrence the run covers, which a catch-up run needs to process the right period:

use ironflow_store::models::TriggerKind;

if let TriggerKind::Cron { scheduled_for: Some(occurrence), .. } = ctx.trigger().await? {
    // Process the day of `occurrence`, not the day of now.
}

scheduled_for is None for a manual trigger of the schedule.

Missed occurrences

Every occurrence dropped by the policies is traced:

• a warn log with the schedule, the reason and the count;
• the ironflow_schedule_missed_total counter, labelled by schedule;
• a schedule_occurrences_missed event, with the reason (outside_window, catchup_max, superseded, catchup_skip or overlap), the count and the first and last occurrence. Subscribe to it like any other event.

API, CLI and MCP

POST /api/v1/schedules accepts the optional fields catchup, catchup_max, catchup_window_secs, overlap and timezone. Every schedule response returns them.

ironflow schedule create morning-report "0 9 * * *" \
  --timezone Europe/Paris --catchup all --catchup-max 5 \
  --catchup-window 259200 --overlap skip

The create_schedule MCP tool takes the same optional fields.

Steps

A Step is an atomic unit of work within a Run. Each step is persisted in the database with its input, output, status, cost, duration, and token counts.

Step kinds

Shell steps

let output = ctx.shell("build", ShellConfig::new("cargo build")).await?;
if output.is_success() {
    // continue
}

ShellConfig::new hands its string to sh -c: pipes, redirects and globs work, and so would a quote or a ; smuggled into it. When the command carries data the workflow does not control (its input, a webhook payload, a human answer, an agent output), use ShellConfig::exec, which spawns the program directly and passes each argument as is:

let input: GreetInput = ctx.input().await?;
ctx.shell("greet", ShellConfig::exec("printf", &["Hello, %s!\n", &input.name])).await?;

The step’s command holds the program and its args the arguments; timeout_secs, dir, env, exit_code_as_output and artifacts work the same in both modes.

HTTP steps

let response = ctx.http("fetch-data", HttpConfig::get("https://api.example.com/data")).await?;

An HTTP step refuses a host that is, or resolves to, a private, loopback, link-local or cloud metadata address (localhost, 10.0.0.5, a Kubernetes service name, 169.254.169.254). Allow an internal service with .allow_host("billing.internal"), or for the whole deployment with the worker’s IRONFLOW_HTTP_ALLOWED_HOSTS (comma-separated).

A shell step reads back through stdout(), stderr() and exit_code(), an HTTP step through status() and body(). A step tolerated with allow_failure(), or an operation run with ctx.operation_allow_failure(), that failed reads back its message through error(). When the run resumes, for instance after a signal, a tolerated failure is replayed with the same {"error": ...} output and the step is not executed again. Files a shell step declares with .output("target/*.log") are handed to later steps through a handle: build.artifact("build.log")?, passed to ShellConfig::input(&handle) or ctx.get_artifact(&handle). A name the step did not declare fails with EngineError::ArtifactNotDeclared.

Agent steps

let result = ctx.agent("analyze", AgentStepConfig::new("Analyze this log file")).await?;
println!("{}", result.text().raw());

#[derive(Deserialize, JsonSchema, AgentAnswer)]
struct Verdict { approved: bool }

// With `.output::<T>()` the step returns the `T` itself.
let verdict = ctx
    .agent("review", AgentStepConfig::new("Review the diff").max_turns(2).output::<Verdict>())
    .await?;

To read the environment_id or account_id of a typed step as well, call ctx.agent_with_meta(..): it returns an AgentReply with the typed answer next to them.

.effort(Effort::High) sets the reasoning effort (low, medium, high, xhigh, max). It is passed as --effort to the Claude CLI providers and as output_config.effort to the Anthropic API provider; other HTTP providers ignore it. Without it no flag is sent and the model default applies.

.timeout(Duration::from_secs(300)) bounds the step whatever the provider: past it the step fails with agent timed out after 300s, and its environment is stopped. The local claude process is killed, the K8s ephemeral pod is deleted, and with Docker, SSH or a persistent pod the remote command is killed by timeout -s KILL 5 s after the limit (the container or host then needs a timeout binary, from coreutils or busybox). Without it, the provider timeout applies; the step timeout may be longer than it. It covers the automatic retries of a structured-output step, while each retry_policy attempt gets a new budget. It is not part of the record/replay hash, so changing it keeps the recorded fixtures.

Tools are an enum: .allow_tool(Tool::Bash), Tool::Custom("mcp__server__tool".into()) for anything else. .available_tools([Tool::Read]) bounds the built-in tools the Claude CLI offers (--tools), where allow_tool only pre-approves. A step can have tools and structured output: with .allow_tool(Tool::Read).output::<Verdict>() the agent reads what it needs, then ctx.agent returns the Verdict. Claude Code fills the structured answer with tools active (tested with 2.1.296 and 2.1.304); count one more turn in max_turns per round of tool calls.

On an HTTP provider, tools are registered on the worker in named profiles (HttpAgentProvider::with_tool_profile). Each profile is a ToolProfile constant, const BUG: ToolProfile = ToolProfile::new("bug");, shared by the worker and the handlers, so a misspelled profile does not compile. A step picks one with .tool_profile(BUG) and sees only its tools; without a profile it gets the with_tools registry, or none. An unknown profile fails the step, a Claude CLI provider refuses any profile, and the run logs name the profile with the tools it exposed.

With Provider Accounts, .account("perso-max") pins a step to one account and .account_pool("team") to the accounts carrying a tag. When none of them has capacity, the run sleeps until the next reset and the step runs again from the start, up to .max_capacity_wait(..) (6 hours by default, .fail_fast_on_capacity() fails fast); see When every account is limited.

Reusable agent roles

An agent used by several workflows (a reviewer, a triager) is defined once with impl AgentRole, then called with a task:

struct Reviewer;

impl AgentRole for Reviewer {
    type Input = String;
    type Output = Verdict;

    fn name(&self) -> &str {
        "reviewer"
    }

    fn prompt(&self, diff: String) -> String {
        Prompt::new().text("Review this diff:").untrusted(&diff).to_string()
    }

    fn append_system_prompt(&self) -> Option<&str> {
        Some("You are a strict reviewer.")
    }

    fn tools(&self) -> Vec<Tool> {
        vec![Tool::Read]
    }

    fn max_turns(&self) -> Option<u32> {
        Some(4)
    }
}

let verdict: Verdict = ctx
    .agent("review", Reviewer.task(diff).max_budget_usd(0.25))
    .await?;

Only prompt is required. The persona, tools, permission mode, MCP config and output type are locked by the role: the task has no setter for them, so a call that widens the rights does not compile (a role with more rights is another role). The model, budget, max_turns, effort, timeout and per-call settings (working_dir, allow_failure, retry_policy, pod labels, resume settings) are defaults the caller may override.

name() names the role. It defaults to the type name without its module path (Reviewer); override it, as above, to keep the name stable across a type rename, since renaming the type splits the role’s steps and costs between the old and the new name. The caller of task cannot override it. The name must not be blank and holds at most 128 bytes: task panics otherwise. It is recorded on every step the role runs, as the step’s agent_role. The steps of a role, across every workflow and run, are listed by GET /api/v1/steps?agent_role=reviewer, ironflow-cli run steps --role reviewer and the dashboard page Agent Roles. The cost, executions and failures per role over a period are returned by GET /api/v1/stats/agent-roles?period=7d and ironflow-cli stats roles. Steps stored before roles were recorded have no role.

The step is stored and replayed like any AgentStepConfig step. An override changes the step config, so it changes the hash of a RecordReplayProvider fixture: a fixture recorded for the role defaults does not match an overridden task.

For a role with tools, put the persona in append_system_prompt: system_prompt replaces the Claude Code prompt, which carries the tool and skill instructions. A structured output needs at least 2 turns.

Untrusted content in a prompt

Text the workflow did not write (an issue, a review comment, a diff, a CI log, a webhook payload) can carry instructions aimed at the agent. Build the prompt with Prompt instead of format!:

use ironflow_engine::config::{AgentStepConfig, Prompt};

let prompt = Prompt::new()
    .text("Propose labels for this issue:")
    .untrusted(&issue.body)
    .text("Answer with the labels only.");
let labels = ctx
    .agent("propose-labels", AgentStepConfig::new(&prompt.to_string()))
    .await?;

Each untrusted(..) part lands in a block whose tags carry a nonce drawn for the prompt, <untrusted nonce="..."> … </untrusted nonce="...">. Inside the content, every < followed by the tag name, in any ASCII case and after any run of whitespace and /, becomes <, so a literal forged tag neither closes the block nor opens another one. Look-alikes the escaping does not catch (a full-width less-than sign, a zero-width character in the name) cannot carry the nonce, which the content’s author never sees. When the prompt holds a block, it opens with DEFAULT_UNTRUSTED_INSTRUCTION, which tells the agent to treat the blocks as data; .instruction(..) replaces that text. Prompt lives in ironflow_core::prompt and works in library mode too (Agent::new().prompt(&prompt.to_string())).

RecordReplayProvider masks the nonce when it names a fixture: the same prompt finds its fixture from one run to the next.

Agent output in a prompt

What an agent wrote is untrusted too: it may repeat, word for word, instructions planted in a page, an issue or a log it read. StepOutput::text() returns an AgentText, which enters a prompt only through untrusted(..). It has no Display, no AsRef<str>: AgentStepConfig::new(&out.text()) and format!("{}", out.text()) do not compile.

use ironflow_engine::config::{AgentAnswer, AgentStepConfig, AgentText, Prompt};

let summary = ctx.agent("summarize", AgentStepConfig::new("Summarize the page")).await?;
let prompt = Prompt::new()
    .text("Act on this summary:")
    .untrusted(&summary.text());
ctx.agent("act", AgentStepConfig::new(&prompt.to_string())).await?;

// A typed answer derives `AgentAnswer`; its free-text fields are `AgentText`.
#[derive(Deserialize, JsonSchema, AgentAnswer)]
struct Findings {
    title: AgentText,
    severity: u8,
}

let findings = ctx
    .agent("audit", AgentStepConfig::new("Audit the repo").output::<Findings>())
    .await?;
let prompt = Prompt::new().text("Fix this finding:").untrusted(&findings.title);

The T of .output::<T>() and the Output of an AgentRole implement AgentAnswer, which #[derive(AgentAnswer)] provides when every field does: numbers, bool, AgentText, Uuid, Decimal, collections and options of them, other derived types. A String or a serde_json::Value field fails to compile. AgentText serializes as a plain string with the schema of a String, so the stored output and the fixtures are the same.

{:?} of an AgentText prints AgentText(<N bytes>), never the text, and so does the {:?} of a type deriving Debug that holds one: format!("{:?}", answer) carries no agent text into a prompt.

The raw text stays reachable, explicitly, for what is not a prompt (a log, an artifact, a shell argument, a comment): .raw(), into_raw(), serde_json::to_string or to_value of an AgentText or of a typed answer, StepOutput::json::<String>(), the public output.output Value (and the {:?} of a StepOutput, which prints it), and output_schema_raw with json::<T>(). Never put their result in a prompt.

Sub-workflow steps

A child declares its input type with TypedWorkflow and the parent passes that type; see WorkflowHandler.

let child = ctx.workflow(&Collect, CollectInput { host: "db-1".into() }).await?;
println!("child run {}", child.run_id());

Reading the child output

A child publishes a typed result with ctx.set_output(&value) (see Run output); the parent reads it back with child.output::<T>():

#[derive(Serialize, Deserialize)]
struct Verdict {
    approved: bool,
}

let child = ctx.workflow(&Review, ReviewInput { mr: input.mr }).await?;
match child.output::<Verdict>()? {
    Some(verdict) if verdict.approved => println!("approved"),
    Some(_) => println!("changes requested"),
    None => println!("the reviewer gave no verdict"),
}

• Ok(None) when the child set no output.
• Err(EngineError::Serialization) when the output does not deserialize into T.
• The output is recorded in the parent’s Workflow step: a resumed parent reads the same value, without reading the child run again.
• A child tolerated with allow_failure keeps the output it set before failing.

Exclusive children

ctx.workflow_with takes WorkflowOptions. With a concurrency key, the child is created only if no non-terminal run holds the same key; otherwise no child is created and the step completes with the run that holds it:

let options = WorkflowOptions::new().concurrency_key(format!("issue:{}", input.issue));
match ctx.workflow_with(&FixIssue, FixInput { issue: input.issue }, options).await? {
    SubWorkflowOutcome::Completed(child) => println!("fixed in run {}", child.run_id()),
    SubWorkflowOutcome::Conflict(conflict) => {
        println!("issue already handled by run {}", conflict.run_id());
    }
}

The conflict is recorded as the step output ({"concurrency_conflict": {"key", "run_id"}}) and replayed as-is when the parent resumes, even if the key was released in between: the branch taken by the handler stays stable. ctx.workflow and ctx.workflow_dyn set no key and keep returning the child output directly.

To tolerate a failed child, use ctx.workflow_with(&Collect, input, WorkflowOptions::new().allow_failure()): the child run is still marked failed, but the step completes with a Completed outcome whose status() is Failed (or Cancelled) and child.error() carrying the message, and the parent ends as Warning. A suspension is never tolerated. A resume replays the step and creates no new child.

let outcome = ctx
    .workflow_with(&Collect, CollectInput { host: "db-1".into() }, WorkflowOptions::new().allow_failure())
    .await?;
if let Some(child) = outcome.output().filter(|c| c.status() == RunStatus::Failed) {
    println!("collect failed: {:?}", child.error());
}

Step names and labels

WorkflowOptions names the parent’s step and labels the child run:

let options = WorkflowOptions::new()
    .step_name(format!("delegate-{tour}-{agent}"))
    .label("team.example/tour", tour.to_string())
    .label("team.example/agent", agent.as_str());
ctx.workflow_with(&Delegate, DelegateInput { agent: agent.clone() }, options).await?;

• The step is named after the child handler by default. The name identifies the step in the run timeline and on resume: a handler that delegates to the same child several times gives each call its own name. An empty or blank name panics.
• The child inherits the labels of its parent run; a label set here replaces an inherited one with the same key. ironflow.io/* keys are set by the engine (ironflow.io/parent-run-id, ironflow.io/root-run-id) and panic.

Children in parallel

ctx.workflows_with(&Child, children) takes one (input, WorkflowOptions) per child and runs the whole wave concurrently, with a Workflow step per child at the same position. The cost cap of the root holds for the whole wave. See Parallel Execution.

Suspension of a child

A child may suspend like any run: an approval, a human input, a signal wait or a delay. The whole chain is then suspended, durably:

• the child run keeps its own status (AwaitingApproval or Sleeping) and owns the wake-up (scheduled_at of a delay or a signal deadline);
• the parent’s Workflow step stays Running and records the child run id;
• the parent, and every ancestor up to the root, takes the same status without any scheduled_at.

The child run carries two labels, ironflow.io/parent-run-id and ironflow.io/root-run-id, so it can be listed and resolved like a top-level run: answer its human input, approve its gate or send it a signal through the usual API routes. Resuming the child resumes the root run: it replays its completed steps (a child that already completed is not run again) and re-enters the same child run, which picks up where it stopped. Rejecting an approval gate inside the child fails the child and every ancestor.

Several children of a workflows_with wave may be suspended at once. Each one publishes its own event (approval requested, signal awaited) and resumes the root on its own. A child resolved while a sibling already resumed the root waits for the root to stop, then resumes it again unless the root re-entered it in the meantime: the root never runs twice at once and no child is failed for it.

Cancellation and stopped parents

A child only runs inside its parent’s execution, so no child outlives its parent:

• POST /api/v1/runs/:id/cancel cancels the run and every sub-workflow run below it that has not finished. Their open steps are closed and their concurrency keys released; the response lists them in cancelled_descendants. GET /api/v1/runs/:id gives the count beforehand in active_descendant_count.
• A parent that fails, schedules a retry (a timeout, a panic, a handler error) or runs out of lease recoveries cancels the children its attempt left running. A retry starts new children.
• Cancelling a child directly ends the parent’s Workflow step with the child’s Cancelled status. A suspended chain is woken for it. Without allow_failure the step and the parent fail with ChildRunCancelled, never retried; with it the step completes with a Cancelled output and the parent ends as Warning.
• Pausing a root run pauses its children with it, and resuming it resumes them. A child is never paused or resumed on its own (see Pausing runs and workflows).

Approval steps

ctx.approval("prod-gate", ApprovalConfig::new("Deploy to production?")).await?;

The run suspends on AwaitingApproval until a human answers. A gate can also carry an SLA: a deadline persisted on the step and an escalation policy applied when it expires.

use std::time::Duration;

ctx.approval(
    "prod-gate",
    ApprovalConfig::new("Deploy to production?")
        .assigned_to(Assignee::group("release-managers"))
        .with_deadline(Duration::from_secs(3600))
        .on_timeout(EscalationPolicy::AutoReject),
).await?;

See Approval Gates for the full list of escalation policies, where the remaining time surfaces, and how to require several approvers.

Human input steps

#[derive(Deserialize, JsonSchema)]
struct Answers {
    answers: Vec<String>,
}

let answers: Answers = ctx
    .human_input("clarify", HumanInputConfig::new("Answer the clarification questions"))
    .await?;

The run suspends on AwaitingApproval until a person posts an answer matching the JSON schema of Answers to POST /api/v1/runs/:id/steps/:step_id/input. The config takes the same deadline, escalation, assignee and approvers options as an approval gate. A rejected input reaches the handler as EngineError::HumanInputRejected. See Human Input.

Decision steps

A decision step asks a DecisionProvider (System One / Jev) the questions declared by a struct and returns that struct, filled with the answers.

#[derive(DecisionChoice)]
enum Team { Billing, Technical }

#[derive(DecisionAnswers)]
struct Routing {
    #[choice("Which team?")]
    team: Team,
}

let routing = ctx.decision(
    "triage",
    DecisionConfig::new("Payouts have been failing for 3 days")
        .answers::<Routing>()
        .escalate_below(0.7),
).await?;
if let Team::Billing = routing.team { /* .. */ }

Below the escalate_below confidence threshold, the run suspends for human approval; on resume the stored answers are replayed as-is. See Decisions.

Conditions

A handler branches with plain Rust if/else. That is invisible to the execution planner, which is why two helpers exist to declare a branch explicitly.

ctx.when(label, predicate) deserializes the run input into the type the closure takes and evaluates the predicate on it. It returns the predicate’s value, and the planner records it as evaluated together with the label you gave the branch. The label is a name for the operator, never parsed:

#[derive(Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
enum Env { Prod, Staging }

#[derive(Deserialize)]
struct DeployInput { env: Env }

if ctx.when("production run", |i: &DeployInput| i.env == Env::Prod).await? {
    ctx.shell("deploy-prod", ShellConfig::new("./deploy prod")).await?;
} else {
    ctx.skip("deploy-prod", "not a production run").await?;
}

A payload that does not match the type ("env": "prd") fails with EngineError::Serialization instead of silently taking the else branch.

ctx.when_dynamic(label, value) declares a branch whose value comes from a previous step’s output. It returns value unchanged; the planner records the condition as unevaluable, because step outputs are synthetic while planning:

let build = ctx.shell("build", ShellConfig::new("cargo build")).await?;
if ctx.when_dynamic("build succeeded", build.is_success()) {
    ctx.shell("deploy", ShellConfig::new("./deploy")).await?;
}

Both helpers are optional: a plain if still runs exactly the same way. They only make the branch legible to whoever reads the plan.

Step status lifecycle

Steps follow this state machine:

stateDiagram-v2
    [*] --> Pending
    Pending --> Running
    Running --> Completed
    Running --> Failed
    Pending --> Skipped

Every step transition is recorded. Failed steps report their error in the step output.

Artifacts

Steps produce text and JSON outputs by default. When a step produces files, declare them: they are stored with their size, MIME type and SHA-256, then made available to later steps and to the dashboard.

#![allow(unused)]
fn main() {
use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::error::EngineError;

async fn release(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
    // Produce: every glob match becomes an artifact named after the file.
    let build = ctx
        .shell(
            "build",
            ShellConfig::new("cargo build --release && ./gen-report")
                .dir("/app")
                .output("target/report.html")
                .output("target/*.log"),
        )
        .await?;

    // The producing step hands out the handle; a name it did not declare fails here.
    let report = build.artifact("report.html")?;

    // Consume: the file is written into the working directory before the command runs.
    ctx.shell(
        "publish",
        ShellConfig::new("./publish report.html").dir("/app").input(&report),
    )
    .await?;
    Ok(())
}
}

ctx.put_artifact(&step_output, ..) and ctx.get_artifact(&handle) cover custom operations and agent steps, which have no working directory to collect files from.

Download one with GET /api/v1/runs/{id}/steps/{step_id}/artifacts/{name}, or from the artifact list on the step in the dashboard.

Enabling artifacts

Artifacts stay off until ARTIFACTS_DIR is set on the API server. Until then, the artifact routes answer 501 and a step that declares one fails with an explicit error; everything else works. ARTIFACT_MAX_BYTES caps the size of one file (100 MiB by default).

LocalBlobStore writes to the API server’s filesystem, so a deployment with several API replicas needs a shared volume. Workers hold no storage credential: they stream artifact bytes through the internal API.

Operations

An Operation is the extensibility mechanism for custom step types. Operations let you integrate external services (GitLab, Slack, any HTTP API) as tracked steps.

The trait

use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use serde_json::Value;

pub trait Operation: Send + Sync {
    fn kind(&self) -> &str;
    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>>;
    fn input(&self) -> Option<Value> { None }
}

• kind() returns a short identifier (e.g. "slack", "gitlab") stored in the database
• execute() runs the operation and returns JSON output
• input() optionally returns structured input for observability

Using an operation in a workflow

Operations are invoked via ctx.operation(), which takes a step name and a reference to the operation:

use ironflow_engine::context::WorkflowContext;

let slack = SlackNotify::new(&webhook_url);
ctx.operation("notify-team", &slack).await?;

Implementing an operation

use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use ironflow_engine::operation::Operation;
use serde_json::{Value, json};

pub struct SlackNotify {
    webhook_url: String,
    message: String,
}

impl Operation for SlackNotify {
    fn kind(&self) -> &str {
        "slack-notify"
    }

    fn input(&self) -> Option<Value> {
        Some(json!({ "message": self.message }))
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>> {
        Box::pin(async move {
            // Send to Slack webhook using self.webhook_url
            Ok(json!({ "ok": true }))
        })
    }
}

Built-in vs custom

Built-in step types (Shell, Http, Agent, Approval) have dedicated methods on WorkflowContext. Operations are for everything else – they give you a typed extension point without modifying the engine.

Pre-built ops crates

Ironflow ships with 13 ready-to-use ops crates under ops/ for common services: GitLab, Slack, Kubernetes, Docker, Helm, PostgreSQL, S3, Grafana, Loki, Mimir, Tempo, Git, and shared helpers. Each provides typed operations that plug directly into ctx.operation().

See Using Pre-built Ops Crates for the full catalog and usage examples, or Writing an Operation to implement your own from scratch.

Engine & Worker

Engine

The Engine is the in-memory registry that maps workflow names to handlers and orchestrates run execution. It holds references to the Store (persistence), the Provider (agent backends), and the event publisher.

let mut engine = Engine::new(store, provider);
engine.register(Box::new(MyWorkflow))?;

The Engine is used by both the API server (for metadata and describe endpoints) and the Worker (for execution).

Before running an agent step, the engine stamps the ironflow.io/run-id, ironflow.io/root-run-id and ironflow.io/step pod labels on its config (the step name is sanitized into a valid label value), so the Kubernetes providers can tag the pod and clean up a previous attempt of the same step on retry. The root run is the run itself, or the top-level run inside a sub-workflow (ctx.root_run_id()).

Before every execution of a run (Engine::execute_handler_run, the first one included, and Engine::resume_run after a gate under ExecutionMode::Local), the engine calls AgentProvider::release_run with the run id. The default does nothing; K8sEphemeralProvider deletes the pods left by a dead attempt of the run or of its sub-workflows, and waits until they are gone. A failed release fails the execution with a replayable error: the run goes to Retrying while it has retries left.

Execution mode

A run suspended on an approval, a human input or an escalation resumes once the gate is resolved. Engine::with_execution_mode decides where that happens:

• ExecutionMode::Local (default): the API process moves the run to Running and calls Engine::resume_run itself. Use it for single-process deployments where the API also registers the handlers. TestEngine always resumes this way.
• ExecutionMode::Workers: the API moves the run back to Pending. A worker claims it through pick_next_pending and finishes it with Engine::execute_handler_run, replaying the steps that already completed. Use it when the API runs without the workspace, tools or handlers the workflow needs.

The server generated by /ironflow setup and examples/server use ExecutionMode::Workers. A split server/worker deployment must do the same: under Local, the server resumes woken runs itself and fails on the first step that needs the worker’s agent or tools.

let engine = Engine::new(store, provider)
    .with_execution_mode(ExecutionMode::Workers);

Worker

A Worker is a background process that:

1. Waits for a free execution slot (concurrency)
2. Polls the API for a pending run and acquires a lease on it
3. Executes the workflow handler via the Engine
4. Refreshes the lease periodically during execution
5. Reports the result back to the API

A saturated worker does not poll: a run is only claimed once a slot can execute it, so its lease never expires while it waits.

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .concurrency(2)
    .poll_interval(Duration::from_secs(2))
    .register(Box::new(MyWorkflow))
    .build()?;

worker.run().await?;

Workflows that use ctx.decision(...) need a decision provider on the worker too: .decision_provider(Arc::new(TypeSafeProvider::new(api_key))). See Decisions.

When Provider Accounts exist, the worker picks one for every agent step and injects its credential; WorkerBuilder::account_strategy chooses how. When every account is limited the run sleeps until the next reset, up to WorkerBuilder::max_capacity_wait (6 hours by default), and the step runs again when it wakes.

A step or run update sent to the API is checked against MAX_API_BODY_BYTES (2 MiB, the API body limit) before it leaves the worker: an oversized output fails the run with a payload too large error naming the size and the limit, and an HTTP 413 from the API reports the same error. Network errors and 5xx responses are retried after 1s, 2s and 4s (4 attempts); other 4xx responses fail at once.

Metrics

With the prometheus feature, metrics_addr serves the worker metrics in the Prometheus text format on GET /metrics. run installs the process-global recorder and fails if one is already installed. Without it, nothing listens.

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .metrics_addr("0.0.0.0:9100".parse()?)
    .build()?;

The library does not read an environment variable: the binary decides where the address comes from.

Concurrency groups

Worker concurrency caps one process. To cap the runs that share an external resource across every worker (a repository, a tenant, an environment), give the run concurrency groups when it is created:

let options = EnqueueOptions {
    concurrency_limits: vec![ConcurrencyLimit::new("repo:acme", 2)],
    ..Default::default()
};
engine
    .enqueue_handler_with_options("fix-issue", TriggerKind::Api, payload, options)
    .await?;

POST /api/v1/runs takes the same list as concurrency_limits. When a worker polls, the store only hands out a run whose groups all count fewer running root runs than the run’s limit. A blocked run stays Pending and the worker takes the next free one, so one saturated group never stalls the others. Sub-workflow runs execute inside their parent’s slot and are not counted. The ironflow_worker_queue_blocked_runs{group} gauge reports the pending runs each saturated group holds back.

Run priority

Every run carries a queue priority, an integer from -100 to 100 (default 0). When a worker polls, the store hands out the pending run with the highest priority first, then the oldest among equal priorities. A run whose scheduled_at is still in the future is not eligible, whatever its priority, and concurrency groups still apply: a blocked high priority run lets the next eligible one through.

A workflow declares the priority of its runs; the creation request can override it:

impl WorkflowHandler for Hotfix {
    fn priority(&self) -> i16 {
        50
    }
    // ...
}

let options = EnqueueOptions {
    priority: Some(80),
    ..Default::default()
};
engine
    .enqueue_handler_with_options("hotfix", TriggerKind::Api, payload, options)
    .await?;

A handler priority outside the range is clamped; an explicit one is refused with EngineError::InvalidPriority. POST /api/v1/runs takes priority in the body (400 when out of range), GET /api/v1/runs?priority=50 lists the runs of one priority, and a schedule gives its own priority to every run it creates. A retry or a replay keeps the priority of the original run, and a sub-workflow takes WorkflowOptions::priority.

Priority only orders the queue:

• No preemption. A running run is never paused or evicted for a higher priority one; it keeps its worker slot until it finishes.
• No aging. A low priority run does not move up while it waits. A steady flow of higher priority runs can delay it indefinitely, so keep negative priorities for work that can wait.

Routing runs to workers

A worker only takes the runs it can execute. When it polls, it sends the workflows it registered and the tags it carries; the store hands out the oldest pending run whose workflow is in that list and whose required tags are all carried by the worker. Other runs stay Pending for a better-suited worker, and the next one in the queue goes ahead.

Tags describe the host: hardware, region, network access.

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .register(Box::new(Transcode))
    .tags(["gpu", "region:eu"])
    .build()?;

A run requires the tags its handler declares with required_worker_tags, plus the ones given at creation: EnqueueOptions::worker_tags, worker_tags in POST /api/v1/runs, ironflow-cli run create --worker-tag gpu or the worker_tags argument of the MCP create_run tool. A tag is 1 to 64 ASCII letters, digits or - _ . : / =, at most 32 per worker or run; build() rejects an invalid one. Retrying a run keeps its tags; replaying it adds the current handler’s tags to the original ones.

A worker that sends no capabilities (released before this routing) still takes every run, so workers can be upgraded one at a time. The ironflow_worker_queue_depth gauge of a worker counts only the runs it can take.

A run no worker can take waits in silence. While a run is pending or retrying, GET /api/v1/runs/{id} returns worker_routing: the workers seen recently by this API process and how many of them could take the run. The dashboard and ironflow-cli run get warn when none was seen, or when none is eligible.

Inside a worker, a ctx.workflow(..) step refuses a child whose required tags the worker does not carry, since the child would run in its parent’s slot on the wrong host.

Lease & Reaper

Workers hold a time-limited lease on each run they execute. If a worker crashes or is evicted, the lease expires and the Reaper (a background task in the API server) detects the orphaned run and requeues it.

A worker refreshes its lease every 30 seconds and the lease lasts 90 seconds, so three missed refreshes make the run recoverable. A worker that cannot refresh its lease (the API took the run away, or stayed unreachable longer than the lease) abandons the run instead of executing it twice. Every 60 seconds the Reaper recovers at most 100 expired runs.

AppState::spawn_background_tasks starts the Reaper; a custom server that does not call it must start one itself, or expired runs stay Running. Both sides can be tuned:

let reaper = Reaper::new(store, engine)
    .interval(Duration::from_secs(30))
    .batch_size(50);

let worker = WorkerBuilder::new(api_url, token)
    .worker_id("worker-eu-west-1a") // default: worker-<uuid>, new at every start
    .lease_ttl(Duration::from_secs(120))
    .lease_refresh_interval(Duration::from_secs(30))
    .build()?;

A sub-workflow child that was suspended (human input, signal, delay) is picked like any run, but the worker resumes its root run instead. The lease follows: the root takes the child’s lease when it goes back to Running, the child releases it, and the worker keeps refreshing the root until it finishes. A root resumed this way is recovered by the Reaper like any other run.

Resuming after a lost lease

A requeued run resumes where its worker stopped, in the same attempt:

• Steps that completed, were skipped, or got their approval or human input are replayed from the store. They are not executed again, and nobody is asked twice.
• A step that was running when the worker died is marked Failed with interrupted: worker lease lost. The next worker executes it again at the same position, and the interrupted record stays in the step history. Pending and AwaitingApproval steps are left as they are.
• An interrupted ctx.workflow step re-enters the child run it had started instead of starting a new one. The child’s running steps are interrupted the same way, and its finished steps are replayed.
• An interrupted agent step resumes the Claude Code session it was running in, with a resume prompt, instead of starting the agent from scratch. See Resuming an interrupted agent step.

The interrupted step really runs twice, so make it idempotent: a deploy, a payment or a notification must tolerate a second call.

Lease recoveries are counted in lease_recoveries, apart from retry_count. The counter is exposed in the run API, the SDK, ironflow run get and the dashboard run page. A run can be recovered max_retries times. One more lost lease fails the run with worker lease expired, and its open steps are failed with the same error; the sub-workflow runs it left running are cancelled. A requeued run keeps its children: it re-enters them when it resumes. A handler failure still uses retry_count and starts a new attempt that replays nothing. A run whose handler changed to an incompatible version since it was created replays nothing either and fails with HANDLER_VERSION_MISMATCH.

Waker

Runs paused in Sleeping (a ctx.delay step, a ctx.wait_for_signal step waiting for its signal, or an agent step waiting for provider capacity) carry their wake-up time in scheduled_at. The Waker, a background task of the API server, claims every due run every 10 seconds and moves it back to Pending exactly once, even with several API instances. Under ExecutionMode::Local the API then resumes the run in-process; under ExecutionMode::Workers a worker picks it up. A delivered signal wakes its runs right away, without waiting for the Waker.

Pausing runs and workflows

An administrator can hold a run, or a whole workflow, without cancelling anything.

A run. POST /api/v1/runs/{id}/pause (Engine::pause_run) moves a root run and every active sub-workflow run below it to Paused. Each run keeps the state it was paused from in resume_status:

• a queued (Pending, Retrying) run is no longer picked by a worker;
• a Sleeping run is no longer woken by the Waker;
• a Running run has its step in flight interrupted at once: the step is marked Failed with interrupted: worker lease lost and no further step is started. A worker loses the lease of the run, and the reaper never touches a paused run;
• an approval, a human input or a signal the run waits for can still be resolved while it is paused. The decision is recorded and only changes the state the run resumes to; a rejection fails the run;
• the SLA deadline of an approval gate keeps running during the pause. When it expires, the escalator applies the gate’s policy as usual: AutoApprove records the approval and the run resumes to Pending, AutoReject fails the run, Notify and Escalate re-arm the timer. The run stays Paused until an operator resumes it, and a deadline that expired during the pause has already been escalated by then.

POST /api/v1/runs/{id}/resume (Engine::resume_paused_run) puts every run back in the state it was paused from. A run paused while it executed goes back to Pending and replays: finished steps are skipped and the interrupted step is executed again, like a run resumed after a lost lease. A Sleeping run whose deadline passed during the pause goes back to Pending. A run resumed to AwaitingApproval keeps the gate’s current deadline. A paused run can also be cancelled.

Under ExecutionMode::Local, a run resumed while its previous execution is still inside a step is not executed twice: that execution carries on, and the background restart waits for it before deciding whether anything is left to replay.

Only a root run is paused or resumed: the API answers 400 for a sub-workflow run, a run already finished, or (on resume) a run that is not paused.

A workflow. POST /api/v1/workflows/{name}/pause (Engine::pause_workflow) records a pause for the workflow: runs are still created, but workers leave them queued until POST /api/v1/workflows/{name}/resume. Runs already executing are not affected; pause them one by one. Under ExecutionMode::Local a run created while the workflow is paused stays Pending and is started by resume_workflow. GET /api/v1/workflows and GET /api/v1/workflows/{name} report paused_at while the workflow is paused. Every pause publishes a workflow_paused event, and a resume of a paused workflow a workflow_resumed event, both carrying the user_id of the caller (Engine::resume_workflow takes it as its second argument, None outside a user context); the audit log records them.

The CLI exposes the same actions as ironflow-cli run pause|resume <id> and ironflow-cli workflow pause|resume <name>, the MCP server as the pause_run, resume_run, pause_workflow and resume_workflow tools.

Scaling

Workers are stateless. Add more workers to increase throughput. Each worker polls independently – no coordination is needed beyond the API server.

Provider Accounts

A Provider Account is an account at an AI provider: in v1, a Claude Pro/Max subscription. Each account has a credential and usage limits (windows such as the 5 hour and 7 day windows of a subscription). Admins manage accounts live from the dashboard (Settings > Accounts), the REST API (/api/v1/provider-accounts), the CLI (ironflow accounts) and the MCP server.

When at least one account of the right kind exists, the worker picks one for every agent step, injects its credential into the Claude CLI process, and records the usage windows the CLI reports. With no account, agent steps run with the worker’s own environment, exactly as before.

The claude_subscription kind

Run claude setup-token on a machine logged into the subscription and paste the sk-ant-oat01-... token. Before storing anything, the server checks the format and then sends a one-token request to the Anthropic API:

• a malformed or rejected token is refused with 422, and nothing is stored;
• a rate-limited token (429) is stored and shown as limited until its window resets;
• an unreachable provider gives 502.

Where the credential lives

The credential is stored as the system secret accounts/<id>/credential, encrypted like every other secret. That namespace is hidden from the Secrets page and refused by the Secrets API. No response, log, audit entry or event carries the token.

Injection per transport

The token never appears on a command line. The worker forces the CLI into stream-json mode so it can read the rate_limit_event lines that report the windows.

Selection

The worker keeps only available accounts: enabled, token not rejected, not expired, no applicable window rejected until its reset, and under max_concurrency. A window scoped to a model family (for example the Opus 7 day window) only blocks steps using that family. A strategy then picks one:

Choose it on the worker:

use ironflow_core::account_strategy::Priority;

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .account_strategy(Arc::new(Priority))
    .build()?;

Targeting an account or a pool

By default a step may use any account of the kind. A step can narrow that:

use std::time::Duration;

// Only accounts tagged `team`.
let review = AgentStepConfig::new("Review the diff").account_pool("team");
// Exactly this account: never another one, never the worker's own token.
let audit = AgentStepConfig::new("Audit the release").account("perso-max");

A name that matches no account fails the step with AgentError::AccountNotFound. A pool with no account fails it with AgentError::NoCapacity.

When every account is limited

If the CLI reports a rejected window while a step runs on an account, the worker records it and runs the step again on the next available account. A step pinned with .account(..) has no other account and does not fail over.

When no targeted account is available, the run is not failed: it goes Sleeping until the earliest window reset, and the step runs again from the start when the run wakes. The run shows the provider kind it waits for (capacity_wait_kind in the API, a line in ironflow-cli run get <id>, a note in the dashboard) and its wake-up time in scheduled_at. Adding an account of that kind, re-enabling one or renewing its token wakes the run right away. An account that is only saturated (max_concurrency reached) is tried again after a minute.

The wait is bounded, counting every time the same step was parked: 6 hours by default. A reset further away fails the step with AgentError::NoCapacity, which carries the next reset time. Set the bound per step or on the worker; the step wins:

// Fail fast instead of waiting.
let urgent = AgentStepConfig::new("Triage the incident").fail_fast_on_capacity();

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .max_capacity_wait(Duration::from_secs(2 * 3600))
    .build()?;

A step marked allow_failure() still sleeps: waiting for capacity is not a failure. With no account at all, a rate limit reported on the worker’s own token puts the run to sleep the same way.

CLI

claude setup-token | ironflow accounts add perso-max --token-stdin --tag perso --priority 10
ironflow accounts list
ironflow accounts usage perso-max
ironflow accounts update perso-max --max-concurrency 2
ironflow accounts test perso-max
ironflow accounts remove perso-max --yes

Usage history is kept PROVIDER_ACCOUNT_USAGE_RETENTION_DAYS days (30 by default).

Approval Gates

An approval gate pauses a workflow run until a human approves or rejects it. This enables human-in-the-loop workflows like deploy pipelines where production deploys require sign-off.

How it works

1. The handler calls ctx.approval() with a prompt message
2. The run transitions to AwaitingApproval
3. The worker releases the run and moves on to other work
4. A human calls POST /api/v1/runs/:id/approve or POST /api/v1/runs/:id/reject
5. On approval, the run resumes. Under ExecutionMode::Workers it is requeued to Pending: a worker picks it up, replays completed steps from cache, skips the approved gate, and continues execution. Under ExecutionMode::Local (the default) the API process resumes it the same way itself
6. On rejection, the run transitions to Failed

If the gate carries an SLA deadline and nobody answers in time, step 4 is performed by the server instead of a human – see SLA timers below.

Example

//! Deploy workflow with human approval gate before production.

use std::time::Duration;

use ironflow_engine::config::{
    ApprovalConfig, Assignee, EscalationPolicy, NotificationTarget, ShellConfig,
};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};

/// Deploy pipeline that requires human approval before shipping to production.
///
/// 1. **build** -- compile the project
/// 2. **test** -- run the test suite
/// 3. **deploy-staging** -- deploy to staging environment
/// 4. **approval gate** -- pause and wait for human approval
/// 5. **deploy-production** -- resumes after approval via step replay
///
/// After approval (POST /api/v1/runs/:id/approve), the engine
/// re-executes the handler: completed steps return cached output,
/// the approved gate is skipped, and execution continues with
/// deploy-production.
///
/// If the approval is rejected, the run transitions to `Failed`.
/// If cancelled, the run transitions to `Cancelled`.
///
/// The gate carries a one-hour SLA. If nobody answers in time, the escalation
/// chain runs one policy per expiry: the first hour pings a webhook and keeps
/// waiting, the second gives up and fails the run. The deadline lives in the
/// database, so it survives an API or worker restart.
pub struct DeployApproval;

impl WorkflowHandler for DeployApproval {
    fn name(&self) -> &str {
        "deploy-approval"
    }

    fn description(&self) -> &str {
        "Deploy pipeline with human approval gate before production. \
         Demonstrates ctx.approval() for human-in-the-loop workflows."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("deploy_approval.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            // Step 1: Build
            ctx.shell(
                "build",
                ShellConfig::new("echo 'Compiling...' && sleep 0.2 && echo 'Build OK'"),
            )
            .await?;

            // Step 2: Test
            ctx.shell(
                "test",
                ShellConfig::new("echo 'Running tests...' && sleep 0.3 && echo '87 tests passed'"),
            )
            .await?;

            // Step 3: Deploy to staging
            ctx.shell(
                "deploy-staging",
                ShellConfig::new(
                    "echo 'Deploying to staging...' && sleep 0.2 && echo 'Staging live'",
                ),
            )
            .await?;

            // Step 4: Human approval gate
            // The run pauses here and transitions to AwaitingApproval.
            // A human must call POST /api/v1/runs/:id/approve to continue,
            // or POST /api/v1/runs/:id/reject to fail the run.
            ctx.approval(
                "prod-approval",
                ApprovalConfig::new("Staging looks good. Deploy to production?")
                    .assigned_to(Assignee::group("release-managers"))
                    .with_deadline(Duration::from_secs(3600))
                    .on_timeout(EscalationPolicy::Chain(vec![
                        // After 1 h without an answer: warn, keep waiting.
                        EscalationPolicy::Notify(vec![NotificationTarget::Webhook {
                            url: "https://example.com/hooks/deploy-sla".to_string(),
                        }]),
                        // After 2 h: give up rather than ship unreviewed.
                        EscalationPolicy::AutoReject,
                    ])),
            )
            .await?;

            // Step 5: Deploy to production (only reached after approval)
            ctx.shell(
                "deploy-production",
                ShellConfig::new(
                    "echo 'Deploying to production...' && sleep 0.3 && echo 'Production live'",
                ),
            )
            .await?;

            Ok(())
        })
    }
}

Configuration

use std::time::Duration;

ApprovalConfig::new("Deploy to production?")
    .assigned_to(Assignee::group("release-managers")) // Who is expected to answer
    .with_deadline(Duration::from_secs(3600))         // SLA: one hour to answer
    .on_timeout(EscalationPolicy::AutoReject)         // What happens when it expires

assigned_to takes an Assignee – Assignee::user("alice") or Assignee::group("release-managers"). It drives notification routing and audit, and it decides who may resolve the gate:

• an admin resolves any gate;
• a gate assigned to a user is resolved by that user, admin or not, and by whoever holds an active delegation from them;
• a gate assigned to a group, or to nobody, is admin-only.

The assignee is matched to the caller by user ID, so an API key resolves its owner’s gates whatever the key is named.

A user assignee is bound to the account holding that username when the gate opens, or when it is reassigned. If no account has that name at that moment, the gate is admin-only, and an account created later under that name cannot resolve it. Create the assignee’s account before the run reaches the gate.

Everything past the message is optional. Without a deadline, the run waits indefinitely.

SLA timers

with_deadline (or with_deadline_secs) arms a timer on the approval step. The deadline is stored in the database next to the step, not in memory, so it survives an API or worker restart: a fresh process picks the expired gate up on its next pass. The API server checks for expired gates every 30 seconds.

The timer is cleared the moment the gate resolves – approved, rejected, or escalated – so a gate is never escalated after a human answered it. A deadline fires at most once, even with several API instances running.

The deadline keeps running while the run is paused: the policy fires during the pause and only changes the state the run resumes to.

with_timeout_seconds is the legacy spelling: it is now enforced, as a deadline with an implicit AutoReject policy. Setting both keeps the explicit with_deadline.

Escalation policies

on_timeout takes an EscalationPolicy. Without one, an expired deadline auto-rejects.

Notify and Escalate do not resolve the gate: on their own, they fire again at every expiry until a human answers. Each firing writes an audit entry, so the loop is visible rather than silent. Wrap them in a Chain to advance one policy per expiry instead:

use std::time::Duration;

ApprovalConfig::new("Deploy to production?")
    .with_deadline(Duration::from_secs(3600))
    .on_timeout(EscalationPolicy::Chain(vec![
        // After 1 h: ping the on-call channel, keep waiting.
        EscalationPolicy::Notify(vec![NotificationTarget::Slack {
            webhook_url: slack_webhook_url,
            channel: "#deploys".to_string(),
        }]),
        // After 2 h: give up.
        EscalationPolicy::AutoReject,
    ]))

Once a chain runs out, the gate stays open with no timer and a warning is logged – it is never silently auto-rejected.

NotificationTarget is delivered as a plain HTTP POST, with the engine’s shared retry and backoff: Webhook { url } posts the escalation event as JSON, Slack { webhook_url, channel } posts a message to a Slack incoming webhook. A dead endpoint is logged and never blocks the timer reset.

Seeing the remaining time

The countdown surfaces in three places:

• the API: approval_seconds_remaining and approval_assignee on every step of GET /api/v1/runs/:id (clamped at 0, null without a deadline);
• the dashboard: a countdown badge on the gate in the run’s step list;
• the CLI: the SLA column of ironflow-cli run steps <id>, yellow in the last tenth of the window and red once expired.

Every escalation is also recorded in the audit log as an approval_escalated event carrying the stage, the policy, what it did, and why it fired.

Delegation and absence

An approval gate assigned to one person stops every run behind it the moment that person is away. A delegation hands their approval power to a colleague for a bounded window, without making anyone an admin and without reassigning the gates one by one.

A delegation records who grants it, who receives it, the window it is valid for, and an optional glob on the workflow name:

# Alice hands her deploy approvals to Bob for a week.
ironflow-cli delegation create <bob-user-id> \
    --until 2026-10-01T00:00:00Z \
    --workflow 'deploy-*'

# Everything Alice granted, plus everything she received (20 per page).
ironflow-cli delegation list --page 1 --per-page 20

# Back early.
ironflow-cli delegation delete <delegation-id>

The same three endpoints back the CLI:

What a delegation covers

Only a gate assigned to an individual can be delegated:

ApprovalConfig::new("Deploy to production?")
    .assigned_to(Assignee::user("alice")) // Bob can answer this through a delegation.

ApprovalConfig::new("Deploy to production?")
    .assigned_to(Assignee::group("release-managers")) // Admin-only; no single delegator.

A gate assigned to a group, or with no assignee at all, stays admin-only: there is no single person whose power could have been handed over.

The workflow_filter glob narrows a delegation to part of the catalogue. "deploy-*" covers deploy-prod but not cleanup; omitting it covers every workflow. A pattern that does not parse matches nothing, so a corrupted row can never widen someone’s reach.

Several delegations can be active at once – one per colleague, one per workflow family, or from several delegators to the same person. The newest one that matches both the gate’s assignee and the run’s workflow wins.

Expiry

The window is half-open: a delegation is live at valid_from and already over at valid_until. There is no cleanup job. Expired and not-yet-started rows are filtered out every time delegations are read, so they can neither be listed nor used to approve – they are still reachable by ID, which is what makes an expired delegation revocable.

Audit

A delegated decision names both people. The approval_granted (or approval_rejected) audit entry reads:

{
  "type": "approval_granted",
  "run_id": "01932f...",
  "approved_by": "bob (delegated from alice)",
  "user_id": "01932e...",
  "at": "2026-09-22T10:15:00Z"
}

An admin, or the assignee resolving their own gate, is recorded under their own name alone. user_id is always the caller (bob here): the label names the delegation, the id names who clicked.

Requiring several approvers

A gate can require more than one approval, and restrict who may vote, depending on the run itself: a small payment needs one approver, a large one needs two people from finance. The handler decides in plain Rust, from its typed input and the outputs of earlier steps, and passes the result to requiring:

use ironflow_engine::config::{ApprovalConfig, Approvers};

let payment: Payment = ctx.input().await?;
let approvers = match payment.amount {
    a if a > 100_000 => Approvers::at_least(3)
        .from_groups(["finance", "board"])
        .because("amount > 100k"),
    a if a > 10_000 => Approvers::at_least(2)
        .from_groups(["finance"])
        .because("amount > 10k"),
    _ => Approvers::any(),
};
ctx.approval(
    "payment-gate",
    ApprovalConfig::new("Release the payment?").requiring(approvers),
).await?;

A typo in a field name or a comparison does not compile, and the compiler checks every branch of the match. A gate without requiring behaves as a single approval gate.

The approvers are stored on the step as an ApprovalRequirement (reason, required_approvers, approver_groups) when the gate opens. That record is the source of truth from then on: replaying or resuming the run never recomputes it, even if the handler would now compute other approvers. GET /api/v1/runs/:id exposes it on the step as approval_requirement, with the votes cast so far in approvals and the count needed in approvals_required; the dashboard shows it as an n/m approvals badge whose tooltip gives the reason.

Approvers::at_least(0) and a blank group name panic, so a broken gate fails when the workflow runs the builder, not when a human votes. A JSON config with required_approvers: 0 is rejected on deserialization.

Voting

• One vote per user. Votes are counted by user ID: an API key votes as its owner, and the same user approving twice gets 409 Conflict.
• An admin’s approval is one vote. Admins may always vote, even on a gate restricted to groups, but they do not override the count.
• A rejection vetoes. One rejection from anyone allowed to vote fails the run, even after partial approvals.
• Until the count is reached, POST /approve returns 200 with the run still awaiting_approval, the gate keeps its SLA timer, and the CLI prints Approval recorded; more approvals are required.

An EscalationPolicy::AutoApprove still resolves the gate outright, whatever the required count.

Approver groups

When the approvers list approver_groups, only members of at least one of those groups (and admins) may vote. The gate’s assignee and approval delegations are not consulted. A listed group without members leaves the gate to admins.

Group membership is managed by admins:

# Put alice in finance and legal (replaces her current groups).
ironflow user set-groups <alice-id> --group finance --group legal

# Show her groups.
ironflow user groups <alice-id>

# Remove her from every group.
ironflow user set-groups <alice-id>

The same operations are available as GET and PUT /api/v1/users/:id/groups. Group names are 1 to 64 characters from [A-Za-z0-9_.-], at most 50 per user.

Audit events

• approval_requested is published when the gate opens and carries the recorded requirement.
• approval_granted is published for every vote, with the step_id, approvals_received, approvals_required and the requirement. The gate resolves when approvals_received >= approvals_required.
• approval_rejected carries the step_id and the requirement. Rejecting a human input publishes it too, with the reason given (or input rejected by <caller>); a gate rejection has no reason.
• approval_granted and approval_rejected carry the user_id of the caller, also stored as the audit entry’s user_id: for an API key it is the key’s owner, while approved_by / rejected_by stay the display label (the key name, or bob (delegated from alice)). A decision of the SLA timer (system:timeout) has no user_id. Events recorded before this field existed read back with user_id: null.

{
  "type": "approval_granted",
  "run_id": "01932f...",
  "step_id": "01932f...",
  "approved_by": "alice",
  "user_id": "01932e...",
  "approvals_received": 1,
  "approvals_required": 2,
  "requirement": {
    "reason": "amount > 10k",
    "required_approvers": 2,
    "approver_groups": ["finance"]
  },
  "at": "2026-09-24T10:15:00Z"
}

Step replay

After an approval, the engine re-executes the handler from the beginning: in the API process under ExecutionMode::Local, on the worker that picks up the requeued run under ExecutionMode::Workers (see execution mode). Completed steps return their cached output immediately – they do not re-run. The approved gate is skipped, and execution resumes with the next step.

The gate is skipped only when its votes are recorded: one, or the required_approvers of its requirement. The approve route records them before it resumes the run, and so does TestEngine::resume in tests. A run moved back to Running without them (a direct store update, a sub-workflow cancelled under a root that waits on its own gate) is suspended again on the same gate: it stays AwaitingApproval and keeps its deadline. For a gate inside a sub-workflow, the vote is cast on the child run that owns the gate.

Human Input

A human input step pauses a workflow run until a person submits a typed answer. Where an approval gate asks “yes or no?”, a human input asks for data: answers to clarification questions, a choice, a value. The handler gets the answer back as a Rust type.

How it works

1. The handler calls ctx.human_input::<T>() with a message. T derives Deserialize and JsonSchema.
2. The engine records a step of kind human_input whose input holds the message and the JSON schema of T (under the key schema).
3. The step and the run move to AwaitingApproval, the same status as an approval gate, and an input_required event is published on the run’s event stream.
4. A person posts an answer to POST /api/v1/runs/:id/steps/:step_id/input. The API validates it against the stored schema, completes the step and resumes the run: in the API process under ExecutionMode::Local (the default), or by requeuing it to Pending for a worker under ExecutionMode::Workers (see execution mode).
5. The handler is replayed: completed steps come from cache and human_input returns the answer, deserialized into T.

Example

use ironflow_engine::prelude::*;
use schemars::JsonSchema;
use serde::Deserialize;

#[derive(Deserialize, JsonSchema)]
struct Answers {
    answers: Vec<String>,
}

let answers: Answers = ctx
    .human_input("clarify", HumanInputConfig::new("Answer the clarification questions"))
    .await?;

ctx.shell("plan", ShellConfig::new(&format!("./plan.sh {}", answers.answers.len())))
    .await?;

Configuration

HumanInputConfig reuses the approval gate machinery:

use std::time::Duration;

HumanInputConfig::new("Which environment should we target?")
    .assigned_to(Assignee::user("alice"))            // Who is expected to answer
    .requiring(Approvers::any().from_groups(["product"])) // Who may answer
    .with_deadline(Duration::from_secs(3600))         // SLA: one hour to answer
    .on_timeout(EscalationPolicy::AutoReject)         // What happens when it expires

Who may answer follows the approval rules: an admin, a member of the requiring groups, the assignee, or someone holding a delegation from the assignee. The person who answered is recorded on the step like an approval vote.

EscalationPolicy::AutoApprove has no meaning without a value: on_timeout(EscalationPolicy::AutoApprove) panics, and so does a Chain containing it. Notify, Escalate, AutoReject and Chain behave as for approval gates.

API

Answer the input with a body matching the schema:

curl -X POST "$IRONFLOW_URL/api/v1/runs/$RUN_ID/steps/$STEP_ID/input" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"answers": ["staging", "eu-west-1"]}'

• 200 returns the run, now running.

• 422 with code INVALID_INPUT when the body does not match the schema; each violation is listed in error.details.errors:

  {
    "error": {
      "code": "INVALID_INPUT",
      "message": "input does not match the expected schema",
      "details": { "errors": ["3 is not of type \"array\""] }
    }
  }
  

• 409 when the input was already answered or rejected.

• 400 when the step is not a human input, or the run is not waiting on it.

Refuse the input, with an optional reason:

curl -X POST "$IRONFLOW_URL/api/v1/runs/$RUN_ID/steps/$STEP_ID/reject" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"reason": "out of scope for this sprint"}'

POST /api/v1/runs/:id/approve refuses a run waiting on a human input with a 400: approving it would resume the handler without an answer.

The dashboard shows a form for every pending input on the run page. The CLI has ironflow-cli run input <run> <step> --value '{..}' (or --value-file) and ironflow-cli run reject-input <run> <step> --reason ..; the MCP server has the submit_input and reject_input tools.

Rejection

A rejected input does not fail the run by itself. The step is marked Rejected with the reason, the run resumes, and human_input returns EngineError::HumanInputRejected, so the handler decides what happens next:

match ctx.human_input::<Answers>("clarify", config).await {
    Ok(answers) => { /* use the answers */ }
    Err(EngineError::HumanInputRejected { reason, .. }) => {
        // The reason is free text typed by a person: an argument, not a command line.
        ctx.shell("notify", ShellConfig::exec("./notify.sh", &[&reason]))
            .await?;
    }
    Err(err) => return Err(err),
}

A handler that propagates the error fails the run; it is never retried automatically. The run-level POST /api/v1/runs/:id/reject still fails the run outright.

Replay and retries

• A run resumed without an answer suspends again on the same step; no new step is created.
• An answer given before an automatic retry is carried over to the next attempt: the person is not asked twice.
• If the handler changed and the stored answer no longer fits T, the run fails with a step configuration error.

Events

input_required is published on GET /api/v1/runs/:id/events when the step opens. It carries run_id, step_id, step_name, step_index, message and schema, everything a client needs to render a form.

Execution plans

Planning never suspends. The step is recorded with kind human_input, and T is built from {}: a type with #[serde(default)] lets the plan continue past the input. Otherwise the plan stops there with the reason human input '<name>' has no answer while planning.

Testing

TestEngine::with_mock_human_input answers every human input without waiting:

use ironflow_engine::testing::{HumanInputOutcome, TestEngine};
use serde_json::json;

let result = TestEngine::new()
    .with_handler(Clarify)
    .with_mock_human_input(|_name, _config| {
        HumanInputOutcome::Provided(json!({"answers": ["staging"]}))
    })
    .run(json!({}))
    .await?;

HumanInputOutcome::reject("reason") makes the handler receive EngineError::HumanInputRejected. Without the mock, the run ends in AwaitingApproval: write the answer on the step through the store, then call TestEngine::resume.

Signals

A signal is an external message sent to Ironflow to resume the runs waiting for it. It has a name (what happened, e.g. ci.pipeline_finished) and a key (which occurrence, e.g. a commit SHA). A run waits with ctx.wait_for_signal; a producer delivers with POST /api/v1/signals, the CLI, the MCP server or Engine::send_signal, and another workflow with ctx.send_signal.

Declaring a signal

A signal is a typed payload: a struct implementing Signal, which names it.

use ironflow_engine::signal::Signal;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, JsonSchema)]
struct PipelineFinished {
    status: String,
}

impl Signal for PipelineFinished {
    const NAME: &'static str = "ci.pipeline_finished";
}

Waiting for a signal

let finished = ctx
    .wait_for_signal::<PipelineFinished>("wait-ci", &sha, Duration::from_secs(3600))
    .await?;
match finished {
    Some(pipeline) if pipeline.status == "success" => { /* deploy */ }
    Some(_) => return Err(EngineError::StepConfig("CI failed".to_string())),
    None => return Err(EngineError::StepConfig("CI timed out".to_string())),
}

• The key is the occurrence: wait on a commit SHA, not on a merge request. A signal for an older push must never resume a run waiting for the newest one.
• Received early: a signal delivered after the run was created but before the step opened resolves the step at once, without suspending.
• Suspended: otherwise the run goes Sleeping until the deadline. A delivery whose payload matches the JSON schema of the type wakes it, and the handler receives Some(payload).
• Timeout: when the deadline passes first, the step completes as timed out and the handler receives None.
• Broadcast: every run waiting on the same (name, key) receives the signal.
• Invalid payload: a run whose schema the payload does not match keeps waiting and is listed under rejected in the response. The signal is still stored.
• Idempotency: a second delivery with the same idempotency_id returns duplicate: true and delivers nothing.

Woken runs resume in-process under ExecutionMode::Local, or go back to Pending for a worker under ExecutionMode::Workers. Timeouts are applied by the waker task of the API server (see Engine and worker).

Sending a signal

Sending from a workflow

A workflow sends a signal as a step of its own run, without holding any API key. A router started by a webhook can wake the run waiting for a human reply:

impl WorkflowHandler for Router {
    fn name(&self) -> &str { "route-gitlab-note" }

    fn sent_signals(&self) -> SentSignals {
        SentSignals::new().allow::<HumanReplied>()
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let note: GitlabNote = ctx.input().await?;
            let reply = HumanReplied { body: note.body };
            let delivery = ctx.send_signal("wake-agent", &reply, &note.discussion_id).await?;
            ctx.set_output(&delivery.resumed.len())?;
            Ok(())
        })
    }
}

• Declared: the handler lists the signal types it sends in sent_signals. Any other fails the step with SignalNotAllowed and sends nothing. The declaration keeps a workflow from sending a signal that is not its own by mistake; it is not a security boundary.
• Trusted input: whoever can start the sending workflow chooses the key and the payload it sends. Check the webhook that starts it.
• Timeline and audit: the step (signal_send) records the delivery (signal_id, duplicate, resumed, rejected). The signal_received audit entry carries the sending run and step (sender_run_id, sender_step_id).
• Sent once: a completed step is replayed without sending again, and a step interrupted by a lost worker is deduplicated. A retry of the run is a new attempt and sends again.
• On a worker: the worker delivers through the API with its worker token. Upgrade the API before the workers.

Example: wait for CI

The workflow pushes a commit and waits for its pipeline:

ctx.shell("push", ShellConfig::new("git push origin HEAD")).await?;
let finished = ctx
    .wait_for_signal::<PipelineFinished>("wait-ci", &sha, Duration::from_secs(3600))
    .await?;

The CI webhook handler filters the terminal pipeline statuses and forwards them, using the delivery ID of the webhook so a retried delivery is not counted twice:

if matches!(status.as_str(), "success" | "failed" | "canceled") {
    engine
        .send_signal(&PipelineFinished { status }, &sha, Some(&delivery_id))
        .await?;
}

Signals are kept SIGNAL_RETENTION_DAYS days (default 7) and then purged. The example server wires the variable into its RunPurger; a custom server builds it with RunPurger::from_config(store, &config).

Decisions

A decision is a typed machine verdict: classify, route, score, or answer yes/no, with a calibrated confidence instead of free text. It is the shape of TypeSafe AI’s System One model (Jev): for a structured verdict it is far faster and cheaper than routing a prompt through a conversational LLM.

Decisions use a dedicated abstraction rather than the agent one: there is no single prompt, no tools, and no streaming. A run wires a DecisionProvider independently of its agent provider.

Wiring a provider

use ironflow_core::providers::http::TypeSafeProvider; // feature "provider-typesafe"
use std::sync::Arc;

let engine = Engine::new(store, agent_provider)
    .with_decision_provider(Arc::new(TypeSafeProvider::new(api_key)));

When the workflow runs in a worker, wire the provider on the WorkerBuilder instead; every run the worker executes gets it:

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(agent_provider)
    .decision_provider(Arc::new(TypeSafeProvider::new(api_key)))
    .build()?;

Without a provider, a decision step fails with NoDecisionProvider. In tests, use RecordReplayDecisionProvider::replay(dir) to serve captured JSON fixtures with no network.

Through OpenRouter

OpenRouter serves the same System One wire contract on its Decisions endpoint, so the same provider reaches Jev with only the base URL and key changing. Use the openrouter constructor and select the OpenRouter model slug on the config (typesafe/jev-1.13, exposed as OPENROUTER_MODEL). OpenRouter requires this concrete versioned slug; the jev-latest alias returns 400 "Model does not exist":

use ironflow_core::providers::http::typesafe::OPENROUTER_MODEL;
use ironflow_core::providers::http::TypeSafeProvider;
use ironflow_engine::config::DecisionConfig;
use std::sync::Arc;

let engine = Engine::new(store, agent_provider)
    .with_decision_provider(Arc::new(TypeSafeProvider::openrouter(openrouter_key)));

let config = DecisionConfig::new(state).model(OPENROUTER_MODEL);

OpenRouter’s decisions route is on an alpha path that may move; override it with TypeSafeProvider::with_endpoint(url) if it relocates. OpenRouter also requires question instructions and criteria to be strings, which the derive’s string literals already satisfy.

The three question types

The questions are the fields of a struct deriving DecisionAnswers; the options of a choice are the unit variants of an enum deriving DecisionChoice. ctx.decision returns the struct itself.

use ironflow_engine::config::DecisionConfig;
use ironflow_engine::decision::{DecisionAnswers, DecisionChoice};

#[derive(DecisionChoice)]
enum Team {
    #[choice(description = "Payments, invoices, refunds")]
    Billing,
    Technical,
    Sales,
}

#[derive(DecisionAnswers)]
struct Triage {
    #[noul("Does this convey urgency?")]
    is_urgent: f64,
    #[choice("Which team?")]
    team: Team,
    #[score("How frustrated?", levels = ["Calm", "Frustrated", "Very angry"])]
    mood: f64,
}

let triage = ctx.decision(
    "triage",
    DecisionConfig::new("Payouts have been failing for 3 days")
        .answers::<Triage>()
        .escalate_below(0.7),
).await?;

match triage.team {
    Team::Billing => { /* .. */ }
    Team::Technical | Team::Sales => { /* .. */ }
}

A question is named after its field. An option is labelled with its variant name in snake_case; #[choice(rename = "..")] changes the label and #[choice(description = "..")] tells the model what the option means (doc comments are never sent). A field without a question, a score without levels or a field of the wrong type does not compile. An option the provider returns that is not a variant fails the step with DecisionError::UnknownChoice.

The options are fixed at compile time. choice and score answers carry a confidence; a noul answer reports only a probability p, and its confidence is derived as 2 * |p - 0.5| (a coin flip is 0, a certain yes/no is 1). Confidence drives escalation, below; it is not part of the typed answer.

Escalation

When escalate_below(threshold) is set and any answer’s confidence falls below it, the run suspends in AwaitingApproval, exactly like an approval gate. On resume the decision is not re-run: the stored answers are replayed as-is, so downstream routing stays deterministic across the suspend/resume boundary.

Cost

The provider reports token usage; the engine imputes the cost to the run’s budget like an agent step. For Jev, only input tokens are billed (output is unmetered).

Writing a Workflow

This guide walks through creating a workflow handler from scratch.

1. Define your input

If your workflow accepts input, define a struct with Deserialize and JsonSchema:

use schemars::JsonSchema;
use serde::Deserialize;

#[derive(Deserialize, JsonSchema)]
struct DeployInput {
    environment: String,
    version: String,
}

The JsonSchema derive lets the dashboard render a dynamic form for triggering the workflow.

2. Implement WorkflowHandler

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use serde_json::Value;

pub struct Deploy;

impl WorkflowHandler for Deploy {
    fn name(&self) -> &str {
        "deploy"
    }

    fn description(&self) -> &str {
        "Deploy a version to an environment"
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<DeployInput>())
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: DeployInput = ctx.input().await?;

            ctx.shell(
                "build",
                ShellConfig::exec("printf", &["Building %s\n", &input.version]),
            ).await?;

            ctx.shell(
                "deploy",
                ShellConfig::exec(
                    "printf",
                    &["Deploying %s to %s\n", &input.version, &input.environment],
                ),
            ).await?;

            Ok(())
        })
    }
}

The input comes from whoever triggers the run, so it never goes into a command line: ShellConfig::new hands its string to sh -c, and a quote or a ; in version would run arbitrary commands on the worker. ShellConfig::exec spawns the program directly and passes each argument as is. Keep ShellConfig::new for commands you write in full.

Branching is plain Rust if/else. When a branch depends on the run input, declare it with ctx.when("production run", |i: &DeployInput| i.environment == "production") so it shows up in the execution plan; use ctx.when_dynamic when the branch depends on a previous step’s output. Run ironflow-cli run plan <name> --input '{}' to see the steps your handler would create before triggering it.

An agent reused across workflows is a role rather than a config built in each handler: impl AgentRole for Reviewer, then ctx.agent("review", Reviewer.task(input)). See Reusable agent roles.

What an agent wrote reaches the next prompt only through Prompt::untrusted: see Agent output in a prompt.

3. Register in your handlers list

pub fn handlers() -> Vec<Box<dyn WorkflowHandler>> {
    vec![
        Box::new(Deploy),
        // ... other handlers
    ]
}

Both the server and the worker must register the same handlers. The recommended pattern is a shared handlers() function in a library crate.

4. Complete example

The greeting workflow in the examples directory demonstrates all features:

use std::collections::HashMap;

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

/// Input payload for the greeting workflow.
///
/// Derives [`JsonSchema`] so the dashboard can render a dynamic form.
#[derive(Deserialize, JsonSchema)]
struct GreetingInput {
    /// Person to greet.
    name: String,
    /// Greeting language (en, fr, es).
    #[serde(default = "default_language")]
    language: String,
    /// Number of times to repeat the greeting.
    #[serde(default = "default_repeat")]
    repeat: u32,
    /// Whether to output in uppercase.
    #[serde(default)]
    uppercase: bool,
}

fn default_language() -> String {
    "en".to_string()
}

fn default_repeat() -> u32 {
    1
}

pub struct Greeting;

impl WorkflowHandler for Greeting {
    fn name(&self) -> &str {
        "greeting"
    }

    fn category(&self) -> Option<&str> {
        Some("examples")
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<GreetingInput>())
    }

    fn default_labels(&self) -> HashMap<String, String> {
        HashMap::from([("project".to_string(), "ironflow".to_string())])
    }

    fn description(&self) -> &str {
        "A demo workflow that greets someone. \
         Shows how input_schema generates a dynamic form in the dashboard."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("greeting.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: GreetingInput = ctx.input().await?;

            let greeting = match input.language.as_str() {
                "fr" => format!("Bonjour, {} !", input.name),
                "es" => format!("Hola, {}!", input.name),
                _ => format!("Hello, {}!", input.name),
            };

            let mut message = (0..input.repeat)
                .map(|_| greeting.as_str())
                .collect::<Vec<_>>()
                .join("\n");

            if input.uppercase {
                message = message.to_uppercase();
            }

            // `name` is untrusted: pass it as an argument, never inside a
            // command line, so no shell ever parses it.
            ctx.shell("greet", ShellConfig::exec("printf", &["%s\n", &message]))
                .await?;

            Ok(())
        })
    }
}

Writing an Operation

Operations let you extend Ironflow with custom step types for integrating external services.

1. Implement the Operation trait

use std::env;
use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use ironflow_engine::operation::Operation;
use serde_json::{Value, json};

pub struct SlackNotify {
    webhook_url: String,
    message: String,
}

impl SlackNotify {
    pub fn new(message: &str) -> Self {
        let webhook_url = env::var("SLACK_WEBHOOK_URL")
            .expect("SLACK_WEBHOOK_URL env var required");
        Self {
            webhook_url,
            message: message.to_string(),
        }
    }
}

impl Operation for SlackNotify {
    fn kind(&self) -> &str {
        "slack-notify"
    }

    fn input(&self) -> Option<Value> {
        Some(json!({ "message": self.message }))
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>> {
        Box::pin(async move {
            let client = reqwest::Client::new();
            let resp = client
                .post(&self.webhook_url)
                .json(&json!({ "text": self.message }))
                .send()
                .await
                .map_err(|e| EngineError::OperationFailed {
                    kind: "slack-notify".to_string(),
                    message: e.to_string(),
                })?;

            Ok(json!({ "status": resp.status().as_u16() }))
        })
    }
}

2. Use it in a workflow

let notifier = SlackNotify::new("Deploy complete!");
ctx.operation("notify-team", &notifier).await?;

The step is tracked in the database like any other step, with its input, output, and status.

Using a Pre-built Ops Crate

Ironflow ships with 13 ops crates under ops/ that provide ready-to-use integrations for common services. Instead of implementing the Operation trait yourself, you can use these crates to get typed, tracked operations in a single cargo add.

The pattern

Every ops crate follows the same three-step pattern:

1. Add the dependency to your workflow crate
2. Build a client from your workflow’s OperationContext (via from_context())
3. Run a tracked operation via ctx.operation()

use ironflow_ops_slack::SlackClient;
use ironflow_ops_slack::chat::ChatPostMessage;
use slack_morphism::api::SlackApiChatPostMessageRequest;
use slack_morphism::{SlackChannelId, SlackMessageContent};

// 1. Build the client (reads slack_bot_token from the secret store)
let slack = SlackClient::from_context(&ctx).await?;

// 2. Create the operation
let req = SlackApiChatPostMessageRequest::new(
    SlackChannelId::new("#deployments".to_string()),
    SlackMessageContent::new().with_text("Deploy complete".to_string()),
);
let op = ChatPostMessage::new(&slack, req);

// 3. Execute as a tracked workflow step
let output = ctx.operation("notify-team", &op).await?;

The step is tracked in the database with its input, output, kind, and status, just like a shell or HTTP step.

Available crates

Example: GitLab integration

use ironflow_ops_gitlab::GitLab;
use gitlab::api::projects::issues::CreateIssue;

// Build from workflow context (reads gitlab_token from the secret store)
let gitlab = GitLab::from_context(&ctx).await?;

// For a self-hosted instance
let gitlab = GitLab::from_context_with_host(&ctx, "gitlab.example.com").await?;

// Create an issue as a tracked step
let endpoint = CreateIssue::builder()
    .project("my-group/my-project")
    .title("Automated bug report")
    .description("Detected by workflow")
    .build()?;

let output = ctx.operation("create-issue", &gitlab.op(endpoint)).await?;

gitlab.op masks secrets in the request body before sending it: the client’s own token, the values added with GitLab::with_secrets, and token-shaped strings (sk-ant-..., glpat-...). op.masks_secrets() tells whether anything is masked, and the step input records it as "secrets_masked": true. A request that writes a secret on purpose (a CI/CD variable, a webhook token) goes through gitlab.op(endpoint).without_masking(), or GitLab stores [REDACTED] as its value.

Example: Kubernetes + Helm

use ironflow_ops_k8s::{KubeClient, verb};
use ironflow_ops_helm::HelmClient;
use ironflow_ops_helm::release::Upgrade;
use k8s_openapi::api::apps::v1::Deployment;

// Check deployment status
let kube = KubeClient::from_context(&ctx).await?;
let deployments = kube.namespaced::<Deployment>("production");
let op = kube.op(deployments, verb::Get::new("my-app"));
ctx.operation("check-deployment", &op).await?;

// Upgrade the Helm release
let helm = HelmClient::from_context(&ctx).await?;
let upgrade = Upgrade::new(helm, "my-app", "charts/my-app")
    .namespace("production")
    .set("image.tag", "v2.1.0");
ctx.operation("upgrade-release", &upgrade).await?;

Secrets and authentication

Each crate documents its required and optional secrets in its README. The general pattern is:

1. Register secrets in your workflow’s secret store
2. Call from_context(&ctx) which resolves them automatically
3. The client handles authentication transparently

See each crate’s README for the exact secret names and formats.

Writing your own operation

If none of the pre-built crates cover your service, see Writing an Operation to implement the Operation trait directly.

Parallel Execution

Ironflow supports running multiple steps in parallel within a workflow.

Using ctx.parallel()

Pass a list of step configurations to ctx.parallel(). All steps run concurrently and the method returns when all complete:

//! Demo workflow showcasing parallel execution and conditional branching.

use ironflow_engine::config::{ShellConfig, StepConfig};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};

/// Simulated CI pipeline that demonstrates DAG features:
///
/// 1. **build** (sequential)
/// 2. **test-unit + test-integration + lint** (parallel)
/// 3. **deploy** or **notify-failure** (conditional branch on test results)
pub struct CiPipeline;

impl WorkflowHandler for CiPipeline {
    fn name(&self) -> &str {
        "ci-pipeline"
    }

    fn description(&self) -> &str {
        "Simulated CI pipeline with parallel tests and conditional deploy. \
         Demonstrates ctx.parallel() and native Rust if/else branching."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("ci_pipeline.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            // Step 1: Build
            let build = ctx
                .shell(
                    "build",
                    ShellConfig::new(
                        "echo 'Compiling project...' && sleep 0.2 && echo 'Build successful'",
                    ),
                )
                .await?;

            if !build.is_success() {
                ctx.shell(
                    "notify-build-failure",
                    ShellConfig::new("echo 'BUILD FAILED - notifying team'"),
                )
                .await?;
                return Ok(());
            }

            // Step 2: Parallel tests + lint
            let results = ctx
                .parallel(
                    vec![
                        (
                            "test-unit",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running unit tests...' && sleep 0.3 && echo '42 tests passed'",
                            )),
                        ),
                        (
                            "test-integration",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running integration tests...' && sleep 0.5 && echo '12 tests passed'",
                            )),
                        ),
                        (
                            "lint",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running linter...' && sleep 0.1 && echo 'No warnings'",
                            )),
                        ),
                    ],
                    true,
                )
                .await?;

            // Step 3: Conditional deploy
            let all_passed = results.iter().all(|r| r.output.is_success());

            if all_passed {
                ctx.shell(
                    "deploy",
                    ShellConfig::new("echo 'Deploying to production...' && sleep 0.2 && echo 'Deployed successfully'"),
                )
                .await?;
            } else {
                ctx.shell(
                    "notify-test-failure",
                    ShellConfig::new("echo 'TESTS FAILED - deployment skipped'"),
                )
                .await?;
            }

            Ok(())
        })
    }
}

How it works

• All steps in a parallel() call start at the same time
• The method returns a Vec<ParallelStepResult> with outputs in the same order as the input; each output reads through the same typed accessors as a single step (stdout(), status(), body(), is_success()) and hands out the artifacts its step declared (r.output.artifact("report.html")?)
• If fail_fast is true (the second argument), the remaining steps are cancelled when one fails
• If fail_fast is false, all steps run to completion regardless of individual failures
• Every step of a wave needs its own name: a wave with two steps of the same name fails with EngineError::StepConfig before anything runs (a dry-run plan reports it too)
• When a run resumes (after an approval, a human input or a delay), the steps of the wave that already completed are replayed from the store; only the others run again
• ctx.parallel() takes shell, HTTP and agent steps. Sub-workflows run concurrently through ctx.workflows_with(), below

Sub-workflows in parallel

ctx.workflows_with(&Child, children) starts one child run of Child per entry, each with its own input and WorkflowOptions, and runs them concurrently:

let wave = agents
    .iter()
    .map(|agent| {
        let options = WorkflowOptions::new()
            .step_name(format!("delegate-{tour}-{agent}"))
            .label("team.example/agent", agent.as_str())
            .allow_failure();
        (DelegateInput { agent: agent.clone() }, options)
    })
    .collect();
for outcome in ctx.workflows_with(&Delegate, wave).await? {
    if let SubWorkflowOutcome::Completed(child) = outcome {
        println!("{} ended {}", child.run_id(), child.status());
    }
}

• One Workflow step per child, all at the same position; the outcomes come back in the order of the entries. Two entries with the same step name fail with EngineError::StepConfig before anything runs
• The child runs are created in list order, then run concurrently. A concurrency key already held, by an earlier sibling or by another run, gives a Conflict for that entry
• The call waits for every child. allow_failure is set per child; a failure it does not tolerate is returned once every sibling has finished and its step is recorded
• The cost cap of the root run is shared live by the whole tree: every agent step checks it against what the root, the siblings and their descendants spent and hold for their running steps, so a wave never takes the root past its cap. A child refused by the cap ends Cancelled with RUN_BUDGET_EXCEEDED
• The workflow guard counts each child as one invocation, and refuses a wave that would cross max_fan_out before anything is created
• A child that suspends leaves its step open while the others run to the end, then the parent suspends with it. On resume, finished children are replayed and suspended ones re-entered: no child run is created twice

Conditional branching

Since workflows are Rust code, conditional logic is just if/else:

let results = ctx.parallel(steps, true).await?;
let all_passed = results.iter().all(|r| r.output.is_success());

if all_passed {
    ctx.shell("deploy", ShellConfig::new("echo 'Deploying'")).await?;
} else {
    ctx.shell("notify", ShellConfig::new("echo 'Tests failed'")).await?;
}

No special DSL for branching – Rust control flow works directly.

Execution Plans

An execution plan answers one question before you trigger anything: what would this workflow do with this input? It lists the steps the run would create, in order, with their dependencies, their parallel waves, their branch conditions and — when the workflow has a history — an estimate of how long each one takes.

What plan mode does, and does not

Ironflow workflows are Rust-native handlers, not declarative graphs. There is no static file to read, so the only way to know which steps a run would create is to execute the handler with every step method short-circuited. That is plan mode.

In plan mode:

• no shell command is spawned, no HTTP request is sent, no agent is called;
• no custom Operation is executed, so no third-party API is touched;
• an approval gate is recorded and stepped over instead of suspending the run;
• a delay is recorded without sleeping;
• ctx.store() and ctx.secrets() return an error, so a handler cannot read or write the store or a secret while planning: the plan stops there, partial; check ctx.is_planning() first to plan past such a call;
• nothing is written: no run, no step, no dependency, no artifact.

Each step method returns a synthetic, success-shaped output instead (exit_code: 0 for a shell step, status: 200 for an HTTP step). A handler that branches on build.is_success() therefore follows the happy path: a plan shows the nominal branch, not every branch the run might take.

From the CLI

ironflow-cli run plan deploy --input '{"env":"prod"}'

workflow deploy  estimated ~2m 14s
├─ build [shell] ~1m 02s
├─ parallel-1
  ├─ test-unit [shell] ~41s
  ├─ test-integration [shell] ~58s
  ├─ lint [shell] ~12s
└─ deploy-prod [shell] ~14s (when production run = true)

Useful flags:

From the API

POST /api/v1/workflows/{name}/plan

{
  "payload": { "env": "prod" },
  "max_depth": 3,
  "estimate_durations": true
}

The response is the usual envelope around an execution plan:

{
  "data": {
    "workflow": "deploy",
    "max_depth": 3,
    "truncated": false,
    "estimated_duration_ms": 134000,
    "steps": [
      {
        "name": "build",
        "kind": "shell",
        "workflow": "deploy",
        "depth": 0,
        "depends_on": [],
        "estimated_duration_ms": 62000
      },
      {
        "name": "deploy-prod",
        "kind": "shell",
        "workflow": "deploy",
        "depth": 0,
        "depends_on": ["build"],
        "condition": {
          "state": "evaluated",
          "expression": "production run",
          "value": true
        }
      }
    ]
  }
}

The route needs an authenticated caller, like GET /api/v1/workflows/{name}: planning has no side effect, so it is not admin-only. It answers 400 for a max_depth outside 1..=10 or a payload that is not a JSON object, and 404 for a workflow that is not registered.

Conditions

A condition appears on a step in one of three states:

A plain Rust if on a step output stays invisible to the planner by construction. ctx.when_dynamic exists precisely to surface it.

Durations

When estimate_durations is on (the default), the planner samples the most recent completed runs of the workflow and averages each step’s duration by name. A step with no history carries no estimate, and a workflow with no history at all reports estimated_duration_ms as absent rather than zero. The plan total sums sequential steps and counts each parallel wave once, at its slowest member.

Sub-workflows and limits

ctx.workflow(...) is expanded in place: the child’s steps appear inline with depth incremented and workflow set to the child’s name. max_depth bounds that expansion, so a workflow that invokes itself stops there.

Three guards keep a plan bounded, and all set truncated with an incomplete_reason:

• the depth limit above;
• a hard cap of 1000 planned steps, which stops a handler that loops, whatever the step kind: shell, HTTP, agent, operation, parallel, sub-workflow, delay, signal, approval, decision, skip or human input;
• a timeout, PlanOptions::timeout (DEFAULT_PLAN_TIMEOUT, 10 s by default), which stops a handler that awaits forever. It interrupts the handler at an .await: a loop that never yields is bounded by the step cap instead.

A handler that returns an error mid-plan does not fail the request either: the partial plan is returned, with incomplete_reason carrying the error.

Limits worth knowing

• Synthetic outputs are success-shaped, so only the nominal branch is planned.
• A handler that unwraps a decision answer, or deserializes ctx.input::<T>() against a payload it does not match, aborts the plan; you get the steps recorded so far plus the reason.
• Secrets and artifacts are not resolved while planning.

Agent Providers

A provider is what an agent step talks to: the Claude Code CLI, a remote Claude Code, or an HTTP API such as OpenAI. Every provider implements AgentProvider, so a workflow written against one runs against any other. Only ClaudeCodeProvider is always available; the others are behind a feature flag.

The SSH, Docker and Kubernetes providers run Claude Code somewhere else; they are described in Transports.

With a tool profile (or with_tools) and .output::<T>(), the model can call registry tools before returning the structured answer. GeminiProvider, and AnthropicApiProvider on models without native JSON output, send the registry tools together with a structured_output tool and require the model to call a tool every turn. On Anthropic models with native JSON output (Sonnet 5.5, Opus 5.5, Fable 5.1, Mythos 5.1), only the registry tools are sent: the schema stays in output_config.format and the final text turn is the answer.

Reasoning effort

AgentConfig::effort(Effort::High) (also available on AgentStepConfig) takes low, medium, high, xhigh or max. The Claude Code providers (local, SSH, Docker, Kubernetes) pass it as --effort, also when a session is resumed, and AnthropicApiProvider sends output_config.effort. The other HTTP providers ignore it. When unset, no flag is sent.

Offered tools

AgentConfig::available_tools([Tool::Read, Tool::Edit]) sets the built-in tools the Claude Code session offers (--tools); an empty list offers none, and without the call the CLI offers its default set. allow_tool only pre-approves: in dontAsk, Claude Code still runs a read-only Bash or Monitor command that the allowlist leaves out, so a step that must not run a shell sets both. Only built-in tools are filtered, the tools of --mcp-config stay offered; the CLI ignores an unknown name or a permission pattern (Bash(git log:*)) without error. HTTP providers ignore the field.

HTTP providers

HTTP providers are used exactly like the local one:

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;
use ironflow_core::providers::http::{NvidiaModel, NvidiaProvider};

async fn example() -> Result<(), OperationError> {
let provider = NvidiaProvider::from_env(); // reads NVIDIA_API_KEY

let result = Agent::new()
    .prompt("Summarize the changelog")
    .model(NvidiaModel::DEEPSEEK_V4_FLASH)
    .max_budget_usd(0.10)
    .run(&provider)
    .await?;
Ok(())
}
}

Tools for HTTP providers

The Claude Code CLI brings its own tools. HTTP providers do not, so tools are opt-in, one feature each: tool-bash, tool-read-file, tool-grep, tool-glob, tool-web-fetch, tool-web-search, and tool-mcp to bring any MCP server into the agent’s toolset. GrepTool and GlobTool search a codebase without handing the agent a shell: they are confined to the directories you give them.

#![allow(unused)]
fn main() {
use std::path::PathBuf;

use ironflow_core::providers::http::tools::glob::GlobTool;
use ironflow_core::providers::http::tools::grep::GrepTool;
use ironflow_core::providers::http::tools::{ToolError, ToolRegistry};

fn example() -> Result<(), ToolError> {
let repo = vec![PathBuf::from("/srv/repo")];
let tools = ToolRegistry::new()
    .register(GrepTool::with_allowed_paths(repo.clone())?)
    .register(GlobTool::with_allowed_paths(repo)?);
Ok(())
}
}

Tool profiles

A step picks a named tool profile and sees only its tools; without one it gets the with_tools registry, or none. Declare each ToolProfile once as a constant shared by the provider and the steps, so a misspelled profile does not compile. An MCP server shared between profiles is opened once:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use ironflow_core::prelude::*;
use ironflow_core::provider::{AgentConfig, ToolProfile};
use ironflow_core::providers::http::OpenAiProvider;
use ironflow_core::providers::http::tools::ToolRegistry;
use ironflow_core::providers::http::tools::mcp::{McpConnection, register_shared_mcp_tools};

const SUGGESTION: ToolProfile = ToolProfile::new("suggestion");
const BUG: ToolProfile = ToolProfile::new("bug");

async fn example() -> Result<(), Box<dyn std::error::Error>> {
let mut gitlab = McpConnection::stdio("mcp-gitlab", &[], &[]).await?;
gitlab.initialize().await?;
let gitlab = Arc::new(gitlab);
let provider = OpenAiProvider::from_env()
    .with_tool_profile(SUGGESTION, register_shared_mcp_tools(ToolRegistry::new(), &gitlab, "gitlab").await?)
    .with_tool_profile(BUG, register_shared_mcp_tools(ToolRegistry::new(), &gitlab, "gitlab").await?);

let config = AgentConfig::new("Find the root cause").tool_profile(BUG);
let result = Agent::from_config(config).model("gpt-5.5").run(&provider).await?;
Ok(())
}
}

In a workflow, an agent step selects the profile with .tool_profile(BUG); see Agent steps.

Routing between providers

ProviderRouter picks the provider from the model name, so a single workflow can mix vendors:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use ironflow_core::prelude::*;
use ironflow_core::providers::http::NvidiaProvider;

async fn example() -> Result<(), OperationError> {
let claude = Arc::new(ClaudeCodeProvider::new());
let nvidia = Arc::new(NvidiaProvider::from_env());

let router = ProviderRouter::new(claude)
    .route(ProviderMatcher::ModelPrefix("nvidia/".into()), nvidia);

// Goes to Claude Code
let a = Agent::new().prompt("Review").model(Model::SONNET).run(&router).await?;

// Goes to NVIDIA
let b = Agent::new().prompt("Review").model("nvidia/deepseek-v4-flash").run(&router).await?;
Ok(())
}
}

To spread agent steps over several Claude subscriptions or API keys, see Provider Accounts.

Transports

Transports control where agent steps execute. By default, agents run on the local machine via ClaudeCodeProvider. Ironflow ships additional transports for running agents in isolated environments.

Available transports

Docker transport

Executes agent commands inside a running Docker container via docker exec:

//! Docker transport example.
//!
//! Executes `claude` inside a running Docker container via `docker exec`.
//! The container must already be running and have the `claude` binary installed.
//!
//! # Usage
//!
//! ```sh
//! DOCKER_CONTAINER=claude-worker cargo run --bin docker-transport
//! ```

use std::env;

use ironflow_core::prelude::*;
use ironflow_core::providers::claude::DockerProvider;

#[tokio::main]
async fn main() -> Result<(), OperationError> {
    let container = env::var("DOCKER_CONTAINER").expect("DOCKER_CONTAINER env var required");

    let provider = DockerProvider::new(&container).working_dir("/workspace");

    let result = Agent::new()
        .prompt("What is 2 + 2?")
        .max_budget_usd(0.10)
        .run(&provider)
        .await?;

    println!("Response: {}", result.text());
    println!("Model: {}", result.model().unwrap_or("unknown"));
    println!("Duration: {}ms", result.duration_ms());

    Ok(())
}

SSH transport

Connects to a remote host via SSH:

//! SSH transport example.
//!
//! Connects to a remote host via SSH and runs `claude` there.
//! Requires the `claude` binary to be installed on the remote host.
//!
//! # Usage
//!
//! ```sh
//! SSH_HOST=build-server SSH_USER=deploy SSH_PASSWORD=secret cargo run --bin ssh-transport
//! ```

use std::env;

use ironflow_core::prelude::*;
use ironflow_core::providers::claude::SshProvider;
use ironflow_core::providers::claude::ssh::HostKeyPolicy;

#[tokio::main]
async fn main() -> Result<(), OperationError> {
    let host = env::var("SSH_HOST").expect("SSH_HOST env var required");
    let user = env::var("SSH_USER").expect("SSH_USER env var required");
    let password = env::var("SSH_PASSWORD").expect("SSH_PASSWORD env var required");

    // For production, use HostKeyPolicy::Fingerprint or HostKeyPolicy::KnownHostsFile
    let provider = SshProvider::new(&host, &user)
        .password(&password)
        .host_key_policy(HostKeyPolicy::AcceptAll);

    let result = Agent::new()
        .prompt("What is 2 + 2?")
        .max_budget_usd(0.10)
        .run(&provider)
        .await?;

    println!("Response: {}", result.text());
    println!("Model: {}", result.model().unwrap_or("unknown"));
    println!("Duration: {}ms", result.duration_ms());

    Ok(())
}

Kubernetes transport

Two modes are available:

• Ephemeral – creates a pod for each agent call, deletes it when done
• Persistent – reuses a long-lived pod for multiple calls

See the examples/transports/ directory for complete Kubernetes examples.

For untrusted prompts or multi-tenant clusters, K8sEphemeralProvider::sandboxed runs each agent in a hardened pod: non-root, read-only root filesystem, secrets read from Kubernetes Secrets, managed-settings presets and egress profiles. See Kubernetes Sandbox.

Step timeout

A step with AgentConfig::timeout (see Agent steps) replaces the provider timeout. The ephemeral Kubernetes pod is deleted when the step times out. With Docker, SSH or a persistent pod, a step with a timeout runs claude under timeout -s KILL, so the image or host needs a timeout binary (coreutils or busybox) on its PATH: the remote side then kills claude 5 s after the limit, even once the connection is gone. A step without a timeout runs the command unchanged.

Provider Account credentials

The local, Docker and SSH transports inject the credential of the Provider Account the worker selected: through the process environment, the Docker exec environment, or the first line of stdin over SSH. The token never appears on a command line. The Kubernetes transports do not inject accounts yet and keep using the pod environment.

Resuming an interrupted agent step

Before an agent step launches, the engine pins the Claude Code session it runs in (--session-id <uuid>) and records it on the step as session_id. When the worker running the step loses its lease, the step is interrupted (see Engine & Worker). The next execution of the same step, in the same attempt, resumes that session (--resume <uuid>) with a resume prompt instead of starting the agent from scratch, and keeps the conversation and the work already done.

The resume prompt defaults to DEFAULT_RESUME_PROMPT. Set your own on the step:

ctx.agent(
    "review",
    AgentStepConfig::new("Review the diff")
        .max_budget_usd(0.50)
        .resume_prompt("You were interrupted. Finish the review where you stopped."),
)
.await?;

Claude Code keeps its sessions under ~/.claude/projects/, keyed by the working directory. A step can only resume a session the transport can still read:

When the session is gone, the step does not fail: it logs session <uuid> not found, restarting the agent from scratch and runs the original prompt again in a session of the same id. A step that sets resume or session_id itself is left alone.

Each resume writes the system log line agent step resumed from session <uuid> on the step and publishes an agent_step_resumed event on the run’s event stream. Only a step interrupted by a lost lease resumes: a step failed by the agent, a manual retry of the run (a new attempt) or a parked step that never launched starts a fresh session.

Cost and tokens of a resumed step only cover the resume invocation. The interrupted invocation never wrote its final result line, so what it spent before the worker lost its lease stays unknown and is not added to the step, the run or the budget counters. Expect the real spend of a resumed step to be higher than the recorded one.

A step retry (retry_policy) stays in the session of the step: the retry sends the original prompt into the session the first try created (the CLI refuses to create a session id twice), so the step records a single session id.

Choosing a transport

• Development: use ClaudeCodeProvider (local). No setup needed.
• CI/CD pipelines: Docker or Kubernetes for isolation.
• Remote build servers: SSH for machines you already manage.
• Multi-tenant production: Kubernetes ephemeral pods for strong isolation between tenants.

Kubernetes Sandbox

K8sEphemeralProvider::sandboxed(image) runs every agent step in a hardened, single-use pod. Everything is opt-in: K8sEphemeralProvider::new(image) keeps its behaviour, apart from the run/step labels, the expiry annotation and the cleanup of a previous attempt described below.

The image

The official image registry.gitlab.com/thomastartrau/ironflow/ironflow-claude-runner:<claude-code-version>-<n> is built from docker/claude-runner/ by the build-claude-runner-image CI job. The current tag is in docker/claude-runner/IMAGE_TAG. There is no latest tag: pin the full tag. The image also bundles chrome-devtools-mcp at a pinned version, with no browser, for agents that drive a remote Chromium.

# Official ironflow Claude Code runner image, used by
# K8sEphemeralProvider::sandboxed.
#
# Published as ironflow-claude-runner:<claude-code-version>-<n>, the content of
# IMAGE_TAG next to this file. Never `latest`. CLAUDE_CODE_VERSION is the
# <claude-code-version> part: see README.md for a local build.
#
# Layout:
#   /home/claude                 HOME of uid 10001. The sandboxed provider mounts
#                                an emptyDir here: the root filesystem is read-only.
#   /tmp                         emptyDir mounted by the sandboxed provider.
#   /etc/claude-code             managed-settings.json, read by Claude Code on
#                                Linux. A default is baked in; a managed-settings
#                                ConfigMap mounted by the provider replaces it.
#   /etc/ironflow/claude-profile Claude profile ConfigMap (CLAUDE.md, settings,
#                                agents), copied into ~/.claude before the agent
#                                starts because ~/.claude must stay writable.
#
# Both /etc directories are owned by root with mode 0755: the agent can read
# them, never change them.

# Pinned by digest so a rebuild of the same tag starts from the same base. To
# move to a newer node image, replace the digest and bump <n> in IMAGE_TAG.
# 22.23.4 fixes Node2Shell (CVE-2026-111497, Brotli decoder).
FROM node:22.23.4-bookworm-slim@sha256:de170b39439eb0e5a3444ad7d60b015614a381e10fecae2e30ec9e735122f38d

ARG CLAUDE_CODE_VERSION

# upgrade: the base image ships libbrotli1 1.0.9-2+b6 (CVE-2026-111497, decoded by
# libcurl under git and curl), fixed in 1.0.9-2+deb12u1.
RUN apt-get update \
    && apt-get upgrade -y \
    && apt-get install -y --no-install-recommends ca-certificates git curl \
    && rm -rf /var/lib/apt/lists/*

# Fails the build when the version is missing, or when the installed CLI
# reports another one.
RUN : "${CLAUDE_CODE_VERSION:?pass --build-arg CLAUDE_CODE_VERSION=<version>}" \
    && npm install -g "@anthropic-ai/claude-code@${CLAUDE_CODE_VERSION}" \
    && npm cache clean --force \
    && test "$(claude --version | cut -d' ' -f1)" = "${CLAUDE_CODE_VERSION}"

# chrome-devtools-mcp: stdio MCP server the Claude CLI starts in the agent pod,
# driving a browser that runs in another pod (--browser-url
# http://<pod-ip>:9223). No Chromium here. Installed at an exact version because
# the agent pod has no egress to the npm registry, so `npx` at startup cannot work
# and a floating download is not reproducible. Fails the build when the
# installed server reports another version.
RUN PUPPETEER_SKIP_DOWNLOAD=1 npm install -g chrome-devtools-mcp@1.10.1 \
    && npm cache clean --force \
    && test "$(chrome-devtools-mcp --version)" = "1.10.1"

RUN groupadd -g 10001 claude \
    && useradd -u 10001 -g 10001 -m -d /home/claude claude

# Marketplace neutralisation: no plugin marketplace, no hooks, no auto-update,
# no non-essential traffic. A mounted managed-settings ConfigMap replaces the
# baked file.
RUN mkdir -p /etc/claude-code /etc/ironflow/claude-profile \
    && printf '%s\n' '{"strictKnownMarketplaces": [], "disableAllHooks": true}' \
        > /etc/claude-code/managed-settings.json \
    && chown -R root:root /etc/claude-code /etc/ironflow \
    && chmod 0755 /etc/claude-code /etc/ironflow /etc/ironflow/claude-profile \
    && chmod 0644 /etc/claude-code/managed-settings.json

ENV CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 \
    DISABLE_AUTOUPDATER=1 \
    HOME=/home/claude

USER 10001:10001
WORKDIR /home/claude

sandboxed() defaults

A relaxation called on a provider built with new() panics.

Secrets

A sandboxed provider refuses oauth_credentials(json) and env("ANTHROPIC_API_KEY", ..) / env("CLAUDE_CODE_OAUTH_TOKEN", ..): the value would sit in clear text in the pod spec. Read it from a Kubernetes Secret instead; the pod only carries a secretKeyRef:

let provider = K8sEphemeralProvider::sandboxed(&image)
    // Long-lived token from `claude setup-token`.
    .oauth_token_from_secret("claude-oauth", "token")
    // Or the full credentials JSON, written to ~/.claude/.credentials.json.
    .oauth_credentials_from_secret("claude-credentials", "credentials.json")
    // Any other variable.
    .env_from_secret("GITLAB_TOKEN", "gitlab-bot", "token");

Auth proxy: no Claude credential in the pod

A secretKeyRef keeps the credential out of the pod spec, not out of the pod: anything running in the agent container can still read CLAUDE_CODE_OAUTH_TOKEN. With auth_proxy, the pod never receives it:

let provider = K8sEphemeralProvider::sandboxed(&image)
    .namespace("ironflow-agents")
    .auth_proxy("http://ironflow-auth-proxy.ironflow-system");

The ironflow-auth-proxy service (crate ironflow-auth-proxy, manifests in examples/k8s/sandbox/auth-proxy.yaml) holds the credential instead. Its official image registry.gitlab.com/thomastartrau/ironflow/ironflow-auth-proxy:<version> is built from docker/auth-proxy/Dockerfile by the build-auth-proxy-image CI job. <version> is the version of the ironflow-auth-proxy crate: the job publishes it once that version is released, and never rebuilds a published tag. There is no latest tag: pin the version.

• Token lifecycle. At pod launch the worker calls the proxy admin API and gets an opaque token (ifap_...) bound to the run id, the step and the pod expiry (ironflow.io/expires-at). The pod receives ANTHROPIC_BASE_URL (the proxy URL), ANTHROPIC_AUTH_TOKEN (the opaque token) and CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1. The worker revokes the token when the step ends, whatever the outcome, including a step dropped before it ends (run cancelled, lease lost, run or workflow timed out, parallel branch aborted), and every token of a run when the run is released; the expiry is the backstop, after a worker crash or shutdown. A step that ends waits at most 5 seconds for the revocation, which goes on in the background past it. A cancelled run is dropped at the next lease renewal (every 30 s by default). An unknown, revoked or expired token gets a 401. If the proxy cannot issue a token, the step fails before any pod is created.
• Restrictions. The proxy only relays GET and POST requests under /v1/ to api.anthropic.com. A path outside /v1/ (including /admin, which needs the admin key) or a request for another host (absolute-form URI, CONNECT) gets a 403, any other method a 405.
• Which credential. The step’s Provider Account when one is attached (the provider reports the claude_subscription account kind once auth_proxy is set), else CLAUDE_CODE_OAUTH_TOKEN, then ANTHROPIC_API_KEY, from the worker environment. The worker also needs IRONFLOW_AUTH_PROXY_ADMIN_KEY (or .auth_proxy_admin_key(..)). Rate-limit windows reported through the proxy are recorded on the Provider Account as with the Docker provider.
• No credential on the pod side. With auth_proxy set, a Claude credential configured for the pod (oauth_token_from_secret, oauth_credentials, oauth_credentials_from_secret, a step env_from_secret("CLAUDE_CODE_OAUTH_TOKEN", ..), any plain value starting with sk-ant) fails the step with an error naming the variable.
• Registry. In memory by default (one replica). Set IRONFLOW_AUTH_PROXY_DATABASE_URL and an encryption key (IRONFLOW_SECRET_KEYS) for a shared PostgreSQL registry: several replicas, tokens survive restarts, the proxy needs egress to PostgreSQL. Under Cilium, uncomment the 5432 rule in section (b) of cilium-egress-auth-proxy.yaml (below). With standard NetworkPolicies only, nothing restricts the proxy’s egress unless the operator adds a policy on the proxy pods.
• Logs. The proxy and the worker log the first 12 characters of the token id (a SHA-256 of the token), never the token or the credential.

The network side changes too: the agent pods only reach the proxy, and the proxy only reaches api.anthropic.com (and PostgreSQL with the shared registry):

# Cilium egress for the auth proxy variant.
#
# Use INSTEAD of cilium-egress-anthropic.yaml: the agent pods no longer reach
# api.anthropic.com, only the auth proxy, which alone holds the credential
# and alone reaches the API. The per-profile policies (gitlab below) are
# unchanged and stay additive. With proxied secrets (the GitLab token held by
# the proxy, git and the API reached through /r/), the agent pods no longer
# need the gitlab profile: open gitlab.com on the proxy pods instead, in (b).
#
# (a) Agent pods: DNS, and the auth proxy pods on 8080.
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: claude-runner-egress-auth-proxy
  namespace: ironflow-agents
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/component: claude-runner
  egress:
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: kube-system
            k8s:k8s-app: kube-dns
      toPorts:
        - ports:
            - port: "53"
              protocol: ANY
          rules:
            dns:
              - matchPattern: "*"
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: ironflow-system
            app.kubernetes.io/name: ironflow-auth-proxy
      toPorts:
        - ports:
            - port: "8080"
              protocol: TCP
---
# Per-profile opening, as in cilium-egress-anthropic.yaml: pods labelled
# ironflow.io/egress-profile=gitlab may also reach gitlab.com.
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: claude-runner-egress-gitlab
  namespace: ironflow-agents
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/component: claude-runner
      ironflow.io/egress-profile: gitlab
  egress:
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: kube-system
            k8s:k8s-app: kube-dns
      toPorts:
        - ports:
            - port: "53"
              protocol: ANY
          rules:
            dns:
              - matchPattern: "*"
    - toFQDNs:
        - matchName: gitlab.com
      toPorts:
        - ports:
            - port: "443"
              protocol: TCP
---
# (b) The auth proxy pods: egress to DNS and api.anthropic.com:443 only, plus
# the hosts of the proxied secrets (commented-out rule below), plus
# PostgreSQL on 5432 with the shared registry (commented-out rule below);
# ingress on 8080 only from the agent pods and from the worker. Adjust the
# worker selector to the namespace and labels the worker runs under.
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: ironflow-auth-proxy
  namespace: ironflow-system
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/name: ironflow-auth-proxy
  ingress:
    - fromEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: ironflow-agents
            app.kubernetes.io/component: claude-runner
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: ironflow
            app.kubernetes.io/name: ironflow-worker
      toPorts:
        - ports:
            - port: "8080"
              protocol: TCP
  egress:
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: kube-system
            k8s:k8s-app: kube-dns
      toPorts:
        - ports:
            - port: "53"
              protocol: ANY
          rules:
            dns:
              - matchPattern: "*"
    - toFQDNs:
        - matchName: api.anthropic.com
      toPorts:
        - ports:
            - port: "443"
              protocol: TCP
    # Proxied secrets: every host in the allowlist of a proxied secret must be
    # opened here for the proxy pods, else the relay answers 502. Wildcard
    # allowlist entries map to matchPattern.
    # - toFQDNs:
    #     - matchName: api.github.com
    #     - matchName: gitlab.com
    #   toPorts:
    #     - ports:
    #         - port: "443"
    #           protocol: TCP
    #
    # Shared PostgreSQL registry (IRONFLOW_AUTH_PROXY_DATABASE_URL set in
    # auth-proxy.yaml): uncomment ONE of the two rules below, else the proxy
    # times out at startup. Not needed with the in-memory registry. If the
    # database has its own ingress policy, open 5432 there to the proxy pods.
    #
    # In-cluster database: adjust its namespace and pod labels.
    # - toEndpoints:
    #     - matchLabels:
    #         k8s:io.kubernetes.pod.namespace: ironflow-db
    #         app.kubernetes.io/name: postgresql
    #   toPorts:
    #     - ports:
    #         - port: "5432"
    #           protocol: TCP
    #
    # Managed database outside the cluster: adjust to its subnet.
    # - toCIDR:
    #     - 10.0.0.0/24
    #   toPorts:
    #     - ports:
    #         - port: "5432"
    #           protocol: TCP

Proxied secrets

The same proxy keeps other secrets out of the pod: a GitHub token, a GitLab token, any API key. proxied_secret declares one on the provider (every step) or on a step, a step entry replacing a provider entry with the same env:

use std::env::var;

use ironflow_core::auth_proxy::{ProxiedSecret, SecretInjection};

let provider = K8sEphemeralProvider::sandboxed(&image)
    .auth_proxy("http://ironflow-auth-proxy.ironflow-system")
    .proxied_secret(ProxiedSecret {
        name: "GITHUB_TOKEN".to_string(),
        env: "GITHUB_TOKEN".to_string(),
        value: var("GITHUB_TOKEN")?,
        injection: SecretInjection::Bearer,
        hosts: vec!["api.github.com".to_string()],
    });

let config = AgentConfig::new("Open the merge request").proxied_secret(ProxiedSecret {
    name: "GITLAB_TOKEN".to_string(),
    env: "GITLAB_TOKEN".to_string(),
    value: var("GITLAB_TOKEN")?,
    injection: SecretInjection::Basic { username: "oauth2".to_string() },
    hosts: vec!["gitlab.com".to_string()],
});

The worker issues one opaque token per secret, bound to the run and the step like the Claude token and revoked with it. The pod receives the token in <env> and the relay base <proxy>/r in <env>_URL, never the value, and calls $<env>_URL/<host>/<path> presenting the token as Authorization: Bearer, x-api-key, Private-Token or the password of Authorization: Basic. The proxy strips it and injects the real secret:

• Allowlist. hosts holds exact host names or a leading wildcard (*.example.com matches api.example.com, not example.com). No regex, no port, no IP address; the proxy only speaks https upstream. A host outside the list, a path with .., // or percent-encoding, a Claude token on /r/ and a secret token on the Anthropic API get a 403; an unknown, expired or revoked token a 401. The query string is relayed untouched.
• git. A Basic secret also sets the git config of the pod through GIT_CONFIG_COUNT / GIT_CONFIG_KEY_n / GIT_CONFIG_VALUE_n: for each exact host, https://<host>/ is rewritten to <proxy>/r/<host>/, so git clone https://gitlab.com/group/repo.git works unchanged. A step setting a GIT_CONFIG_* variable itself is refused, as is a proxied <env> or <env>_URL the pod already receives.
• Redirects are returned to the pod with their Location, never followed: the secret never travels to another host.
• Logs and metrics. Each /r/ request logs host, secret (the name), run_id, step, the short token id, result and upstream_status, never the token, the secret, a header or the query string. The counter ironflow_auth_proxy_requests_total{secret, result} is served at /metrics on the proxy port.
• Egress. The proxy pods need egress to every allowlisted host: uncomment the proxied secrets toFQDNs rule in section (b) of cilium-egress-auth-proxy.yaml. The agent pods then no longer need the gitlab egress profile.

Per-step settings

A step adds to or overrides the provider’s settings through its AgentConfig. Other providers ignore these fields.

let config = AgentConfig::new("Open the merge request")
    .env_from_secret("GITLAB_TOKEN", "gitlab-bot", "token") // wins over the provider's
    .service_account("gitlab-reader")                       // wins over the provider's
    .read_only_pvc("repos", "/data/repos")                  // after the provider's volumes
    .read_only_config_map("guidelines", "/data/guidelines")
    .managed_settings("readonly")                           // preset registered on the provider
    .runtime_class("gvisor")                                // wins over the provider's
    .egress_profile("gitlab");                              // ironflow.io/egress-profile label

Read-only mounts must be absolute, unique, and cannot shadow /, /home/claude, /tmp, /etc/claude-code, or /etc/ironflow/claude-profile and anything below it.

Claude profile

A Claude profile (CLAUDE.md, settings.json, rules/, agents/, commands/) reaches ~/.claude through ConfigMaps. A ConfigMap key cannot contain /, so each directory of the profile is its own ConfigMap, mapped to its sub-directory:

let provider = K8sEphemeralProvider::sandboxed(&image)
    .claude_profile_configmap("claude-profile")                 // ~/.claude
    .claude_profile_configmap_at("claude-profile-rules", "rules") // ~/.claude/rules
    .claude_profile_configmap_at("claude-profile-agents", "agents");

Each ConfigMap is mounted read-only in its own directory, then only its keys are copied into ~/.claude/<subdir> before the agent starts: never the ..data entries of the volume. Profiles are copied before the credentials, which they cannot overwrite. A subdir must be relative, made of [A-Za-z0-9._-] segments without . or .., and mapped once: the provider panics at build time otherwise.

With kustomize, one configMapGenerator per directory. Disable the name hash: the provider refers to the ConfigMaps by name.

# kustomization.yaml, next to claude-home/
namespace: ironflow-agents
generatorOptions:
  disableNameSuffixHash: true
configMapGenerator:
  - name: claude-profile
    files:
      - claude-home/CLAUDE.md
      - claude-home/settings.json
  - name: claude-profile-rules
    files:
      - claude-home/rules/rust.md
      - claude-home/rules/security.md

kustomize lists every file. To pick up a whole directory instead: kubectl create configmap claude-profile-rules --from-file=claude-home/rules/ --dry-run=client -o yaml.

Managed-settings presets map a name to a ConfigMap holding managed-settings.json. An unknown preset fails the step, never falls back:

let provider = K8sEphemeralProvider::sandboxed(&image)
    .managed_settings_preset("locked", "claude-managed-locked")
    .managed_settings_preset("readonly", "claude-managed-readonly")
    .default_managed_settings("locked");

Labels and retry cleanup

The engine stamps ironflow.io/run-id, ironflow.io/root-run-id and ironflow.io/step on every agent step (outside the engine, call AgentConfig::run_scope(run_id, step)). Step names are sanitized into valid label values, with a hash suffix when altered. The root run is the run itself, or the top-level run inside a sub-workflow.

Before every execution of a run, first one included, the engine calls release_run: the provider deletes every pod, JobRun Job and prompt ConfigMap labelled with the run id or with the run as root, and waits until the pods are gone (.previous_attempt_timeout(d), 60s by default). A retry that starts by resetting shared state (a worktree) never runs next to an agent of the dead attempt still writing to it. If the pods are still there, or the Kubernetes API fails, the execution fails with a replayable error and the next attempt tries again.

Tag a pod you create yourself with the same labels so that it is released too:

let run = PodRun::new(&kube, "check", &image, "cargo test")
    .label(LABEL_RUN_ID, &ctx.run_id().to_string())
    .label(LABEL_ROOT_RUN_ID, &ctx.root_run_id().to_string());

A service pod started with PodUp::new(run, lifetime) returns once ready and keeps running; tag it with the labels above on its PodRun, and its ironflow.io/expires-at covers lifetime. Only a pod in the provider’s namespace is released or reaped: elsewhere, put the run id in its name, or a new attempt of the run finds the old pod in the way. Once tagged, it is released when the run executes again, while its step is replayed with the IP of the deleted pod: keep the steps that use the service in the same execution, with no approval, signal or human input between them and the PodUp.

Before creating a pod, the provider also deletes the pods and prompt ConfigMaps of a previous attempt of the same step of the same run. If they are still terminating, the step fails: two agents never run side by side. Two branches of a ctx.parallel() group with the same name would delete each other’s pod, so the engine fails such a group before creating any step.

Deleting JobRun Jobs needs list and delete on jobs; without them, Jobs are skipped with a warning and the pods are still released.

Tying pods to the worker pod

release_run only runs when the run executes again, after the worker lease expired. Until then, the agent pod of a killed worker keeps running, and writing to shared state. .owner_pod(name, uid) puts an ownerReference to the worker pod on every agent pod and prompt ConfigMap (controller: false, blockOwnerDeletion: false): the Kubernetes garbage collector deletes them as soon as the worker pod is gone. PodRun::owner_pod and JobRun::owner_pod do the same for the pods and Jobs of ironflow-ops-k8s; a Job takes its pods with it.

The worker reads its name and uid through the downward API:

env:
  - name: POD_NAME
    valueFrom: { fieldRef: { fieldPath: metadata.name } }
  - name: POD_UID
    valueFrom: { fieldRef: { fieldPath: metadata.uid } }

let (name, uid) = (var("POD_NAME")?, var("POD_UID")?);
let provider = K8sEphemeralProvider::sandboxed(&image)
    .namespace(&namespace)
    .owner_pod(&name, &uid);
// A container restarted in the same pod (OOM, crash) keeps its pod, so the
// garbage collector leaves its agents running: release them at boot.
provider.release_owned().await?;
let check = PodRun::new(&kube, "check", &image, "cargo test")
    .namespace(&namespace)
    .owner_pod(&name, &uid);

An owner reference only holds in the namespace of its owner: with an owner the garbage collector cannot find in the namespace of the pod, it deletes the pod right after its creation. The worker must therefore run in the namespace of its agent pods, unlike the two-namespace layout of examples/k8s/sandbox/namespace-rbac.yaml. Before creating anything, each invocation, PodRun::run and JobRun::run read the owner pod there and fail when it is missing or carries another uid. An empty name or uid fails too, before any request.

release_owned deletes the ironflow pods, JobRun Jobs (and their pods) and prompt ConfigMaps with an ownerReference to the owner’s uid, and waits until the pods are gone (.previous_attempt_timeout(d)). Objects of another worker or without owner are left alone. It fails when no owner is set.

PodRun and JobRun keep at most 1 MiB of logs: past that, only the tail is kept, behind a [... N bytes truncated ...] marker, and logs_truncated is set on the output. .max_log_bytes(n) changes the limit. A PodRun with .security(..) sets fsGroupChangePolicy: OnRootMismatch, so a large volume is not re-chowned on every pod; .fs_group_change_policy("Always") restores the Kubernetes default.

A PodRun running a script it does not trust takes .sandboxed(): the same hardening as K8sEphemeralProvider::sandboxed (read-only root, all capabilities dropped, seccomp RuntimeDefault, non-root uid 10001 unless .security(..) sets one, /tmp on a 512Mi emptyDir, no service account token). .stdin(input) feeds the script up to 255 KiB through a downward API file, and the output reports how it ended: exit_code from the container, reason from the pod (DeadlineExceeded when active_deadline_seconds killed it, with no exit_code) or from the container (OOMKilled, Error). logs holds stdout and stderr interleaved; redirect stderr in the command to parse stdout alone.

let out = ctx
    .operation(
        "extract",
        &PodRun::new(&kube, "extract", &image, "python /scripts/extract.py 2>/dev/null")
            .sandboxed()
            .stdin(document_json)
            .active_deadline_seconds(Duration::from_secs(60)),
    )
    .await?;
let timed_out = out.reason.as_deref() == Some("DeadlineExceeded");

A read-only root needs writable volumes: .scratch_volume(name, mount_path, size, storage_class) adds one. Without a storage class it is an emptyDir bounded by size; with one it is an ephemeral claim created and deleted with the pod. Tolerations written as key[=value]:Effect (comma-separated, key alone means Exists) are parsed once by K8sToleration::parse_list and apply to the agent providers and to a PodRun alike, through .k8s_tolerations(..).

let tolerations = K8sToleration::parse_list("dedicated=worker:NoSchedule,gpu:NoExecute")?;
let provider = tolerations
    .iter()
    .cloned()
    .fold(K8sEphemeralProvider::sandboxed(&image), |p, t| p.toleration(t));
let run = PodRun::new(&kube, "build", &image, "make")
    .sandboxed()
    .k8s_tolerations(tolerations)
    .scratch_volume("target", "/target", "10Gi", None)
    .scratch_volume("cache", "/cache", "50Gi", Some("fast-ssd"));

Persistent environment

By default an agent pod starts from an empty working directory and loses everything when it exits. environment_volume gives each agent step a ReadWriteOnce PersistentVolumeClaim that outlives the pod, so a later step can continue in the files an earlier one left (a clone, a build cache, a half-written patch):

use ironflow_core::provider::{EnvironmentVolume, StorageUnit, VolumeSize};

let provider = K8sEphemeralProvider::sandboxed(&image)
    .environment_volume(
        EnvironmentVolume::new("/workspace")
            .size(VolumeSize::new(20, StorageUnit::Gi)) // 10Gi by default
            .storage_class("fast-ssd")           // cluster default otherwise
            .ttl(Duration::from_secs(3 * 86400)), // 7 days by default
    )
    .working_dir("/workspace");

A step without a resume id gets a new claim named ironflow-env-...; its name comes back as the environment id, in StepOutput::environment_id (and AgentResult::environment_id outside the engine). The id is stored with the step, shown by the API and replayed with the step when a run resumes. Pass it to the next step to mount the same claim again:

let clone = ctx
    .agent(
        "clone",
        AgentStepConfig::new("Clone the repository and run the tests").allow_tool(Tool::Bash),
    )
    .await?;
// `None` on a provider without persistent environments.
if let Some(environment) = clone.environment_id.as_deref() {
    ctx.agent(
        "fix",
        AgentStepConfig::new("Fix the failing test")
            .allow_tool(Tool::Bash)
            .resume_environment(environment),
    )
    .await?;
}

A structured step (.output::<T>()) answers with T, which hides the id from ctx.agent. Use ctx.agent_with_meta: its AgentReply holds the typed answer, the environment_id and the account_id. A replayed step returns the same ids.

The claim carries the ironflow labels with app.kubernetes.io/component=environment, the run and step labels of the step that created it, and the ironflow.io/expires-at annotation. Every use pushes the expiry ttl forward (never less than the pod lifetime). A resumed step fails before any pod is created when the claim does not exist, is not an ironflow environment, or is being deleted. A step that sets resume_environment on a provider without environment_volume fails the same way. Other providers return no environment id and ignore the resume id.

The claim is ReadWriteOnce: two steps resuming the same environment at the same time can only run on the same node. Run them one after the other. There is no snapshot: a step that breaks the workspace leaves it broken for the next one.

The worker needs create, get, patch, list and delete on persistentvolumeclaims (see examples/k8s/sandbox/namespace-rbac.yaml).

Sessions volume

An agent step interrupted by a lost lease resumes its Claude Code session on its next execution (see Resuming an interrupted agent step). HOME of a sandboxed pod is an emptyDir, so the session dies with the pod and the step restarts from scratch. sessions_volume mounts an existing PersistentVolumeClaim on ~/.claude/projects, where Claude Code writes its sessions, so the next pod finds the session again:

let provider = K8sEphemeralProvider::sandboxed(&image)
    .sessions_volume("claude-sessions")
    .working_dir("/workspace");

The claim is yours to create; the provider never creates nor deletes it. Use a ReadWriteMany claim, or make sure the next pod lands on the same node. Claude Code keys its sessions by working directory: keep the same working_dir across executions. A Claude profile mapped to projects (or below) and a step volume mounted on ~/.claude/projects are refused, since they would hide the sessions.

The reaper

Every object ironflow creates carries app.kubernetes.io/managed-by=ironflow, an app.kubernetes.io/component (claude-runner, prompt-data, pod-run, job-run) and the ironflow.io/expires-at annotation (unix seconds: creation

• timeout + margin). That covers the agent pods and their prompt ConfigMaps, and the pods of PodRun and Jobs of JobRun from ironflow-ops-k8s (.expiry_margin(d), 60s by default). A caller cannot set managed-by or component: pod_label, PodRun::label and JobRun::label panic, an agent step carrying one fails.

The reaper selects on managed-by=ironflow and deletes pods and Jobs past their expiry or killed with DeadlineExceeded, expired prompt ConfigMaps and expired environment claims (a claim being deleted is left alone). A Job goes with its pods; a pod a Job controls is left to its Job. Objects without a parseable annotation are never touched.

let report = provider.reap_orphans().await?;            // one pass
let handle = provider.spawn_orphan_reaper(Duration::from_secs(300)); // background
// Without a K8sEphemeralProvider, e.g. a worker that only runs PodRun:
let report = reap_orphans(&K8sClusterConfig::Default, "ironflow-agents").await?;

Reaping Jobs needs list and delete on jobs (see examples/k8s/sandbox/namespace-rbac.yaml). Without them, the Job pass is skipped with a warning and the rest of the pass runs. The same holds for list and delete on persistentvolumeclaims and the environment pass.

gVisor (RuntimeClass)

A runc container shares the node kernel: a kernel exploit in untrusted code reaches the node. gVisor runs the pod against a user-space kernel instead. On Talos, install the gvisor system extension on the nodes, then declare the RuntimeClass once per cluster:

apiVersion: node.k8s.io/v1
kind: RuntimeClass
metadata:
  name: gvisor
handler: runsc

Set it on the provider to cover every agent pod, and override it for one step (the step wins):

let provider = K8sEphemeralProvider::sandboxed(&image)
    .runtime_class("gvisor");

let config = AgentConfig::new("Review this untrusted patch")
    .runtime_class("kata"); // wins over the provider's

A check pod created with PodRun takes the same setting:

let run = PodRun::new(&kube, "check", "rust:1.94", "cargo test")
    .runtime_class("gvisor");

Without a value, spec.runtimeClassName stays absent and the cluster default runtime (runc) applies. A blank name is refused.

Warning: builds are noticeably slower under gVisor (compilation, many small file syscalls, cargo and npm installs). Keep gVisor for the steps that handle untrusted code and leave trusted build steps on runc.

Network policies

The worker does not create network policies: its Role in examples/k8s/sandbox/namespace-rbac.yaml has no access to them. A cluster administrator applies a default deny (networkpolicy-deny-all.yaml) and FQDN openings selected on the ironflow.io/egress-profile label:

# FQDN egress for agent pods with Cilium.
#
# Every claude-runner pod may resolve names through kube-dns and reach
# api.anthropic.com on 443. The DNS rule with `matchPattern: "*"` is required:
# Cilium learns the IPs behind toFQDNs names by inspecting DNS answers.
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: claude-runner-egress-anthropic
  namespace: ironflow-agents
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/component: claude-runner
  egress:
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: kube-system
            k8s:k8s-app: kube-dns
      toPorts:
        - ports:
            - port: "53"
              protocol: ANY
          rules:
            dns:
              - matchPattern: "*"
    - toFQDNs:
        - matchName: api.anthropic.com
      toPorts:
        - ports:
            - port: "443"
              protocol: TCP
---
# Per-profile opening: pods labelled ironflow.io/egress-profile=gitlab (set
# with `.egress_profile("gitlab")` on the provider or the step) may also reach
# gitlab.com. Cilium policies are additive: this adds to the rule above.
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: claude-runner-egress-gitlab
  namespace: ironflow-agents
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/component: claude-runner
      ironflow.io/egress-profile: gitlab
  egress:
    - toEndpoints:
        - matchLabels:
            k8s:io.kubernetes.pod.namespace: kube-system
            k8s:k8s-app: kube-dns
      toPorts:
        - ports:
            - port: "53"
              protocol: ANY
          rules:
            dns:
              - matchPattern: "*"
    - toFQDNs:
        - matchName: gitlab.com
      toPorts:
        - ports:
            - port: "443"
              protocol: TCP

Full example

//! Sandboxed K8s ephemeral transport example.
//!
//! Runs Claude Code in a hardened pod: non-root, read-only root filesystem,
//! all capabilities dropped, OAuth token read from a Kubernetes Secret, a
//! locked-down managed-settings preset, and the `anthropic-only` egress
//! profile label. A background task reaps orphaned pods every five minutes.
//!
//! Apply the manifests of `examples/k8s/sandbox/` first. Requires a reachable
//! Kubernetes cluster (via kubeconfig or in-cluster).
//!
//! # Usage
//!
//! ```sh
//! K8S_IMAGE=registry.gitlab.com/thomastartrau/ironflow/ironflow-claude-runner:2.1.316-1 \
//!     cargo run --bin k8s-sandboxed
//! ```

use std::env;
use std::time::Duration;

use ironflow_core::prelude::*;
use ironflow_core::providers::claude::K8sEphemeralProvider;

#[tokio::main]
async fn main() -> Result<(), OperationError> {
    let image = env::var("K8S_IMAGE").expect("K8S_IMAGE env var required");

    let provider = K8sEphemeralProvider::sandboxed(&image)
        .namespace("ironflow-agents")
        .oauth_token_from_secret("claude-oauth", "token")
        .managed_settings_preset("locked", "claude-managed-locked")
        .default_managed_settings("locked")
        .egress_profile("anthropic-only")
        .timeout(Duration::from_secs(600));
    let _reaper = provider.spawn_orphan_reaper(Duration::from_secs(300));

    // Outside the engine, `run_scope` sets the run/step labels the engine
    // stamps on every agent step.
    let config = AgentConfig::new("List the top-level directories of /data/repos.")
        .max_budget_usd(0.10)
        .read_only_pvc("repos", "/data/repos")
        .run_scope("demo-run", "investigate");

    let result = Agent::from_config(config).run(&provider).await?;

    println!("Response: {}", result.text());
    println!("Model: {}", result.model().unwrap_or("unknown"));
    println!("Duration: {}ms", result.duration_ms());

    Ok(())
}

Checks

Run these against a live agent pod before trusting the setup:

# Non-root, read-only root filesystem, capabilities dropped.
kubectl -n ironflow-agents get pod -l app.kubernetes.io/component=claude-runner \
  -o jsonpath='{.items[0].spec.containers[0].securityContext}'
# No secret value in the spec, only secretKeyRef.
kubectl -n ironflow-agents get pod <pod> -o yaml | grep -A3 CLAUDE_CODE_OAUTH_TOKEN
# No service account token mounted.
kubectl -n ironflow-agents exec <pod> -- ls /var/run/secrets/kubernetes.io/serviceaccount
# Root filesystem is read-only.
kubectl -n ironflow-agents exec <pod> -- touch /usr/local/probe
# Egress is limited to the profile.
kubectl -n ironflow-agents exec <pod> -- curl -sS -m 5 https://example.com
# The worker cannot touch network policies.
kubectl auth can-i create networkpolicies -n ironflow-agents \
  --as=system:serviceaccount:ironflow:ironflow-worker
# Run, root run and step labels, and the expiry annotation.
kubectl -n ironflow-agents get pods \
  -L ironflow.io/run-id,ironflow.io/root-run-id,ironflow.io/step \
  -o custom-columns='NAME:.metadata.name,EXPIRES:.metadata.annotations.ironflow\.io/expires-at'

The exec checks against the service account, the root filesystem and egress must fail; kubectl auth can-i must answer no.

With the auth proxy, also check that the pod holds no Claude secret and that the proxy refuses an unknown token:

kubectl -n ironflow-agents exec <pod> -- env | grep -c 'sk-ant'   # 0
kubectl -n ironflow-agents exec <pod> -- sh -c \
  'curl -s -o /dev/null -w "%{http_code}" -H "Authorization: Bearer invalide" "$ANTHROPIC_BASE_URL/v1/messages"'   # 401

With proxied secrets, the pod holds the opaque token only, and a host outside the allowlist is refused:

kubectl -n ironflow-agents exec <pod> -- sh -c 'echo "$GITHUB_TOKEN" | cut -c1-5'   # ifap_
kubectl -n ironflow-agents exec <pod> -- sh -c \
  'curl -s -o /dev/null -w "%{http_code}" -H "Authorization: Bearer $GITHUB_TOKEN" "$GITHUB_TOKEN_URL/example.com/"'   # 403

The relay and the metric are covered by the proxy test suites:

cargo test -p ironflow-auth-proxy --test relay_secrets
cargo test -p ironflow-auth-proxy --test metrics

Testing Workflows

ironflow_engine::testing::TestEngine runs a WorkflowHandler against an in-memory store with mocked steps. The run, the steps, the FSM transitions and the persistence are the production ones – only the outside world is swapped out.

What it replaces:

A first test

use ironflow_engine::prelude::*;
use ironflow_engine::testing::{MockShellOutput, TestEngine};
use ironflow_store::models::{RunStatus, StepStatus};
use serde_json::json;

use crate::handlers::Deploy;

#[tokio::test]
async fn deploy_runs_build_then_ship() {
    let result = TestEngine::new()
        .with_handler(Deploy)
        .with_mock_shell(|cfg| match cfg.command.as_str() {
            "cargo build" => Ok(MockShellOutput::ok("compiled")),
            _ => Ok(MockShellOutput::ok("shipped")),
        })
        .run(json!({"environment": "staging"}))
        .await
        .expect("the harness ran the handler");

    assert_eq!(result.status(), RunStatus::Completed);
    assert_eq!(result.step_names(), vec!["build", "ship"]);
    assert_eq!(result.step("build").step_output().stdout(), "compiled");
    assert_eq!(result.step("ship").status(), StepStatus::Completed);
}

A handler that fails is not an Err: the returned TestResult carries RunStatus::Failed and the message in error(). Only wiring failures – no handler registered, two handlers sharing a name, a store rejection – come back as Err.

Building the harness

Every with_* method panics if called after the first run: the engine is built once, so a later change would be silently ignored.

Then run:

Asserting on the result

TestResult reads the run and its steps back from the store, so an assertion sees exactly what the API and the dashboard would serve.

Each TestStep exposes name(), kind(), status(), step_output(), output(), input(), error(), duration(), cost_usd(), is_completed(), is_error_handler() and raw(). step_output() reads the persisted output through the typed StepOutput accessors (stdout(), status(), body(), text(), json::<T>()); output() is the raw JSON.

Steps of a parallel wave share a position and a handler may reuse a name: disambiguate those with steps() rather than step(name).

Shell and HTTP parity

The mocks reproduce the asymmetry of the real executors, so allow_failure, step retries and run failure behave exactly as in production:

• A MockShellOutput with a non-zero exit_code is an error, like a real non-zero exit. Use MockShellOutput::failed(1, "boom"). When the step sets exit_code_as_output(), the mock completes with that exit code as its output.
• A MockHttpResponse with a non-2xx status is a normal output, like a real 500 response. Return Err(OperationError::Http { status: None, .. }) from the closure to simulate a transport failure instead.

use ironflow_core::error::OperationError;
use ironflow_engine::testing::{MockHttpResponse, TestEngine};
use serde_json::json;

let harness = TestEngine::new()
    .with_handler(Fetch)
    // A 404 the handler is expected to deal with.
    .with_mock_http(|cfg| {
        if cfg.url.ends_with("/missing") {
            Ok(MockHttpResponse::json(404, &json!({"error": "not found"})))
        } else {
            Ok(MockHttpResponse::ok(&json!({"id": 7})))
        }
    });

Approval gates

Two ways to test a gated handler:

use ironflow_engine::testing::{ApprovalOutcome, TestEngine};

// 1. Resolve the gate inline and assert on the whole run.
let approved = TestEngine::new()
    .with_handler(GatedDeploy)
    .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("ok")))
    .with_mock_approval(ApprovalOutcome::Approved)
    .run(json!({}))
    .await?;
assert_eq!(approved.status(), RunStatus::Completed);

// A rejection fails the run with EngineError::ApprovalRejected.
let rejected = TestEngine::new()
    .with_handler(GatedDeploy)
    .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("ok")))
    .with_mock_approval(ApprovalOutcome::reject("budget freeze"))
    .run(json!({}))
    .await?;
assert_eq!(rejected.step("gate").status(), StepStatus::Rejected);

// 2. Without a mock, the gate suspends the run, the way production does.
let mut harness = TestEngine::new()
    .with_handler(GatedDeploy)
    .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("ok")));

let suspended = harness.run(json!({})).await?;
assert_eq!(suspended.status(), RunStatus::AwaitingApproval);

let resumed = harness.resume(suspended.run_id()).await?;
assert_eq!(resumed.status(), RunStatus::Completed);

Agent fixtures

with_recorded_agent(dir) replays fixtures written by RecordReplayProvider. The argument is the directory: each fixture is keyed by a hash of the AgentConfig and stored as <hash>.json inside it (the random nonce of a Prompt untrusted block is masked first). A missing fixture fails the step instead of falling back to the real Claude CLI, so a stale suite never silently starts spending tokens.

let result = TestEngine::new()
    .with_handler(Review)
    .with_recorded_agent("tests/fixtures")
    .run(json!({}))
    .await?;

To record, pass a recording provider through the escape hatch:

use std::sync::Arc;

use ironflow_core::provider::AgentProvider;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_core::providers::record_replay::RecordReplayProvider;

let provider: Arc<dyn AgentProvider> = Arc::new(RecordReplayProvider::record(
    ClaudeCodeProvider::new(),
    "tests/fixtures",
));
let result = TestEngine::new()
    .with_handler(Review)
    .with_agent_provider(provider)
    .run(json!({}))
    .await?;

With no agent backend configured at all, an agent step fails with a message naming the three constructors – a forgotten mock is a loud failure, not a network call.

Parallel waves, error handlers and sub-workflows

The mocks apply to the steps inside them: a step of a ctx.parallel(...) wave, a step fired by ctx.on_error(...), and every step of a child run started with ctx.workflow(...) all go through the same interceptor. Register both handlers and drive the parent:

let mut harness = TestEngine::new()
    .with_handler(Parent)
    .with_handler(Child)
    .with_mock_shell(|_cfg| Ok(MockShellOutput::ok("ok")));
let store = harness.store();

let result = harness.run_workflow("parent", json!({})).await?;
// A workflow step stores a `SubWorkflowOutput`: read it back typed.
let child: SubWorkflowOutput = result.step("child").step_output().json()?;
let child_steps = store.list_steps(child.run_id()).await?;

Limitations

• Custom operations (ctx.operation(...)) are not intercepted. Mock one by passing a test-double Operation to the handler.
• ctx.delay(...) is not intercepted: a non-zero delay still suspends the run with RunStatus::Sleeping. It resumes once RunWaker::tick runs after its scheduled_at.
• ctx.wait_for_signal(...) is resolved with with_mock_signal(|step, name, key| ..), returning SignalOutcome::Received(json!(..)) or SignalOutcome::TimedOut. Without it, the run ends in RunStatus::Sleeping until a signal is delivered.
• ctx.send_signal(...) is not intercepted: it really delivers into the InMemoryStore. Check what was sent with store.list_signals(..), or the delivery the step returned.
• ctx.decision(...) needs a real DecisionProvider, wired with with_decision_provider.

Use the real Engine when the test must exercise real commands, real requests or a real agent; use TestEngine when it must exercise the handler’s logic.

Library Mode

ironflow-core works on its own, without the API server, a worker or a database. Add it to any binary and call the operations directly: shell commands, HTTP requests and AI agents, with timeouts, budgets and cost tracking. Nothing is persisted and there is no dashboard; for that, use the full platform described in the rest of this book.

cargo add ironflow-core tokio --features tokio/full

use ironflow_core::prelude::*;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = ClaudeCodeProvider::new();

    // Run a shell command
    let files = Shell::new("ls -la src/").await?;

    // Feed the output into an agent
    let review = Agent::new()
        .prompt(&format!("Review these source files:\n{}", files.stdout()))
        .model(Model::SONNET)
        .max_budget_usd(0.10)
        .run(&provider)
        .await?;

    println!("{}", review.text());
    Ok(())
}

Shell

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;
use std::time::Duration;

async fn example() -> Result<(), OperationError> {
let output = Shell::new("cargo test")
    .dir("/path/to/project")
    .timeout(Duration::from_secs(120))
    .env("RUST_LOG", "debug")
    .await?;

println!("stdout: {}", output.stdout());
println!("exit code: {}", output.exit_code());
Ok(())
}
}

Shell::new hands its string to sh -c. When the command carries data you do not control, use Shell::exec(program, &[args]), which passes each argument as is.

Http

A non-2xx status is not an error: check is_success().

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;
use std::time::Duration;

async fn example() -> Result<(), OperationError> {
let output = Http::post("https://httpbin.org/post")
    .header("Authorization", "Bearer token123")
    .json(serde_json::json!({"key": "value"}))
    .timeout(Duration::from_secs(30))
    .await?;

println!("status: {}, body: {}", output.status(), output.body());
Ok(())
}
}

Agent

Derive JsonSchema on a type and the provider is constrained to return it.

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

#[derive(Deserialize, JsonSchema)]
struct Review {
    score: u8,
    summary: String,
}

async fn example() -> Result<(), OperationError> {
let provider = ClaudeCodeProvider::new();

let result = Agent::new()
    .system_prompt("You are a senior Rust reviewer.")
    .prompt("Review the codebase")
    .model(Model::OPUS)
    .allowed_tools(&["Read", "Grep"])
    .max_turns(5)
    .max_budget_usd(0.50)
    .output::<Review>()
    .run(&provider)
    .await?;

let review: Review = result.json().expect("schema-validated output");
println!("Score: {}/10 - {}", review.score, review.summary);
println!("Cost: ${:.4}", result.cost_usd().unwrap_or(0.0));
Ok(())
}
}

Agent::timeout(Duration::from_secs(300)) bounds the call independently of the provider, automatic retries included: past it run fails with AgentError::Timeout and the provider stops its process or pod. Without it, the provider timeout applies.

Other providers (OpenAI, Gemini, Mistral, remote Claude Code) are listed in Agent Providers.

Session resume

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
let provider = ClaudeCodeProvider::new();

let first = Agent::new()
    .prompt("Analyze the src/ directory")
    .max_budget_usd(0.10)
    .run(&provider)
    .await?;

let session = first.session_id().expect("provider returned session ID");

let followup = Agent::new()
    .prompt("Now suggest improvements")
    .resume(session)
    .max_budget_usd(0.10)
    .run(&provider)
    .await?;
Ok(())
}
}

Parallel execution

tokio::try_join! when the number of operations is known at compile time:

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
let (files, status) = tokio::try_join!(
    Shell::new("ls -la"),
    Shell::new("git status"),
)?;
Ok(())
}
}

try_join_all when it is decided at run time, try_join_all_limited to cap concurrency:

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
let provider = ClaudeCodeProvider::new();
let prompts = vec!["Summarize file A", "Summarize file B", "Summarize file C"];

let results = try_join_all_limited(
    prompts.iter().map(|p| {
        Agent::new()
            .prompt(p)
            .model(Model::HAIKU)
            .max_budget_usd(0.10)
            .run(&provider)
    }),
    2, // at most 2 agent calls at a time
).await?;
Ok(())
}
}

Tracking cost and duration

WorkflowTracker adds up cost, tokens and duration across operations:

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
let provider = ClaudeCodeProvider::new();
let mut tracker = WorkflowTracker::new("deploy-pipeline");

let files = Shell::new("ls -la").await?;
tracker.record_shell("list-files", &files);

let review = Agent::new()
    .prompt("Review the project")
    .max_budget_usd(0.10)
    .run(&provider)
    .await?;
tracker.record_agent("code-review", &review);

tracker.summary(); // structured log via tracing
println!("Total cost: ${:.4}", tracker.total_cost_usd());
println!("Steps: {}", tracker.step_count());
Ok(())
}
}

Dry-run mode

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
// Global: every operation skips execution
set_dry_run(true);
let output = Shell::new("rm -rf /").await?; // not executed
assert_eq!(output.stdout(), "");

// Per operation, overrides the global setting
set_dry_run(false);
let output = Shell::new("echo hello").dry_run(true).await?;
assert_eq!(output.stdout(), "");
Ok(())
}
}

Record/replay testing

RecordReplayProvider wraps any provider and stores its responses as JSON fixtures, keyed by a hash of the prompt, system prompt and schema. Tests then replay them without spending tokens. The random nonce of a Prompt untrusted block is masked before hashing, so it does not change the fixture.

#![allow(unused)]
fn main() {
use ironflow_core::prelude::*;

async fn example() -> Result<(), OperationError> {
// Record mode when IRONFLOW_RECORD=1, replay otherwise
let provider = RecordReplayProvider::new(ClaudeCodeProvider::new(), "tests/fixtures");

// Or force replay, ignoring the env var
let provider = RecordReplayProvider::replay(ClaudeCodeProvider::new(), "tests/fixtures");

let result = Agent::new()
    .prompt("Explain ownership in Rust")
    .max_budget_usd(0.10)
    .run(&provider)
    .await?;
Ok(())
}
}

To test a full workflow handler, see Testing Workflows.

Standalone Runtime

ironflow-runtime is the path without a database: a small axum server that exposes webhook endpoints and calls ironflow-core operations directly. There is no store, no worker and no dashboard, so nothing is persisted. Use it for a webhook that triggers a quick job; use the full platform when you need run history, approvals or retries.

use ironflow_core::prelude::*;
use ironflow_runtime::prelude::*;

async fn on_push(payload: serde_json::Value, provider: &ClaudeCodeProvider) {
    let branch = payload["ref"].as_str().unwrap_or("main");
    // The branch comes from the webhook: an argument, never a `sh -c` string.
    let range = format!("origin/main...origin/{branch}");
    let diff = Shell::exec("git", &["diff", &range])
        .await
        .expect("git diff");
    let review = Agent::new()
        .prompt(&format!("Review this diff:\n{}", diff.stdout()))
        .model(Model::SONNET)
        .max_budget_usd(0.50)
        .run(provider)
        .await
        .expect("agent review");
    println!("{}", review.text());
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = ClaudeCodeProvider::new();

    Runtime::new()
        .webhook("/hooks/github", WebhookAuth::github("my-secret"), {
            let p = provider.clone();
            move |payload| {
                let p = p.clone();
                async move { on_push(payload, &p).await }
            }
        })
        .serve("0.0.0.0:8080")
        .await?;

    Ok(())
}

Built-in endpoints: GET /health, and GET /metrics with the prometheus feature.

Webhook replays

GitHub and GitLab send a delivery again when the receiver times out or answers 5xx. Use webhook_with_context to get the provider delivery id, already prefixed and ready to pass as an Idempotency-Key, so a replay does not start a second run.

#![allow(unused)]
fn main() {
use ironflow_runtime::prelude::*;

Runtime::new().webhook_with_context(
    "/hooks/github",
    WebhookAuth::github("my-secret"),
    |ctx: WebhookContext| async move {
        // ctx.delivery_id == Some("github:8f4e2a10-...") when the provider
        // stamped the request, None otherwise. Pass it straight to
        // create_run_idempotent as the Idempotency-Key.
        match ctx.delivery_id {
            Some(key) => println!("replay-safe key: {key}"),
            None => println!("no delivery id, the call is not replay-safe"),
        }
    },
);
}

webhook() keeps its original signature and receives only the payload.

Metrics

With the prometheus feature, operations record these metrics, in the runtime as in a worker:

Interfaces

Workflows are written in Rust, but runs can be started and followed from five places: the web dashboard, the CLI, the Rust SDK, the MCP server for AI assistants, and the REST API they all sit on. The Claude Code plugin helps you write the workflows themselves.

Dashboard

A React web UI served by the API server. It covers the workflow catalog (with a form generated from each handler’s input_schema), run history with filters, live steps and logs, approval and rejection, secrets, API keys, users and audit logs.

Two ways to serve it:

• Embedded: build ironflow-api with the dashboard feature and the compiled assets are baked into the binary via rust-embed.
• From disk: set DASHBOARD_DIR to a build output directory, which overrides the embedded copy.

CLI

cargo install ironflow-cli

$ ironflow-cli workflow list
┌───────────────────┬──────────┬─────────┐
│ Name              ┆ Category ┆ Version │
╞═══════════════════╪══════════╪═════════╡
│ ci-pipeline       ┆ -        ┆ -       │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┤
│ deploy-approval   ┆ -        ┆ -       │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┤
│ greeting          ┆ examples ┆ -       │
└───────────────────┴──────────┴─────────┘

$ ironflow-cli run create ci-pipeline
┌──────────┬─────────────┬─────────┬──────────┬─────────┬─────────────────────┬─────────┐
│ ID       ┆ Workflow    ┆ Status  ┆ Duration ┆ Cost    ┆ Created             ┆ Started │
╞══════════╪═════════════╪═════════╪══════════╪═════════╪═════════════════════╪═════════╡
│ 019f9f50 ┆ ci-pipeline ┆ pending ┆ 0ms      ┆ $0.0000 ┆ 2026-07-26 16:44:19 ┆ -       │
└──────────┴─────────────┴─────────┴──────────┴─────────┴─────────────────────┴─────────┘

$ ironflow-cli logs 019f9f50-17c8-73b1-9288-b41cbed28d1a
$ ironflow-cli run list --status completed --workflow ci-pipeline
$ ironflow-cli run get <run-id> --verbose
$ ironflow-cli stats
$ ironflow-cli stats roles --period 30d
$ ironflow-cli run steps --role reviewer

Configuration is resolved in this order: command-line flags (--url, --api-key), then the IRONFLOW_URL and IRONFLOW_API_KEY environment variables, then ~/.ironflow.toml:

base_url = "http://localhost:3000"
api_key = "irfl_..."

The CLI, the SDK and the MCP server only reach the routes the key’s scopes allow: see API key scopes.

Add --json to any command for machine-readable output. ironflow-cli --help lists every command.

Rust SDK

Types are generated from openapi.json at build time, so the client cannot drift from the API.

#![allow(unused)]
fn main() {
use ironflow_sdk::IronflowClient;
use ironflow_sdk::types::CreateRunRequest;

async fn example() -> Result<(), Box<dyn std::error::Error>> {
let client = IronflowClient::new("http://localhost:3000", "irfl_...");

let mut payload = serde_json::Map::new();
payload.insert("branch".to_string(), serde_json::json!("main"));

let created = client
    .create_run(&CreateRunRequest {
        workflow: "ci-pipeline".to_string(),
        payload: Some(payload),
        labels: None,
        scheduled_at: None,
        max_retries: Some(2),
        max_cost_usd: Some(1.0),
        concurrency_key: None,
        priority: None,
        concurrency_limits: Vec::new(),
        worker_tags: Vec::new(),
    })
    .await?;

let detail = client.get_run(created.data.id).await?;
println!("status: {:?}", detail.data.run.status);
Ok(())
}
}

MCP server

Lets an AI assistant list workflows, trigger runs, inspect results, and approve or reject pending gates.

cargo install ironflow-mcp
claude mcp add ironflow --env IRONFLOW_URL=http://localhost:3000 --env IRONFLOW_API_KEY=irfl_... -- ironflow-mcp

Or declare it in .mcp.json:

{
  "mcpServers": {
    "ironflow": {
      "command": "ironflow-mcp",
      "env": {
        "IRONFLOW_URL": "http://localhost:3000",
        "IRONFLOW_API_KEY": "irfl_..."
      }
    }
  }
}

Exposed tools include list_workflows, get_workflow, list_runs, get_run, create_run, approve_run, reject_run, cancel_run, retry_run, get_stats, get_agent_role_stats (cost and executions per agent role) and list_agent_role_steps (the steps of one role).

Claude Code plugin

Skills that teach Claude Code how to build on Ironflow: scaffold a project, write a handler, write a custom operation, test it end to end, and review a handler for the mistakes the compiler cannot catch (side effects around approval gates, unstable step names, leaked secrets).

# 1. Register the Ironflow repository as a plugin marketplace
claude plugin marketplace add https://gitlab.com/ThomasTartrau/ironflow.git

# 2. Install the plugin
claude plugin install ironflow@ironflow

/ironflow setup              # workspace: workflows lib, server, worker, hello workflow, e2e test
/ironflow workflow deploy    # a WorkflowHandler with a typed input schema, registered
/ironflow operation slack    # a custom Operation tracked as a step
/ironflow test deploy        # Engine + InMemoryStore + record/replay test
/ironflow review             # the workflow reviewer agent

Every Rust snippet in the plugin is compiled in CI, and the project template is scaffolded and built against each release. Details in the plugin README.

REST API

Everything above goes through the REST API under /api/v1. With the openapi feature, the server publishes its specification at /api/v1/openapi.json.

Feature Flags

Optional parts of Ironflow are behind Cargo features, so a binary only compiles what it uses.

Architecture Overview

Ironflow follows a client-server architecture with background workers for execution.

Components

graph TD
    Dashboard[Web Dashboard] --> API[API Server]
    CLI[CLI] --> SDK[Rust SDK]
    MCP[MCP Server] --> SDK
    Hooks[Webhooks and cron] --> API
    SDK --> API
    API --> Store[(Database)]
    API --> Artifacts[(Blob Store)]
    Worker1[Worker 1] --> API
    Worker2[Worker 2] --> API
    Worker1 --> Provider[Agent Provider]
    Worker2 --> Provider

The API owns persistence and never executes a workflow itself (except when it resumes a run in local execution mode). Workers poll the API for pending runs, execute the workflow handler locally, and stream steps and logs back. Scaling out means starting more workers. The scaffolded server and the example server run in Workers mode.

ironflow-runtime is a separate, lighter path: a standalone daemon with webhook endpoints and trigger sources that calls ironflow-core operations directly, without a store or an API. See Standalone Runtime.

Crate map

Request flow

1. A client (dashboard, CLI, SDK, or webhook) sends a request to the API
2. The API validates authentication, creates a Run in the Store, and returns it
3. A Worker polls the API, acquires a lease on the Run, and executes the handler
4. The handler calls steps (ctx.shell(), ctx.agent(), etc.), each persisted as they complete
5. Events are published via SSE for real-time updates; a client that falls behind gets a lagged event ({"missed": n}) and reloads its state
6. On completion or failure, the Worker reports the result back to the API

Data flow

Runs and steps are stored in PostgreSQL (or in-memory for development). Artifacts (files produced by steps) are stored in a separate blob store (local filesystem by default). The two are linked by artifact metadata on each step.